feat: allow configuring client.id for rust capture (#24662)

rust/capture/src/config.rs

@@ -85,4 +85,6 @@ pub struct KafkaConfig {
     pub kafka_heatmaps_topic: String,
     #[envconfig(default = "false")]
     pub kafka_tls: bool,
+    #[envconfig(default = "")]
+    pub kafka_client_id: String,
 }

rust/capture/src/sinks/kafka.rs

@@ -144,6 +144,10 @@ impl KafkaSink {
                 (config.kafka_producer_queue_mib * 1024).to_string(),
             );
 
+        if !&config.kafka_client_id.is_empty() {
+            client_config.set("client.id", &config.kafka_client_id);
+        }
+
         if config.kafka_tls {
             client_config
                 .set("security.protocol", "ssl")
@@ -358,6 +362,7 @@ mod tests {
             kafka_exceptions_topic: "events_plugin_ingestion".to_string(),
             kafka_heatmaps_topic: "events_plugin_ingestion".to_string(),
             kafka_tls: false,
+            kafka_client_id: "".to_string(),
         };
         let sink = KafkaSink::new(config, handle, limiter).expect("failed to create sink");
         (cluster, sink)

rust/capture/tests/common.rs

@@ -50,6 +50,7 @@ pub static DEFAULT_CONFIG: Lazy<Config> = Lazy::new(|| Config {
         kafka_exceptions_topic: "events_plugin_ingestion".to_string(),
         kafka_heatmaps_topic: "events_plugin_ingestion".to_string(),
         kafka_tls: false,
+        kafka_client_id: "".to_string(),
     },
     otel_url: None,
     otel_sampling_rate: 0.0,