ref(pull): clean up pull consumer wiring and get pull-arroyo improvments

rust_snuba/src/accepted_outcomes_consumer.rs

@@ -180,7 +180,8 @@ pub fn accepted_outcomes_consumer_impl(
     let dlq_policy = consumer_config.dlq_topic.map(|dlq_topic_config| {
         let dlq_producer_config =
             KafkaConfig::new_producer_config(vec![], Some(dlq_topic_config.broker_config));
-        let dlq_producer = KafkaProducer::new(dlq_producer_config);
+        let dlq_producer =
+            KafkaProducer::new(dlq_producer_config).expect("failed to create kafka producer");
 
         let kafka_dlq_producer = Box::new(KafkaDlqProducer::new(
             dlq_producer,
@@ -219,7 +220,8 @@ pub fn accepted_outcomes_consumer_impl(
     // TODO consider adding a higher linger.ms to the config for batching produced messages
     let producer_config =
         KafkaConfig::new_producer_config(vec![], Some(topic_config.broker_config));
-    let producer = Arc::new(KafkaProducer::new(producer_config));
+    let producer =
+        Arc::new(KafkaProducer::new(producer_config).expect("failed to create kafka producer"));
     let produce_topic = Topic::new(&topic_config.physical_topic_name);
 
     let factory = AcceptedOutcomesStrategyFactory {

rust_snuba/src/consumer.rs

@@ -224,7 +224,8 @@ pub fn consumer_impl(
             let producer = KafkaProducer::new(KafkaConfig::new_producer_config(
                 vec![],
                 Some(dlq_topic_config.broker_config),
-            ));
+            ))
+            .expect("failed to create kafka producer");
 
             let kafka_dlq_producer = Box::new(KafkaDlqProducer::new(
                 producer,
@@ -252,7 +253,8 @@ pub fn consumer_impl(
     } else if let Some(topic_config) = consumer_config.commit_log_topic {
         let producer_config =
             KafkaConfig::new_producer_config(vec![], Some(topic_config.broker_config));
-        let producer = KafkaProducer::new(producer_config);
+        let producer =
+            KafkaProducer::new(producer_config).expect("failed to create kafka producer");
         Some((
             Arc::new(producer),
             Topic::new(&topic_config.physical_topic_name),

rust_snuba/src/factory_v2.rs

@@ -246,7 +246,8 @@ impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactoryV2 {
                         self.replacements_config.clone().expect(
                             "replacements topic required for processors that emit replacements",
                         );
-                    let producer = KafkaProducer::new(replacements_config);
+                    let producer = KafkaProducer::new(replacements_config)
+                        .expect("failed to create kafka producer");
                     ProduceReplacements::new(
                         next_step,
                         producer,

rust_snuba/src/pull/pipelines/eap.rs

@@ -64,16 +64,16 @@ impl EapPipeline {
 impl Pipeline for EapPipeline {
     type Output = BatchMetadata;
 
-    fn stream<'a>(
-        &'a self,
-        source: impl Stream<Item = StageResult<KafkaPayload>> + 'a,
-    ) -> impl Stream<Item = StageResult<BatchMetadata>> + 'a {
+    fn stream(
+        self,
+        source: impl Stream<Item = StageResult<KafkaPayload>> + Send,
+    ) -> impl Stream<Item = StageResult<BatchMetadata>> + Send {
         source
-            .apply_concurrent(&self.processor, self.processing_concurrency)
-            .apply_with_timer(&self.batch, self.idle_timeout, self.max_batch_time)
-            .apply_concurrent(&self.writer, self.writer_concurrency)
-            .apply(&self.commit_log)
-            .apply(&self.cogs)
-            .on_reject(&self.dlq_handler)
+            .apply_concurrent(self.processor, self.processing_concurrency)
+            .apply_with_timer(self.batch, self.idle_timeout, self.max_batch_time)
+            .apply_concurrent(self.writer, self.writer_concurrency)
+            .apply(self.commit_log)
+            .apply(self.cogs)
+            .on_reject(self.dlq_handler)
     }
 }

rust_snuba/src/pull/pipelines/fire_and_forget.rs

@@ -56,14 +56,14 @@ impl FireAndForgetPipeline {
 impl Pipeline for FireAndForgetPipeline {
     type Output = BatchMetadata;
 
-    fn stream<'a>(
-        &'a self,
-        source: impl Stream<Item = StageResult<KafkaPayload>> + 'a,
-    ) -> impl Stream<Item = StageResult<BatchMetadata>> + 'a {
+    fn stream(
+        self,
+        source: impl Stream<Item = StageResult<KafkaPayload>> + Send,
+    ) -> impl Stream<Item = StageResult<BatchMetadata>> + Send {
         source
-            .apply_concurrent(&self.processor, self.processing_concurrency)
-            .apply_with_timer(&self.batch, self.idle_timeout, self.max_batch_time)
-            .apply_concurrent(&self.writer, self.writer_concurrency)
-            .on_reject(&self.rejection_handler)
+            .apply_concurrent(self.processor, self.processing_concurrency)
+            .apply_with_timer(self.batch, self.idle_timeout, self.max_batch_time)
+            .apply_concurrent(self.writer, self.writer_concurrency)
+            .on_reject(self.rejection_handler)
     }
 }

rust_snuba/src/pull/stages/clickhouse_writer_stage.rs

@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use std::time::Instant;
 
 use chrono::Utc;
@@ -14,14 +15,12 @@ use crate::pull::writer::ClickHouseWriter;
 /// commit log offsets, COGS data, and write stats for downstream
 /// handlers. Row bytes are freed after the write.
 pub struct ClickHouseWriterStage {
-    writer: Box<dyn ClickHouseWriter>,
+    writer: Arc<dyn ClickHouseWriter>,
 }
 
 impl ClickHouseWriterStage {
-    pub fn new(writer: impl ClickHouseWriter + 'static) -> Self {
-        Self {
-            writer: Box::new(writer),
-        }
+    pub fn new(writer: Arc<dyn ClickHouseWriter>) -> Self {
+        Self { writer }
     }
 }
 

rust_snuba/src/pull/stages/cogs_stage.rs

@@ -37,12 +37,12 @@ pub struct CogsStage {
 
 impl CogsStage {
     pub fn new(
-        producer: impl Producer<KafkaPayload> + 'static,
+        producer: Arc<dyn Producer<KafkaPayload>>,
         destination: Topic,
         resource_id: String,
     ) -> Self {
         Self {
-            producer: Arc::new(producer),
+            producer,
             destination: TopicOrPartition::Topic(destination),
             resource_id,
             logged_warning: AtomicBool::new(false),

rust_snuba/src/pull/stages/commit_log_stage.rs

@@ -31,13 +31,13 @@ pub struct CommitLogStage {
 
 impl CommitLogStage {
     pub fn new(
-        producer: impl Producer<KafkaPayload> + 'static,
+        producer: Arc<dyn Producer<KafkaPayload>>,
         destination: Topic,
         source_topic: Topic,
         consumer_group: String,
     ) -> Self {
         Self {
-            producer: Arc::new(producer),
+            producer,
             destination: TopicOrPartition::Topic(destination),
             source_topic,
             consumer_group,

rust_snuba/src/pull/test_fixtures/sources.rs

@@ -6,7 +6,6 @@ use sentry_arroyo::processing::stream::{
     MessageMetadata, OffsetCommitter, PipelineEnvelope, PullSource, StageResult,
 };
 use sentry_arroyo::types::{Partition, Topic};
-use std::sync::Arc;
 
 /// In-memory source for testing. Constructs envelopes from raw payloads
 /// with sequential offsets on partition 0.
@@ -26,7 +25,7 @@ impl VecSource {
                     offset: i as u64,
                     timestamp: chrono::Utc::now(),
                 };
-                StageResult::Emit(PipelineEnvelope::new(kp.clone(), md, Arc::new(kp)))
+                StageResult::Emit(PipelineEnvelope::new(kp.clone(), md, kp))
             })
             .collect();
 

rust_snuba/src/pull/tests/eap_pipeline_tests.rs

@@ -63,22 +63,22 @@ async fn test_eap_pipeline() {
         ProcessorStage::new(processor, ProcessorConfig::default()),
         1, // processing_concurrency
         DlqHandler::new(
-            dlq_producer,
+            Arc::new(dlq_producer),
             TopicOrPartition::Topic(Topic::new("snuba-dead-letter-items")),
         ),
         BatchStage::new(PipelineBatchBuffer::new(), 2, u64::MAX),
         Some(Duration::from_secs(2)),
         None,
-        ClickHouseWriterStage::new(Arc::clone(&writer)),
+        ClickHouseWriterStage::new(writer.clone()),
         2,
         CommitLogStage::new(
-            commit_log_producer,
+            Arc::new(commit_log_producer),
             Topic::new("snuba-items-commit-log"),
             Topic::new("snuba-items"),
             "test-group".to_string(),
         ),
         CogsStage::new(
-            cogs_producer,
+            Arc::new(cogs_producer),
             Topic::new("shared-resources-usage"),
             "eap_items_processor".to_string(),
         ),

rust_snuba/src/pull/tests/processor_pipeline_tests.rs

@@ -72,7 +72,7 @@ async fn test_fire_and_forget_pipeline(#[case] processor_name: &str, #[case] top
         BatchStage::new(PipelineBatchBuffer::new(), 2, u64::MAX),
         Some(Duration::from_secs(2)),
         None,
-        ClickHouseWriterStage::new(Arc::clone(&writer)),
+        ClickHouseWriterStage::new(writer.clone()),
         2,
     );
 

rust_snuba/src/pull/writer/clickhouse_writer.rs

@@ -13,9 +13,3 @@ use futures::future::BoxFuture;
 pub trait ClickHouseWriter: Send + Sync {
     fn write(&self, body: Vec<u8>) -> BoxFuture<'_, anyhow::Result<()>>;
 }
-
-impl<T: ClickHouseWriter> ClickHouseWriter for std::sync::Arc<T> {
-    fn write(&self, body: Vec<u8>) -> BoxFuture<'_, anyhow::Result<()>> {
-        (**self).write(body)
-    }
-}