fix(kafka): set client.id on all producers for producer_name metric tag

snuba/cli/subscriptions_executor.py

@@ -134,7 +134,10 @@ def subscriptions_executor(
         build_kafka_producer_configuration(
             result_topic_spec.topic,
             bootstrap_servers=result_bootstrap_server,
-            override_params={"partitioner": "consistent"},
+            override_params={
+                "partitioner": "consistent",
+                "client.id": "snuba-subscriptions-executor",
+            },
         )
     )
 

snuba/cli/subscriptions_scheduler.py

@@ -164,7 +164,10 @@ def subscriptions_scheduler(
             scheduled_topic_spec.topic,
             slice_id,
             bootstrap_servers=scheduled_bootstrap_server,
-            override_params={"partitioner": "consistent"},
+            override_params={
+                "partitioner": "consistent",
+                "client.id": "snuba-subscriptions-scheduler",
+            },
         )
     )
 

snuba/cli/subscriptions_scheduler_executor.py

@@ -120,7 +120,10 @@ def subscriptions_scheduler_executor(
     producer = KafkaProducer(
         build_kafka_producer_configuration(
             result_topic_spec.topic,
-            override_params={"partitioner": "consistent"},
+            override_params={
+                "partitioner": "consistent",
+                "client.id": "snuba-subscriptions-scheduler-executor",
+            },
         )
     )
 

snuba/cogs/accountant.py

@@ -18,7 +18,11 @@ def _accumulator(create: bool = False) -> UsageAccumulator | None:
     global accumulator
     if accumulator is None and create:
         producer = KafkaProducer(
-            build_kafka_producer_configuration(StreamTopic.COGS_SHARED_RESOURCES_USAGE, None)
+            build_kafka_producer_configuration(
+                StreamTopic.COGS_SHARED_RESOURCES_USAGE,
+                None,
+                override_params={"client.id": "snuba-cogs-accountant"},
+            )
         )
         accumulator = UsageAccumulator(producer=producer)
     return accumulator

snuba/state/__init__.py

@@ -56,6 +56,7 @@ def _kafka_producer() -> Producer:
         kfk = Producer(
             build_kafka_producer_configuration(
                 topic=Topic.QUERYLOG,
+                override_params={"client.id": "snuba-querylog"},
             )
         )
     return kfk

snuba/web/bulk_delete_query.py

@@ -71,6 +71,7 @@ def _get_kafka_producer(topic: Topic) -> Producer:
         producer = Producer(
             build_kafka_producer_configuration(
                 topic=topic,
+                override_params={"client.id": "snuba-bulk-delete"},
             )
         )
         PRODUCER_MAP[topic.value] = producer