ref(kafka): Bump arroyo, add client.id to every producer (#97382)

untitaker · getsantry[bot] · web-flow · commit 64a18485a512 · 2025-08-13T14:53:12.000Z
ref STREAM-324 see getsentry/arroyo#479 and getsentry/relay#5031 --------- Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>
diff --git a/bin/mock-replay-recording b/bin/mock-replay-recording
@@ -40,6 +40,7 @@ logger = logging.getLogger(__name__)
 def get_producer() -> KafkaProducer:
     cluster_name = get_topic_definition(Topic.INGEST_REPLAYS_RECORDINGS)["cluster"]
     producer_config = get_kafka_producer_cluster_options(cluster_name)
+    producer_config["client.id"] = "sentry.bin.mock-replay-recording"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/pyproject.toml b/pyproject.toml
@@ -76,7 +76,7 @@ dependencies = [
     "rfc3339-validator>=0.1.2",
     "rfc3986-validator>=0.1.1",
     # [end] jsonschema format validators
-    "sentry-arroyo>=2.28.3",
+    "sentry-arroyo>=2.29.0",
     "sentry-forked-email-reply-parser>=0.5.12.post1",
     "sentry-kafka-schemas>=1.3.18",
     "sentry-ophio>=1.1.3",
diff --git a/requirements-base.txt b/requirements-base.txt
@@ -63,7 +63,7 @@ requests>=2.32.3
 rfc3339-validator>=0.1.2
 rfc3986-validator>=0.1.1
 # [end] jsonschema format validators
-sentry-arroyo>=2.28.3
+sentry-arroyo>=2.29.0
 sentry-kafka-schemas>=1.3.18
 sentry-ophio>=1.1.3
 sentry-protos==0.3.1
diff --git a/requirements-dev-frozen.txt b/requirements-dev-frozen.txt
@@ -179,7 +179,7 @@ rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
 selenium==4.16.0
-sentry-arroyo==2.28.3
+sentry-arroyo==2.29.0
 sentry-cli==2.16.0
 sentry-covdefaults-disable-branch-coverage==1.0.2
 sentry-devenv==1.22.0
diff --git a/requirements-frozen.txt b/requirements-frozen.txt
@@ -121,7 +121,7 @@ rfc3986-validator==0.1.1
 rpds-py==0.20.0
 rsa==4.8
 s3transfer==0.10.0
-sentry-arroyo==2.28.3
+sentry-arroyo==2.29.0
 sentry-forked-email-reply-parser==0.5.12.post1
 sentry-kafka-schemas==1.3.18
 sentry-ophio==1.1.3
diff --git a/src/sentry/consumers/dlq.py b/src/sentry/consumers/dlq.py
@@ -66,6 +66,7 @@ def _get_dlq_producer(topic: Topic | None) -> KafkaDlqProducer | None:
 
     topic_defn = get_topic_definition(topic)
     config = get_kafka_producer_cluster_options(topic_defn["cluster"])
+    config["client.id"] = f"sentry.consumers.dlq.{topic.value}"
     real_topic = topic_defn["real_topic_name"]
     return KafkaDlqProducer(KafkaProducer(config), ArroyoTopic(real_topic))
 
diff --git a/src/sentry/eventstream/kafka/backend.py b/src/sentry/eventstream/kafka/backend.py
@@ -41,6 +41,7 @@ def get_producer(self, topic: Topic) -> Producer:
         if topic not in self.__producers:
             cluster_name = get_topic_definition(topic)["cluster"]
             cluster_options = get_kafka_producer_cluster_options(cluster_name)
+            cluster_options["client.id"] = "sentry.eventstream.kafka"
             self.__producers[topic] = Producer(cluster_options)
 
         return self.__producers[topic]
diff --git a/src/sentry/issues/attributes.py b/src/sentry/issues/attributes.py
@@ -50,6 +50,7 @@ def _get_attribute_snapshot_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.issues.attributes"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py
@@ -39,6 +39,7 @@ def _get_occurrence_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.issues.producer"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/monitors/clock_dispatch.py b/src/sentry/monitors/clock_dispatch.py
@@ -37,6 +37,7 @@ def _get_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.monitors.clock_dispatch"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/monitors/clock_tasks/producer.py b/src/sentry/monitors/clock_tasks/producer.py
@@ -17,6 +17,7 @@ def _get_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.monitors.clock_tasks.producer"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/monitors/logic/incident_occurrence.py b/src/sentry/monitors/logic/incident_occurrence.py
@@ -45,6 +45,7 @@ def _get_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.monitors.logic.incident_occurrence"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/monitors/tasks/clock_pulse.py b/src/sentry/monitors/tasks/clock_pulse.py
@@ -36,6 +36,7 @@ def _get_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.monitors.tasks.clock_pulse"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/preprod/producer.py b/src/sentry/preprod/producer.py
@@ -21,6 +21,7 @@ def _get_preprod_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.preprod.producer"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/profiles/task.py b/src/sentry/profiles/task.py
@@ -80,6 +80,7 @@ def _get_profiles_producer_from_topic(topic: Topic) -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.profiles.task"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/replays/lib/kafka.py b/src/sentry/replays/lib/kafka.py
@@ -20,6 +20,7 @@ def _get_eap_items_producer() -> KafkaProducer:
     """Get a Kafka producer for EAP TraceItems."""
     cluster_name = get_topic_definition(Topic.SNUBA_ITEMS)["cluster"]
     producer_config = get_kafka_producer_cluster_options(cluster_name)
+    producer_config["client.id"] = "sentry.replays.lib.kafka.eap_items"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
@@ -34,6 +35,7 @@ def _get_eap_items_producer() -> KafkaProducer:
 def _get_ingest_replay_events_producer() -> KafkaProducer:
     config = get_topic_definition(Topic.INGEST_REPLAY_EVENTS)
     producer_config = get_kafka_producer_cluster_options(config["cluster"])
+    producer_config["client.id"] = "sentry.replays.lib.kafka.ingest_replay_events"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
@@ -77,8 +79,14 @@ def initialize_replays_publisher(is_async: bool = False) -> KafkaPublisher:
 
 def _init_replay_publisher(is_async: bool) -> KafkaPublisher:
     config = get_topic_definition(Topic.INGEST_REPLAY_EVENTS)
+    producer_config = get_kafka_producer_cluster_options(config["cluster"])
+    producer_config["client.id"] = (
+        "sentry.replays.lib.kafka.publisher.async"
+        if is_async
+        else "sentry.replays.lib.kafka.publisher.sync"
+    )
     return KafkaPublisher(
-        get_kafka_producer_cluster_options(config["cluster"]),
+        producer_config,
         asynchronous=is_async,
     )
 
diff --git a/src/sentry/sentry_metrics/client/kafka.py b/src/sentry/sentry_metrics/client/kafka.py
@@ -52,6 +52,7 @@ def __init__(self) -> None:
         self.kafka_topic = ArroyoTopic(topic_defn["real_topic_name"])
         cluster_name = topic_defn["cluster"]
         producer_config = get_kafka_producer_cluster_options(cluster_name)
+        producer_config["client.id"] = "sentry.sentry_metrics.client.kafka"
         self.producer: Producer = KafkaProducer(
             build_kafka_configuration(default_config=producer_config)
         )
diff --git a/src/sentry/sentry_metrics/consumers/indexer/multiprocess.py b/src/sentry/sentry_metrics/consumers/indexer/multiprocess.py
@@ -26,9 +26,9 @@ def __init__(
         producer: AbstractProducer[KafkaPayload] | None = None,
     ) -> None:
         snuba_metrics = kafka_config.get_topic_definition(output_topic)
-        self.__producer = Producer(
-            kafka_config.get_kafka_producer_cluster_options(snuba_metrics["cluster"]),
-        )
+        producer_config = kafka_config.get_kafka_producer_cluster_options(snuba_metrics["cluster"])
+        producer_config["client.id"] = "sentry.sentry_metrics.multiprocess"
+        self.__producer = Producer(producer_config)
         self.__producer_topic = snuba_metrics["real_topic_name"]
 
         self.__commit = CommitOffsets(commit_function)
diff --git a/src/sentry/sentry_metrics/consumers/indexer/slicing_router.py b/src/sentry/sentry_metrics/consumers/indexer/slicing_router.py
@@ -106,10 +106,14 @@ def __init__(
             current_sliceable,
             current_slice_id,
         ), configuration in settings.SLICED_KAFKA_TOPICS.items():
+            producer_config = kafka_config.get_kafka_producer_cluster_options(
+                configuration["cluster"]
+            )
+            producer_config["client.id"] = (
+                f"sentry.sentry_metrics.slicing_router.{current_sliceable}.{current_slice_id}"
+            )
             self.__slice_to_producer[current_slice_id] = MessageRoute(
-                producer=Producer(
-                    kafka_config.get_kafka_producer_cluster_options(configuration["cluster"])
-                ),
+                producer=Producer(producer_config),
                 topic=Topic(configuration["topic"]),
             )
         # All logical partitions should be routed to a slice ID that's present in the slice
diff --git a/src/sentry/spans/consumers/process/flusher.py b/src/sentry/spans/consumers/process/flusher.py
@@ -54,6 +54,8 @@ def __init__(
 
     def _default_producer_factory(self, producer_config: Mapping[str, object]) -> KafkaProducer:
         """Default factory that creates real KafkaProducers."""
+        producer_config = dict(producer_config)
+        producer_config["client.id"] = "sentry.spans.consumers.process.flusher"
         return KafkaProducer(build_kafka_producer_configuration(default_config=producer_config))
 
     def _setup_producers(self):
diff --git a/src/sentry/spans/consumers/process_segments/factory.py b/src/sentry/spans/consumers/process_segments/factory.py
@@ -62,6 +62,7 @@ def __init__(
         # sime time, assuming some upper bound of spans per segment.
         self.kafka_queue_size = self.max_batch_size * self.num_processes * SPANS_PER_SEG_P95
         producer_config["queue.buffering.max.messages"] = self.kafka_queue_size
+        producer_config["client.id"] = "sentry.spans.consumers.process_segments"
 
         self.producer = KafkaProducer(
             build_kafka_producer_configuration(default_config=producer_config),
diff --git a/src/sentry/uptime/consumers/eap_producer.py b/src/sentry/uptime/consumers/eap_producer.py
@@ -26,6 +26,7 @@ def _get_eap_items_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.uptime.consumers.eap_producer"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/uptime/consumers/results_consumer.py b/src/sentry/uptime/consumers/results_consumer.py
@@ -71,6 +71,7 @@ def _get_snuba_uptime_checks_producer() -> KafkaProducer:
     producer_config = get_kafka_producer_cluster_options(cluster_name)
     producer_config.pop("compression.type", None)
     producer_config.pop("message.max.bytes", None)
+    producer_config["client.id"] = "sentry.uptime.consumers.results_consumer"
     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
 
 
diff --git a/src/sentry/usage_accountant/accountant.py b/src/sentry/usage_accountant/accountant.py
@@ -51,6 +51,7 @@ def record(
             Topic.SHARED_RESOURCES_USAGE,
         )["cluster"]
         producer_config = get_kafka_producer_cluster_options(cluster_name)
+        producer_config["client.id"] = "sentry.usage_accountant"
         producer = KafkaProducer(
             build_kafka_configuration(
                 default_config=producer_config,
diff --git a/uv.lock b/uv.lock
