feat(producer): Rolling backpressure out to FTP instances again

pyproject.toml

@@ -85,7 +85,7 @@ dependencies = [
   "rfc3339-validator>=0.1.2",
   "rfc3986-validator>=0.1.1",
   # [end] jsonschema format validators
-  "sentry-arroyo>=2.41.4",
+  "sentry-arroyo>=2.42.0",
   "sentry-conventions>=0.17.0",
   "sentry-forked-email-reply-parser>=0.5.12.post1",
   "sentry-kafka-schemas>=2.2.0",

src/sentry/issues/producer.py

@@ -6,7 +6,7 @@
 from uuid import uuid4
 
 from arroyo import Topic as ArroyoTopic
-from arroyo.backends.kafka import FutureTrackingProducer, KafkaPayload, KafkaProducer
+from arroyo.backends.kafka import KafkaPayload, KafkaProducer
 from arroyo.types import Message, Value
 from confluent_kafka import KafkaException
 from django.conf import settings
@@ -17,7 +17,7 @@
 from sentry.issues.run import process_message
 from sentry.issues.status_change_message import StatusChangeMessage
 from sentry.utils import json
-from sentry.utils.arroyo_producer import get_arroyo_producer
+from sentry.utils.arroyo_producer import get_arroyo_producer, get_producer
 from sentry.utils.kafka_config import get_topic_definition
 from sentry.utils.safe import get_path, set_path
 
@@ -44,8 +44,8 @@ def _get_occurrence_producer() -> KafkaProducer:
     )
 
 
-_occurrence_producer = FutureTrackingProducer(
-    name="sentry.issues.producer",
+_occurrence_producer = get_producer(
+    producer_name="sentry.issues.producer",
     producer_factory=_get_occurrence_producer,
 )
 

src/sentry/options/defaults.py

@@ -3964,6 +3964,14 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+# If True, FutureTrackingProducer will backpressure on produce futures
+register(
+    "arroyo.ftp.backpressure",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 
 # Arroyo producer poll metrics should be logged every X poll iterations.
 register(

src/sentry/replays/lib/kafka.py

@@ -10,7 +10,7 @@
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
 from sentry.options.rollout import in_random_rollout
 from sentry.taskworker.producer import get_task_producer
-from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer
+from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer, get_producer
 from sentry.utils.kafka_config import get_topic_definition
 
 #
@@ -48,8 +48,8 @@ def _get_eap_items_producer(name: str = "sentry.replays.lib.kafka.eap_items"):
 
 eap_producer = SingletonProducer(_get_eap_items_producer)
 _eap_task_producer_name = "sentry.replays.lib.kafka.eap_items_ftp"
-eap_items_ft_producer = FutureTrackingProducer(
-    name=_eap_task_producer_name,
+eap_items_ft_producer = get_producer(
+    producer_name=_eap_task_producer_name,
     producer_factory=partial(_get_eap_items_producer, name=_eap_task_producer_name),
 )
 

src/sentry/utils/arroyo_producer.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
 import atexit
+import os
 from collections import deque
 from collections.abc import Callable
 
 from arroyo.backends.abstract import ProducerFuture
-from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_producer_configuration
+from arroyo.backends.kafka import (
+    FutureTrackingProducer,
+    KafkaPayload,
+    KafkaProducer,
+    build_kafka_producer_configuration,
+)
+from arroyo.backends.kafka.producer import CloseableProducerProtocol
 from arroyo.types import BrokerValue, Partition
 from arroyo.types import Topic as ArroyoTopic
 
@@ -125,3 +132,45 @@ def get_arroyo_producer(
         poll_metric_frequency=poll_metric_frequency,
         **kafka_producer_kwargs,
     )
+
+
+class _FutureTrackingProducer(FutureTrackingProducer):
+    """
+    HACK(bmckerry): We can't read options during module import (since there's no option cache yet),
+    so this class overrides FutureTrackingProducer's `_backpressure` attribute to be read from the option at
+    runtime. This will be deleted once FTP backpressure is enabled everywhere.
+    """
+
+    _resolved_backpressure: bool | None = None
+
+    @property
+    def _backpressure(self) -> bool:
+        """
+        This makes the `self._backpressure` check during produce() read the option.
+        """
+        if self._resolved_backpressure is None:
+            self._resolved_backpressure = options.get("arroyo.ftp.backpressure")
+        return self._resolved_backpressure
+
+    @_backpressure.setter
+    def _backpressure(self, value: bool) -> None:
+        """
+        This overrides when FTP assigns `self._backpressure` during __init__ to be a no-op.
+        """
+        pass
+
+
+def get_producer(
+    producer_name: str,
+    producer_factory: Callable[[], CloseableProducerProtocol],
+) -> FutureTrackingProducer:
+    """
+    Helper function to get a FutureTrackingProducer instance with future tracking
+    controlled by the ARROYO_TRACK_PRODUCER_FUTURES environment variable.
+    """
+
+    return _FutureTrackingProducer(
+        name=producer_name,
+        producer_factory=producer_factory,
+        should_track_futures=os.getenv("ARROYO_TRACK_PRODUCER_FUTURES", "").lower() == "true",
+    )

src/sentry/utils/outcomes.py

@@ -8,13 +8,13 @@
 from enum import IntEnum
 from threading import Lock
 
-from arroyo.backends.kafka import FutureTrackingProducer, KafkaPayload, KafkaProducer
+from arroyo.backends.kafka import KafkaPayload, KafkaProducer
 from arroyo.types import Topic as ArroyoTopic
 
 from sentry.conf.types.kafka_definition import Topic
 from sentry.constants import DataCategory
 from sentry.utils import json, kafka_config, metrics
-from sentry.utils.arroyo_producer import get_arroyo_producer
+from sentry.utils.arroyo_producer import get_arroyo_producer, get_producer
 from sentry.utils.dates import to_datetime
 
 # Aggregation key for grouping outcomes
@@ -173,11 +173,11 @@ def _get_billing_producer() -> KafkaProducer:
     )
 
 
-outcomes_producer = FutureTrackingProducer(
+outcomes_producer = get_producer(
     "sentry.utils.outcomes",
     _get_outcomes_producer,
 )
-billing_producer = FutureTrackingProducer(
+billing_producer = get_producer(
     "sentry.utils.outcomes.billing",
     _get_billing_producer,
 )