fix(eventstream): Use an arroyo producer for eventstream

src/sentry/eventstream/kafka/backend.py

@@ -6,7 +6,6 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
-from arroyo.backends.kafka import build_kafka_producer_configuration
 from confluent_kafka import KafkaError
 from confluent_kafka import Message as KafkaMessage
 from confluent_kafka import Producer
@@ -20,8 +19,8 @@
 from sentry.eventstream.types import EventStreamEventType
 from sentry.killswitches import killswitch_matches_context
 from sentry.utils import json
-from sentry.utils.confluent_producer import get_confluent_producer
-from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
+from sentry.utils.arroyo_producer import get_arroyo_producer
+from sentry.utils.kafka_config import get_topic_definition
 
 EAP_ITEMS_CODEC: Codec[TraceItem] = get_topic_codec(Topic.SNUBA_ITEMS)
 
@@ -45,12 +44,9 @@ def get_transactions_topic(self, project_id: int) -> Topic:
 
     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"
-            # XXX(markus): We should use `sentry.utils.arroyo_producer.get_arroyo_producer`.
-            self.__producers[topic] = get_confluent_producer(
-                build_kafka_producer_configuration(default_config=cluster_options)
+            self.__producers[topic] = get_arroyo_producer(
+                name="sentry.eventstream.kafka",
+                topic=topic,
             )
 
         return self.__producers[topic]