feat: Added distinct_id to kafka headers

ee/clickhouse/models/test/__snapshots__/test_cohort.ambr

@@ -233,7 +233,7 @@
                      WHERE equals(person_distinct_id_overrides.team_id, 99999)
                      GROUP BY person_distinct_id_overrides.distinct_id
                      HAVING ifNull(equals(argMax(person_distinct_id_overrides.is_deleted, person_distinct_id_overrides.version), 0), 0) SETTINGS optimize_aggregation_in_order=1) AS e__override ON equals(e.distinct_id, e__override.distinct_id)
-                  WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2023-01-21 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-21 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
+                  WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2023-01-23 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-23 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
                GROUP BY actor_id) AS source
             ORDER BY source.id ASC
             LIMIT 100 SETTINGS optimize_aggregation_in_order=1,
@@ -374,7 +374,7 @@
                      actor_id AS id
               FROM
                 (SELECT min(toTimeZone(e.timestamp, 'UTC')) AS min_timestamp,
-                        minIf(toTimeZone(e.timestamp, 'UTC'), greaterOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-06 00:00:00.000000', 6, 'UTC'))) AS min_timestamp_with_condition,
+                        minIf(toTimeZone(e.timestamp, 'UTC'), greaterOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-08 00:00:00.000000', 6, 'UTC'))) AS min_timestamp_with_condition,
                         if(not(empty(e__override.distinct_id)), e__override.person_id, e.person_id) AS actor_id,
                         argMin(e.uuid, toTimeZone(e.timestamp, 'UTC')) AS uuid,
                         argMin(e.distinct_id, toTimeZone(e.timestamp, 'UTC')) AS distinct_id
@@ -386,7 +386,7 @@
                     WHERE equals(person_distinct_id_overrides.team_id, 99999)
                     GROUP BY person_distinct_id_overrides.distinct_id
                     HAVING ifNull(equals(argMax(person_distinct_id_overrides.is_deleted, person_distinct_id_overrides.version), 0), 0) SETTINGS optimize_aggregation_in_order=1) AS e__override ON equals(e.distinct_id, e__override.distinct_id)
-                 WHERE and(equals(e.team_id, 99999), lessOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-21 23:59:59.999999', 6, 'UTC')), equals(e.event, 'signup'))
+                 WHERE and(equals(e.team_id, 99999), lessOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-23 23:59:59.999999', 6, 'UTC')), equals(e.event, 'signup'))
                  GROUP BY if(not(empty(e__override.distinct_id)), e__override.person_id, e.person_id)
                  HAVING ifNull(equals(min_timestamp, min_timestamp_with_condition), isNull(min_timestamp)
                                and isNull(min_timestamp_with_condition)))
@@ -474,7 +474,7 @@
                      WHERE equals(person_distinct_id_overrides.team_id, 99999)
                      GROUP BY person_distinct_id_overrides.distinct_id
                      HAVING ifNull(equals(argMax(person_distinct_id_overrides.is_deleted, person_distinct_id_overrides.version), 0), 0) SETTINGS optimize_aggregation_in_order=1) AS e__override ON equals(e.distinct_id, e__override.distinct_id)
-                  WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2023-01-21 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-21 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
+                  WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2023-01-23 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-23 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
                GROUP BY actor_id) AS source
             ORDER BY source.id ASC
             LIMIT 100 SETTINGS optimize_aggregation_in_order=1,
@@ -488,7 +488,7 @@
                                actor_id AS id
                         FROM
                           (SELECT min(toTimeZone(e.timestamp, 'UTC')) AS min_timestamp,
-                                  minIf(toTimeZone(e.timestamp, 'UTC'), greaterOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-06 00:00:00.000000', 6, 'UTC'))) AS min_timestamp_with_condition,
+                                  minIf(toTimeZone(e.timestamp, 'UTC'), greaterOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-08 00:00:00.000000', 6, 'UTC'))) AS min_timestamp_with_condition,
                                   if(not(empty(e__override.distinct_id)), e__override.person_id, e.person_id) AS actor_id,
                                   argMin(e.uuid, toTimeZone(e.timestamp, 'UTC')) AS uuid,
                                   argMin(e.distinct_id, toTimeZone(e.timestamp, 'UTC')) AS distinct_id
@@ -500,7 +500,7 @@
                               WHERE equals(person_distinct_id_overrides.team_id, 99999)
                               GROUP BY person_distinct_id_overrides.distinct_id
                               HAVING ifNull(equals(argMax(person_distinct_id_overrides.is_deleted, person_distinct_id_overrides.version), 0), 0) SETTINGS optimize_aggregation_in_order=1) AS e__override ON equals(e.distinct_id, e__override.distinct_id)
-                           WHERE and(equals(e.team_id, 99999), lessOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-21 23:59:59.999999', 6, 'UTC')), equals(e.event, 'signup'))
+                           WHERE and(equals(e.team_id, 99999), lessOrEquals(toTimeZone(e.timestamp, 'UTC'), toDateTime64('2025-01-23 23:59:59.999999', 6, 'UTC')), equals(e.event, 'signup'))
                            GROUP BY if(not(empty(e__override.distinct_id)), e__override.person_id, e.person_id)
                            HAVING ifNull(equals(min_timestamp, min_timestamp_with_condition), isNull(min_timestamp)
                                          and isNull(min_timestamp_with_condition)))

posthog/api/capture.py

@@ -858,6 +858,8 @@ def capture_internal(
     if extra_headers is None:
         extra_headers = []
 
+    headers = [("token", token), ("distinct_id", distinct_id), *extra_headers]
+
     parsed_event = build_kafka_event_data(
         distinct_id=distinct_id,
         ip=ip,
@@ -871,7 +873,6 @@ def capture_internal(
 
     if event["event"] in SESSION_RECORDING_EVENT_NAMES:
         session_id = event["properties"]["$session_id"]
-        headers = [("token", token), *extra_headers]
 
         overflowing = False
         if token in settings.REPLAY_OVERFLOW_FORCED_TOKENS:
@@ -900,7 +901,9 @@ def capture_internal(
     else:
         kafka_partition_key = candidate_partition_key
 
-    return log_event(parsed_event, event["event"], partition_key=kafka_partition_key, historical=historical)
+    return log_event(
+        parsed_event, event["event"], partition_key=kafka_partition_key, historical=historical, headers=headers
+    )
 
 
 def is_randomly_partitioned(candidate_partition_key: str) -> bool:

posthog/api/test/__snapshots__/test_cohort.ambr

@@ -174,7 +174,7 @@
                                     WHERE equals(person_distinct_id_overrides.team_id, 99999)
                                     GROUP BY person_distinct_id_overrides.distinct_id
                                     HAVING ifNull(equals(argMax(person_distinct_id_overrides.is_deleted, person_distinct_id_overrides.version), 0), 0) SETTINGS optimize_aggregation_in_order=1) AS e__override ON equals(e.distinct_id, e__override.distinct_id)
-                                 WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2025-01-20 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-21 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
+                                 WHERE and(equals(e.team_id, 99999), greaterOrEquals(timestamp, toDateTime64('2025-01-22 00:00:00.000000', 6, 'UTC')), lessOrEquals(timestamp, toDateTime64('2025-01-23 23:59:59.999999', 6, 'UTC')), equals(e.event, '$pageview')))
                               GROUP BY actor_id) AS source
                            ORDER BY source.id ASC
                            LIMIT 100 SETTINGS optimize_aggregation_in_order=1,

posthog/api/test/test_capture.py

@@ -371,7 +371,10 @@ def _do_test_capture_with_likely_anonymous_ids(self, kafka_produce, expect_rando
                 topic=KAFKA_EVENTS_PLUGIN_INGESTION_TOPIC,
                 data=ANY,
                 key=None if expect_random_partitioning else ANY,
-                headers=None,
+                headers=[
+                    ("token", self.team.api_token),
+                    ("distinct_id", distinct_id),
+                ],
             )
 
             if not expect_random_partitioning:
@@ -1916,10 +1919,11 @@ def test_recording_ingestion_can_write_headers_with_the_message(self, kafka_prod
         with self.settings(
             SESSION_RECORDING_KAFKA_MAX_REQUEST_SIZE_BYTES=20480,
         ):
-            self._send_august_2023_version_session_recording_event()
+            self._send_august_2023_version_session_recording_event(distinct_id="distinct_id123")
 
             assert kafka_produce.mock_calls[0].kwargs["headers"] == [
                 ("token", "token123"),
+                ("distinct_id", "distinct_id123"),
                 (
                     # without setting a version in the URL the default is unknown
                     "lib_version",
@@ -1932,10 +1936,13 @@ def test_recording_ingestion_can_read_version_from_request(self, kafka_produce:
         with self.settings(
             SESSION_RECORDING_KAFKA_MAX_REQUEST_SIZE_BYTES=20480,
         ):
-            self._send_august_2023_version_session_recording_event(query_params="ver=1.123.4")
+            self._send_august_2023_version_session_recording_event(
+                query_params="ver=1.123.4", distinct_id="distinct_id123"
+            )
 
             assert kafka_produce.mock_calls[0].kwargs["headers"] == [
                 ("token", "token123"),
+                ("distinct_id", "distinct_id123"),
                 (
                     # without setting a version in the URL the default is unknown
                     "lib_version",

rust/capture/src/sinks/kafka.rs

@@ -209,6 +209,7 @@ impl KafkaSink {
         let data_type = metadata.data_type;
         let event_key = event.key();
         let session_id = metadata.session_id.clone();
+        let distinct_id = event.distinct_id.clone();
 
         drop(event); // Events can be EXTREMELY memory hungry
 
@@ -255,10 +256,17 @@ impl KafkaSink {
             partition: None,
             key: partition_key,
             timestamp: None,
-            headers: Some(OwnedHeaders::new().insert(Header {
-                key: "token",
-                value: Some(&token),
-            })),
+            headers: Some(
+                OwnedHeaders::new()
+                    .insert(Header {
+                        key: "token",
+                        value: Some(&token),
+                    })
+                    .insert(Header {
+                        key: "distinct_id",
+                        value: Some(&distinct_id),
+                    }),
+            ),
         }) {
             Ok(ack) => Ok(ack),
             Err((e, _)) => match e.rdkafka_error_code() {
@@ -413,9 +421,10 @@ mod tests {
         // We test different cases in a single test to amortize the startup cost of the producer.
 
         let (cluster, sink) = start_on_mocked_sink(Some(3000000)).await;
+        let distinct_id = "test_distinct_id_123".to_string();
         let event: CapturedEvent = CapturedEvent {
             uuid: uuid_v7(),
-            distinct_id: "id1".to_string(),
+            distinct_id: distinct_id.clone(),
             ip: "".to_string(),
             data: "".to_string(),
             now: "".to_string(),