ref(kafka): Always use Arroyo producers

README.md

@@ -80,6 +80,8 @@ which cluster it lives on. Exactly one topic must be consumable; the rest
 (retry, dead-letter) are `produce_only`. The dead-letter and retry topics must
 share a cluster with the upkeep producer.
 
+Kafka producers use Arroyo for retries, deadletters, and demoted-task forwarding.
+
 ```yaml
 kafka_deadletter_topic: taskworker-dlq
 

src/kafka/activation_batcher.rs

@@ -5,10 +5,9 @@ use std::time::{Duration, Instant};
 use chrono::Utc;
 use futures::future::join_all;
 use rdkafka::config::ClientConfig;
-use tracing::error;
 
 use crate::config::Config;
-use crate::kafka::producer::ProducerBackend;
+use crate::kafka::producer::KafkaProducer;
 use crate::runtime_config::RuntimeConfigManager;
 use crate::store::activation::Activation;
 
@@ -50,7 +49,7 @@ pub struct ActivationBatcher {
     forward_batch: Vec<Vec<u8>>, // payload
     config: ActivationBatcherConfig,
     runtime_config_manager: Arc<RuntimeConfigManager>,
-    producer: Arc<ProducerBackend>,
+    producer: Arc<KafkaProducer>,
     producer_cluster: String,
 }
 
@@ -60,9 +59,8 @@ impl ActivationBatcher {
         runtime_config_manager: Arc<RuntimeConfigManager>,
     ) -> Self {
         let producer = Arc::new(
-            ProducerBackend::new(
+            KafkaProducer::new(
                 config.producer_config.clone(),
-                false,
                 Duration::from_millis(config.send_timeout_ms),
             )
             .expect("Could not create kafka producer in activation batcher"),
@@ -198,36 +196,14 @@ impl Reducer for ActivationBatcher {
                 let mut new_config = self.config.producer_config.clone();
                 new_config.set("bootstrap.servers", &forward_cluster);
                 self.producer = Arc::new(
-                    ProducerBackend::new(
+                    KafkaProducer::new(
                         new_config,
-                        self.producer.is_arroyo(),
                         Duration::from_millis(self.config.send_timeout_ms),
                     )
                     .expect("Could not create kafka producer in activation batcher"),
                 );
                 self.producer_cluster = forward_cluster;
             }
-            if self.producer.is_arroyo() != runtime_config.use_arroyo_producer {
-                let mut new_config = self.config.producer_config.clone();
-                new_config.set("bootstrap.servers", &self.producer_cluster);
-                match ProducerBackend::new(
-                    new_config,
-                    runtime_config.use_arroyo_producer,
-                    Duration::from_millis(self.config.send_timeout_ms),
-                ) {
-                    Ok(new_producer) => self.producer = Arc::new(new_producer),
-                    Err(err) => {
-                        let target = if runtime_config.use_arroyo_producer {
-                            "Arroyo"
-                        } else {
-                            "rdkafka"
-                        };
-                        error!(
-                            "Could not switch kafka producer in activation batcher to {target}: {err}"
-                        );
-                    }
-                }
-            }
             let forward_topic = runtime_config
                 .demoted_topic
                 .clone()

src/kafka/producer.rs

@@ -1,97 +1,56 @@
 use std::time::{Duration, Instant};
 
-use rdkafka::ClientConfig;
-use rdkafka::error::{KafkaError, RDKafkaErrorCode};
-use rdkafka::producer::{FutureProducer, FutureRecord};
-use rdkafka::util::Timeout;
-
 use sentry_arroyo::backends::kafka::config::KafkaConfig;
 use sentry_arroyo::backends::kafka::producer::AsyncKafkaProducer;
 use sentry_arroyo::backends::kafka::types::KafkaPayload;
 use sentry_arroyo::backends::{AsyncProducer, ProducerError};
 use sentry_arroyo::types::{Topic, TopicOrPartition};
 
-/// A producer for one backend. Callers replace it when the runtime flag changes.
-pub enum ProducerBackend {
-    Rdkafka {
-        producer: FutureProducer,
-        queue_timeout: Duration,
-    },
-    Arroyo {
-        producer: AsyncKafkaProducer,
-        queue_timeout: Duration,
-    },
-}
+use rdkafka::ClientConfig;
+use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 
-impl ProducerBackend {
-    pub fn is_arroyo(&self) -> bool {
-        matches!(self, Self::Arroyo { .. })
-    }
+/// An Arroyo producer with bounded retries when the local queue is full.
+pub struct KafkaProducer {
+    producer: AsyncKafkaProducer,
+    queue_timeout: Duration,
+}
 
-    pub fn new(
-        config: ClientConfig,
-        use_arroyo: bool,
-        queue_timeout: Duration,
-    ) -> Result<Self, KafkaError> {
-        if use_arroyo {
-            let bootstrap_servers = config
-                .get("bootstrap.servers")
-                .expect("producer config always sets bootstrap.servers")
-                .to_owned();
-            let arroyo_config = KafkaConfig::new_producer_config(
-                vec![bootstrap_servers],
-                Some(config.config_map().clone()),
-            );
-            Ok(Self::Arroyo {
-                producer: AsyncKafkaProducer::new(arroyo_config)?,
-                queue_timeout,
-            })
-        } else {
-            Ok(Self::Rdkafka {
-                producer: config.create()?,
-                queue_timeout,
-            })
-        }
+impl KafkaProducer {
+    pub fn new(config: ClientConfig, queue_timeout: Duration) -> Result<Self, KafkaError> {
+        let bootstrap_servers = config
+            .get("bootstrap.servers")
+            .expect("producer config always sets bootstrap.servers")
+            .to_owned();
+        let arroyo_config = KafkaConfig::new_producer_config(
+            vec![bootstrap_servers],
+            Some(config.config_map().clone()),
+        );
+        Ok(Self {
+            producer: AsyncKafkaProducer::new(arroyo_config)?,
+            queue_timeout,
+        })
     }
 
     pub async fn send(&self, topic: &str, payload: &[u8]) -> Result<(), ProducerError> {
-        match self {
-            Self::Rdkafka {
-                producer,
-                queue_timeout,
-            } => producer
-                .send(
-                    FutureRecord::<(), [u8]>::to(topic).payload(payload),
-                    Timeout::After(*queue_timeout),
-                )
-                .await
-                .map(|_| ())
-                .map_err(|(error, _)| error.into()),
-            Self::Arroyo {
-                producer,
-                queue_timeout,
-            } => {
-                let destination = TopicOrPartition::Topic(Topic::new(topic));
-                let payload = KafkaPayload::new(None, None, Some(payload.to_vec()));
-                let deadline = Instant::now() + *queue_timeout;
-
-                loop {
-                    match producer.produce(&destination, payload.clone()).await {
-                        Err(
-                            error @ ProducerError::Kafka(KafkaError::MessageProduction(
-                                RDKafkaErrorCode::QueueFull,
-                            )),
-                        ) => {
-                            let remaining = deadline.saturating_duration_since(Instant::now());
-                            if remaining.is_zero() {
-                                return Err(error);
-                            }
-                            // Keep rdkafka's retry interval without sleeping past the deadline.
-                            tokio::time::sleep(remaining.min(Duration::from_millis(100))).await;
-                        }
-                        result => return result,
+        let destination = TopicOrPartition::Topic(Topic::new(topic));
+        let payload = KafkaPayload::new(None, None, Some(payload.to_vec()));
+        let deadline = Instant::now() + self.queue_timeout;
+
+        loop {
+            match self.producer.produce(&destination, payload.clone()).await {
+                Err(
+                    error @ ProducerError::Kafka(KafkaError::MessageProduction(
+                        RDKafkaErrorCode::QueueFull,
+                    )),
+                ) => {
+                    let remaining = deadline.saturating_duration_since(Instant::now());
+                    if remaining.is_zero() {
+                        return Err(error);
                     }
+                    // Keep rdkafka's retry interval without sleeping past the deadline.
+                    tokio::time::sleep(remaining.min(Duration::from_millis(100))).await;
                 }
+                result => return result,
             }
         }
     }
@@ -108,16 +67,7 @@ mod tests {
     use sentry_arroyo::backends::{AsyncProducer, ProducerError};
     use sentry_arroyo::types::{Topic, TopicOrPartition};
 
-    use super::ProducerBackend;
-
-    #[test]
-    fn flag_selects_arroyo_producer() {
-        let mut config = ClientConfig::new();
-        config.set("bootstrap.servers", "127.0.0.1:1");
-        let producer = ProducerBackend::new(config, true, Duration::from_millis(500)).unwrap();
-
-        assert!(producer.is_arroyo());
-    }
+    use super::KafkaProducer;
 
     #[tokio::test]
     async fn arroyo_retries_queue_full_until_timeout() {
@@ -126,20 +76,17 @@ mod tests {
             .set("bootstrap.servers", "127.0.0.1:1")
             .set("queue.buffering.max.messages", "1")
             .set("message.timeout.ms", "5000");
-        let backend = ProducerBackend::new(config, true, Duration::from_millis(20)).unwrap();
-        let ProducerBackend::Arroyo { producer, .. } = &backend else {
-            unreachable!();
-        };
+        let producer = KafkaProducer::new(config, Duration::from_millis(20)).unwrap();
 
         // Fill the queue without polling the delivery future.
         let destination = TopicOrPartition::Topic(Topic::new("test"));
-        let _first = producer.produce(
+        let _first = producer.producer.produce(
             &destination,
             KafkaPayload::new(None, None, Some(b"first".to_vec())),
         );
 
         let started = Instant::now();
-        let result = backend.send("test", b"second").await;
+        let result = producer.send("test", b"second").await;
         assert!(matches!(
             result,
             Err(ProducerError::Kafka(KafkaError::MessageProduction(

src/runtime_config.rs

@@ -26,9 +26,6 @@ pub struct RuntimeConfig {
     /// The topic to forward tasks from demoted namespaces to.
     /// If not set, the taskworker-long topic will be used
     pub demoted_topic: Option<String>,
-    /// Use Arroyo's async Kafka producer for retries, deadletters, and forwarding.
-    #[serde(default)]
-    pub use_arroyo_producer: bool,
 }
 
 pub struct RuntimeConfigManager {
@@ -221,8 +218,7 @@ drop_task_killswitch: []
 demoted_namespaces:
   - bad_namespace
 demoted_topic_cluster: kafka:9092
-demoted_topic: taskworker-demoted-topic
-use_arroyo_producer: true"#;
+demoted_topic: taskworker-demoted-topic"#;
 
         let mut config_file = NamedTempFile::new().unwrap();
         writeln!(config_file, "{}", test_yaml).unwrap();
@@ -231,7 +227,6 @@ use_arroyo_producer: true"#;
         let runtime_config =
             RuntimeConfigManager::new(Some(config_file.path().to_str().unwrap().to_string())).await;
         let config = runtime_config.read().await;
-        assert!(config.use_arroyo_producer);
         assert_eq!(
             config.demoted_topic_cluster.as_deref().unwrap(),
             "kafka:9092"

src/test_utils.rs

@@ -17,7 +17,7 @@ use uuid::Uuid;
 use crate::config::deprecated::DeprecatedConfig;
 use crate::config::store::{PgConfig, StoreConfig};
 use crate::config::{Config, DEFAULT_TOPIC};
-use crate::kafka::producer::ProducerBackend;
+use crate::kafka::producer::KafkaProducer;
 use crate::store::activation::{Activation, ActivationBuilder, ActivationStatus};
 use crate::store::adapters::postgres::{self, PostgresStore};
 use crate::store::adapters::sqlite::SqliteStore;
@@ -340,14 +340,9 @@ pub fn create_integration_config() -> Config {
 }
 
 /// Create a kafka producer for a given config
-pub fn create_producer(config: Arc<Config>) -> Arc<ProducerBackend> {
-    create_producer_with_flag(config, false)
-}
-
-pub fn create_producer_with_flag(config: Arc<Config>, use_arroyo: bool) -> Arc<ProducerBackend> {
-    let producer = ProducerBackend::new(
+pub fn create_producer(config: Arc<Config>) -> Arc<KafkaProducer> {
+    let producer = KafkaProducer::new(
         config.kafka_producer_config(),
-        use_arroyo,
         Duration::from_millis(config.kafka_send_timeout_ms),
     )
     .expect("Could not create kafka producer");