ref(producer): Port producers to ReloadingKafkaProducer

Cargo.toml

@@ -37,7 +37,7 @@ validator = { version = "0.20.0", features = ["derive"] }
 prost-types = "0.14"
 rand = "0.8.5"
 rdkafka = { version = "0.37.0", features = ["cmake-build", "ssl"] }
-sentry_arroyo = { version = "=2.45.0", features = ["ssl"] }
+sentry_arroyo = { git = "https://github.com/getsentry/arroyo", branch = "markus/reloading-producer-delivery-futures", features = ["ssl"] }
 rmp-serde = "1.3"
 sentry = { version = "0.41.0", default-features = false, features = [
     # default features, except `release-health` is disabled

src/kafka/activation_batcher.rs

@@ -68,6 +68,7 @@ impl ActivationBatcher {
         let producer = Arc::new(
             KafkaProducer::new(
                 config.producer_config.clone(),
+                &config.kafka_deadletter_topic,
                 Duration::from_millis(config.send_timeout_ms),
             )
             .expect("Could not create kafka producer in activation batcher"),
@@ -105,8 +106,12 @@ impl ActivationBatcher {
             let mut config = self.config.producer_config.clone();
             config.set("bootstrap.servers", &cluster);
             self.producer = Arc::new(
-                KafkaProducer::new(config, Duration::from_millis(self.config.send_timeout_ms))
-                    .expect("Could not create kafka producer in activation batcher"),
+                KafkaProducer::new(
+                    config,
+                    &self.config.kafka_deadletter_topic,
+                    Duration::from_millis(self.config.send_timeout_ms),
+                )
+                .expect("Could not create kafka producer in activation batcher"),
             );
             self.producer_cluster = cluster;
         }

src/kafka/producer.rs

@@ -1,7 +1,10 @@
 use std::time::{Duration, Instant};
 
 use sentry_arroyo::backends::kafka::config::KafkaConfig;
-use sentry_arroyo::backends::kafka::producer::AsyncKafkaProducer;
+use sentry_arroyo::backends::kafka::config_blob::ProducerSelector;
+use sentry_arroyo::backends::kafka::reloading_producer::{
+    ReloadConfig, ReloadError, ReloadingKafkaProducer,
+};
 use sentry_arroyo::backends::kafka::types::KafkaPayload;
 use sentry_arroyo::backends::{AsyncProducer, ProducerError};
 use sentry_arroyo::types::{Topic, TopicOrPartition};
@@ -11,12 +14,19 @@ use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 
 /// An Arroyo producer with bounded retries when the local queue is full.
 pub struct KafkaProducer {
-    producer: AsyncKafkaProducer,
+    producer: ReloadingKafkaProducer,
     queue_timeout: Duration,
 }
 
 impl KafkaProducer {
-    pub fn new(config: ClientConfig, queue_timeout: Duration) -> Result<Self, KafkaError> {
+    /// `topic` only picks this producer's entry in the shared Kafka config.
+    /// It is not the only topic the producer can send to: other topics are
+    /// sent to under their own name, on the same cluster.
+    pub fn new(
+        config: ClientConfig,
+        topic: &str,
+        queue_timeout: Duration,
+    ) -> Result<Self, ReloadError> {
         let bootstrap_servers = config
             .get("bootstrap.servers")
             .expect("producer config always sets bootstrap.servers")
@@ -25,8 +35,19 @@ impl KafkaProducer {
             vec![bootstrap_servers],
             Some(config.config_map().clone()),
         );
+        let settings = ReloadConfig {
+            // Taskbroker sends messages without a key, so their order does not
+            // matter. Swap clients right away instead of holding back sends
+            // until the old client has drained.
+            ignore_key_ordering: true,
+            ..ReloadConfig::default()
+        };
         Ok(Self {
-            producer: AsyncKafkaProducer::new(arroyo_config)?,
+            producer: ReloadingKafkaProducer::from_config(
+                arroyo_config,
+                ProducerSelector::new(topic, "taskbroker"),
+                settings,
+            )?,
             queue_timeout,
         })
     }
@@ -76,7 +97,7 @@ mod tests {
             .set("bootstrap.servers", "127.0.0.1:1")
             .set("queue.buffering.max.messages", "1")
             .set("message.timeout.ms", "5000");
-        let producer = KafkaProducer::new(config, Duration::from_millis(20)).unwrap();
+        let producer = KafkaProducer::new(config, "test", Duration::from_millis(20)).unwrap();
 
         // Fill the queue without polling the delivery future.
         let destination = TopicOrPartition::Topic(Topic::new("test"));

src/test_utils.rs

@@ -343,6 +343,7 @@ pub fn create_integration_config() -> Config {
 pub fn create_producer(config: Arc<Config>) -> Arc<KafkaProducer> {
     let producer = KafkaProducer::new(
         config.kafka_producer_config(),
+        &config.kafka_deadletter_topic,
         Duration::from_millis(config.kafka_send_timeout_ms),
     )
     .expect("Could not create kafka producer");

src/upkeep.rs

@@ -37,6 +37,7 @@ pub async fn upkeep(
     let producer = Arc::new(
         KafkaProducer::new(
             kafka_config,
+            &config.kafka_deadletter_topic,
             Duration::from_millis(config.kafka_send_timeout_ms),
         )
         .expect("Could not create kafka producer in upkeep"),
@@ -367,6 +368,7 @@ pub async fn do_upkeep(
         let forward_producer = Arc::new(
             KafkaProducer::new(
                 forward_producer_config,
+                &forward_topic,
                 Duration::from_millis(config.kafka_send_timeout_ms),
             )
             .expect("Could not create kafka producer in upkeep"),