feat: Allow configurable producer acks (#27664)

PostHog · Jan 18, 2025 · 2e4be90 · 2e4be90
1 parent fb69265
commit 2e4be90
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 1 deletion.
diff --git a/rust/capture/src/config.rs b/rust/capture/src/config.rs
@@ -96,4 +96,6 @@ pub struct KafkaConfig {
     pub kafka_metadata_max_age_ms: u32,
     #[envconfig(default = "2")]
     pub kafka_producer_max_retries: u32,
+    #[envconfig(default = "all")]
+    pub kafka_producer_acks: String,
 }
diff --git a/rust/capture/src/sinks/kafka.rs b/rust/capture/src/sinks/kafka.rs
@@ -155,7 +155,8 @@ impl KafkaSink {
             .set(
                 "queue.buffering.max.kbytes",
                 (config.kafka_producer_queue_mib * 1024).to_string(),
-            );
+            )
+            .set("acks", config.kafka_producer_acks.to_string());
 
         if !&config.kafka_client_id.is_empty() {
             client_config.set("client.id", &config.kafka_client_id);
@@ -400,6 +401,7 @@ mod tests {
             kafka_client_id: "".to_string(),
             kafka_metadata_max_age_ms: 60000,
             kafka_producer_max_retries: 2,
+            kafka_producer_acks: "all".to_string(),
         };
         let sink = KafkaSink::new(config, handle, limiter, None).expect("failed to create sink");
         (cluster, sink)

diff --git a/rust/capture/tests/common.rs b/rust/capture/tests/common.rs
@@ -57,6 +57,7 @@ pub static DEFAULT_CONFIG: Lazy<Config> = Lazy::new(|| Config {
         kafka_client_id: "".to_string(),
         kafka_metadata_max_age_ms: 60000,
         kafka_producer_max_retries: 2,
+        kafka_producer_acks: "all".to_string(),
     },
     otel_url: None,
     otel_sampling_rate: 0.0,