fix(preprod): use separate producer for preprod EAP (#105197)

Realized I accidentally piggy-backed off Replay's Kafka producer, so
this creates our own named producer. I also renamed
`produce_preprod_size_metric_to_eap` to make it more obvious it goes
through Kafka and therefore not a synchronous write.

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: trevor-e

src/sentry/preprod/eap/write.py

@@ -4,26 +4,28 @@
 from typing import Any
 
 from arroyo import Topic as ArroyoTopic
-from arroyo.backends.kafka import KafkaPayload
+from arroyo.backends.kafka import KafkaPayload, KafkaProducer
 from google.protobuf.timestamp_pb2 import Timestamp
+from sentry_kafka_schemas.codecs import Codec
 from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
+from sentry_protos.snuba.v1.trace_item_pb2 import TraceItem
 from sentry_protos.snuba.v1.trace_item_pb2 import TraceItem as EAPTraceItem
 
-from sentry.conf.types.kafka_definition import Topic
+from sentry.conf.types.kafka_definition import Topic, get_topic_codec
 from sentry.preprod.eap.constants import PREPROD_NAMESPACE
 from sentry.preprod.models import PreprodArtifactSizeMetrics
-from sentry.replays.lib.kafka import EAP_ITEMS_CODEC, eap_producer
 from sentry.search.eap.rpc_utils import anyvalue
+from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer
 from sentry.utils.kafka_config import get_topic_definition
 
 
-def write_preprod_size_metric_to_eap(
+def produce_preprod_size_metric_to_eap(
     size_metric: PreprodArtifactSizeMetrics,
     organization_id: int,
     project_id: int,
 ) -> None:
     """
-    Write a PreprodArtifactSizeMetrics to EAP as a TRACE_ITEM_TYPE_PREPROD trace item.
+    Write a PreprodArtifactSizeMetrics to EAP topic as a TRACE_ITEM_TYPE_PREPROD trace item.
 
     NOTE: EAP is append-only, so this function should only be called after the size metric has been successfully committed
     since we cannot update fields later on. EAP does support ReplacingMergeTree deduplication,
@@ -110,4 +112,18 @@ def write_preprod_size_metric_to_eap(
 
     topic = get_topic_definition(Topic.SNUBA_ITEMS)["real_topic_name"]
     payload = KafkaPayload(None, EAP_ITEMS_CODEC.encode(trace_item), [])
-    eap_producer.produce(ArroyoTopic(topic), payload)
+    _eap_producer.produce(ArroyoTopic(topic), payload)
+
+
+EAP_ITEMS_CODEC: Codec[TraceItem] = get_topic_codec(Topic.SNUBA_ITEMS)
+
+
+def _get_eap_items_producer() -> KafkaProducer:
+    """Get a Kafka producer for EAP TraceItems."""
+    return get_arroyo_producer(
+        name="sentry.preprod.lib.kafka.eap_items",
+        topic=Topic.SNUBA_ITEMS,
+    )
+
+
+_eap_producer = SingletonProducer(_get_eap_items_producer)

src/sentry/preprod/tasks.py

@@ -15,7 +15,7 @@
 from sentry.models.commitcomparison import CommitComparison
 from sentry.models.organization import Organization
 from sentry.models.project import Project
-from sentry.preprod.eap.write import write_preprod_size_metric_to_eap
+from sentry.preprod.eap.write import produce_preprod_size_metric_to_eap
 from sentry.preprod.models import (
     PreprodArtifact,
     PreprodArtifactSizeComparison,
@@ -477,7 +477,7 @@ def _assemble_preprod_artifact_size_analysis(
             organization = preprod_artifact.project.organization
             if features.has("organizations:preprod-size-metrics-eap-write", organization):
                 for size_metric in size_metrics_updated:
-                    write_preprod_size_metric_to_eap(
+                    produce_preprod_size_metric_to_eap(
                         size_metric=size_metric,
                         organization_id=org_id,
                         project_id=project.id,

tests/sentry/preprod/eap/test_write.py

@@ -6,7 +6,7 @@
 
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
 from sentry.models.commitcomparison import CommitComparison
-from sentry.preprod.eap.write import write_preprod_size_metric_to_eap
+from sentry.preprod.eap.write import produce_preprod_size_metric_to_eap
 from sentry.preprod.models import (
     PreprodArtifact,
     PreprodArtifactSizeMetrics,
@@ -16,7 +16,7 @@
 
 
 class WritePreprodSizeMetricToEAPTest(TestCase):
-    @patch("sentry.preprod.eap.write.eap_producer.produce")
+    @patch("sentry.preprod.eap.write._eap_producer.produce")
     def test_write_preprod_size_metric_encodes_all_fields_correctly(self, mock_produce):
         commit_comparison = CommitComparison.objects.create(
             organization_id=self.organization.id,
@@ -61,7 +61,7 @@ def test_write_preprod_size_metric_encodes_all_fields_correctly(self, mock_produ
             analysis_file_id=123,
         )
 
-        write_preprod_size_metric_to_eap(
+        produce_preprod_size_metric_to_eap(
             size_metric=size_metric,
             organization_id=self.organization.id,
             project_id=self.project.id,
@@ -117,7 +117,7 @@ def test_write_preprod_size_metric_encodes_all_fields_correctly(self, mock_produ
         assert attrs["git_base_ref"].string_value == "main"
         assert attrs["git_pr_number"].int_value == 42
 
-    @patch("sentry.preprod.eap.write.eap_producer.produce")
+    @patch("sentry.preprod.eap.write._eap_producer.produce")
     def test_write_preprod_size_metric_handles_optional_fields(self, mock_produce):
         artifact = self.create_preprod_artifact(
             project=self.project,
@@ -130,7 +130,7 @@ def test_write_preprod_size_metric_handles_optional_fields(self, mock_produce):
             state=PreprodArtifactSizeMetrics.SizeAnalysisState.COMPLETED,
         )
 
-        write_preprod_size_metric_to_eap(
+        produce_preprod_size_metric_to_eap(
             size_metric=size_metric,
             organization_id=self.organization.id,
             project_id=self.project.id,

tests/sentry/preprod/test_tasks.py

@@ -755,15 +755,15 @@ def test_assemble_preprod_artifact_size_analysis_removes_stale_metrics(self) ->
     def test_assemble_preprod_artifact_size_analysis_writes_to_eap_when_flag_enabled(self) -> None:
         """Test that size metrics are written to EAP when feature flag is enabled"""
         with self.feature("organizations:preprod-size-metrics-eap-write"):
-            with patch("sentry.preprod.tasks.write_preprod_size_metric_to_eap") as mock_eap_write:
+            with patch("sentry.preprod.tasks.produce_preprod_size_metric_to_eap") as mock_eap_write:
                 status, details = self._run_task_and_verify_status(
                     b'{"analysis_duration": 1.5, "download_size": 1000, "install_size": 2000, "treemap": null, "analysis_version": null, "app_components": [{"component_type": 0, "name": "Main App", "app_id": "com.example.app", "path": "/", "download_size": 1000, "install_size": 2000}]}'
                 )
 
                 assert status == ChunkFileState.OK
                 assert details is None
 
-                # Verify write_preprod_size_metric_to_eap was called exactly once
+                # Verify produce_preprod_size_metric_to_eap was called exactly once
                 # We have other integration tests that verify the EAP write itself works
                 assert mock_eap_write.call_count == 1
 
@@ -776,15 +776,15 @@ def test_assemble_preprod_artifact_size_analysis_writes_to_eap_when_flag_enabled
 
     def test_assemble_preprod_artifact_size_analysis_skips_eap_when_flag_disabled(self) -> None:
         """Test that size metrics are NOT written to EAP when feature flag is disabled"""
-        with patch("sentry.preprod.tasks.write_preprod_size_metric_to_eap") as mock_eap_write:
+        with patch("sentry.preprod.tasks.produce_preprod_size_metric_to_eap") as mock_eap_write:
             status, details = self._run_task_and_verify_status(
                 b'{"analysis_duration": 1.5, "download_size": 1000, "install_size": 2000, "treemap": null, "analysis_version": null, "app_components": [{"component_type": 0, "name": "Main App", "app_id": "com.example.app", "path": "/", "download_size": 1000, "install_size": 2000}]}'
             )
 
             assert status == ChunkFileState.OK
             assert details is None
 
-            # Verify write_preprod_size_metric_to_eap was NOT called
+            # Verify produce_preprod_size_metric_to_eap was NOT called
             mock_eap_write.assert_not_called()
 
     def test_assemble_preprod_artifact_size_analysis_eap_write_failure_does_not_fail_task(
@@ -793,7 +793,7 @@ def test_assemble_preprod_artifact_size_analysis_eap_write_failure_does_not_fail
         """Test that EAP write failures don't cause the main task to fail"""
         with self.feature("organizations:preprod-size-metrics-eap-write"):
             with patch(
-                "sentry.preprod.tasks.write_preprod_size_metric_to_eap",
+                "sentry.preprod.tasks.produce_preprod_size_metric_to_eap",
                 side_effect=Exception("EAP write failed"),
             ):
                 status, details = self._run_task_and_verify_status(
@@ -814,7 +814,7 @@ def test_assemble_preprod_artifact_size_analysis_eap_write_failure_does_not_fail
     def test_assemble_preprod_artifact_size_analysis_writes_multiple_metrics_to_eap(self) -> None:
         """Test that all size metrics (main + components) are written to EAP when flag is enabled"""
         with self.feature("organizations:preprod-size-metrics-eap-write"):
-            with patch("sentry.preprod.tasks.write_preprod_size_metric_to_eap") as mock_eap_write:
+            with patch("sentry.preprod.tasks.produce_preprod_size_metric_to_eap") as mock_eap_write:
                 status, details = self._run_task_and_verify_status(
                     b'{"analysis_duration": 2.5, "download_size": 5000, "install_size": 10000, "treemap": null, "analysis_version": "1.0", "app_components": [{"component_type": 0, "name": "Main App", "app_id": "com.example.app", "path": "/", "download_size": 3000, "install_size": 6000}, {"component_type": 1, "name": "Watch App", "app_id": "com.example.app.watchkitapp", "path": "/Watch", "download_size": 2000, "install_size": 4000}]}'
                 )

tests/snuba/preprod/eap/test_preprod_eap_integration.py

@@ -7,7 +7,7 @@
 
 from sentry.models.commitcomparison import CommitComparison
 from sentry.preprod.eap.read import query_preprod_size_metrics
-from sentry.preprod.eap.write import write_preprod_size_metric_to_eap
+from sentry.preprod.eap.write import produce_preprod_size_metric_to_eap
 from sentry.preprod.models import (
     PreprodArtifact,
     PreprodArtifactSizeMetrics,
@@ -58,7 +58,7 @@ def test_write_and_read_size_metric_round_trip(self):
             analysis_file_id=123,
         )
 
-        write_preprod_size_metric_to_eap(
+        produce_preprod_size_metric_to_eap(
             size_metric=size_metric,
             organization_id=self.organization.id,
             project_id=self.project.id,
@@ -153,13 +153,13 @@ def test_write_multiple_size_metrics_same_artifact(self):
             max_install_size=1000,
         )
 
-        write_preprod_size_metric_to_eap(
+        produce_preprod_size_metric_to_eap(
             size_metric=size_metric_main,
             organization_id=self.organization.id,
             project_id=self.project.id,
         )
 
-        write_preprod_size_metric_to_eap(
+        produce_preprod_size_metric_to_eap(
             size_metric=size_metric_watch,
             organization_id=self.organization.id,
             project_id=self.project.id,