Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Support configuring Kafka consumer group
Currently we set the Kafka consumer group based on a fixed pattern: https://github.com/ArroyoSystems/arroyo/blob/2ff997fd3bca7f25fcb7932d2395ee782d183092/arroyo-worker/src/connectors/kafka/source/mod.rs#L132-L137

For some use cases, it would be valuable to be able to set this to a particular static value.

We should enable this by adding an option to the kafka source connector.


Discussion / hints:

Fix patch:
diff --git a/arroyo-connectors/src/kafka.rs b/arroyo-connectors/src/kafka.rs
index c95630d0e..94e0e7afb 100644
--- a/arroyo-connectors/src/kafka.rs
+++ b/arroyo-connectors/src/kafka.rs
@@ -170,6 +170,7 @@ impl Connector for KafkaConnector {
                         Some("read_uncommitted") | None => Some(ReadMode::ReadUncommitted),
                         Some(other) => bail!("invalid value for source.read_mode '{}'", other),
                     },
+                    group_id: opts.remove("source.group_id"),
                 }
             }
             "sink" => {
diff --git a/arroyo-sql/src/lib.rs b/arroyo-sql/src/lib.rs
index 28c2e591d..35b001d65 100644
--- a/arroyo-sql/src/lib.rs
+++ b/arroyo-sql/src/lib.rs
@@ -603,6 +603,7 @@ pub fn get_test_expression(
                 type_: arroyo_connectors::kafka::TableType::Source {
                     offset: arroyo_connectors::kafka::SourceOffset::Latest,
                     read_mode: Some(arroyo_connectors::kafka::ReadMode::ReadUncommitted),
+                    group_id: "test-consumer-group".to_string().try_into().unwrap(),
                 },
             },
             Some(&schema),
diff --git a/arroyo-worker/src/connectors/kafka/source/mod.rs b/arroyo-worker/src/connectors/kafka/source/mod.rs
index 7b9449131..fc8903afc 100644
--- a/arroyo-worker/src/connectors/kafka/source/mod.rs
+++ b/arroyo-worker/src/connectors/kafka/source/mod.rs
@@ -33,6 +33,7 @@ where
 {
     topic: String,
     bootstrap_servers: String,
+    group_id: Option<String>,
     offset_mode: super::SourceOffset,
     format: Format,
     client_configs: HashMap<String, String>,
@@ -59,6 +60,7 @@ where
     pub fn new(
         servers: &str,
         topic: &str,
+        group: Option<String>,
         offset_mode: super::SourceOffset,
         format: Format,
         messages_per_second: u32,
@@ -67,6 +69,7 @@ where
         Self {
             topic: topic.to_string(),
             bootstrap_servers: servers.to_string(),
+            group_id: group,
             offset_mode,
             format,
             client_configs: client_configs
@@ -85,7 +88,12 @@ where
             .expect("Invalid connection config for KafkaSource");
         let table: KafkaTable =
             serde_json::from_value(config.table).expect("Invalid table config for KafkaSource");
-        let TableType::Source { offset, read_mode } = &table.type_ else {
+        let TableType::Source {
+            offset,
+            read_mode,
+            group_id,
+        } = &table.type_
+        else {
             panic!("found non-source kafka config in source operator");
         };
         let mut client_configs = client_configs(&connection);
@@ -96,6 +104,7 @@ where
         Self {
             topic: table.topic,
             bootstrap_servers: connection.bootstrap_servers.to_string(),
+            group_id: group_id.clone(),
             offset_mode: *offset,
             format: config.format.expect("Format must be set for Kafka source"),
             client_configs,
@@ -131,10 +140,12 @@ where
             .set("enable.auto.commit", "false")
             .set(
                 "group.id",
-                format!(
-                    "arroyo-{}-{}-consumer",
-                    ctx.task_info.job_id, ctx.task_info.operator_id
-                ),
+                self.group_id.clone().unwrap_or_else(|| {
+                    format!(
+                        "arroyo-{}-{}-consumer",
+                        ctx.task_info.job_id, ctx.task_info.operator_id
+                    )
+                }),
             )
             .create()?;
 
diff --git a/connector-schemas/kafka/table.json b/connector-schemas/kafka/table.json
index 019b01edd..548fcc465 100644
--- a/connector-schemas/kafka/table.json
+++ b/connector-schemas/kafka/table.json
@@ -31,6 +31,11 @@
                                 "read_committed",
                                 "read_uncommitted"
                             ]
+                        },
+                        "group_id": {
+                            "type": "string",
+                            "title": "group id",
+                            "description": "Sets the Group ID of the consumer for Kafka source. If not specified, an automatically generated ID will be used. CAUTION: Using one consumer group for multiple pipelines may result in incomplete data"
                         }
                     },
                     "required": [


Test patch:
diff --git a/arroyo-worker/src/connectors/kafka/source/test.rs b/arroyo-worker/src/connectors/kafka/source/test.rs
index 940c19b9f..d48a20e86 100644
--- a/arroyo-worker/src/connectors/kafka/source/test.rs
+++ b/arroyo-worker/src/connectors/kafka/source/test.rs
@@ -24,6 +24,7 @@ struct TestData {
 pub struct KafkaTopicTester {
     topic: String,
     server: String,
+    group_id: Option<String>,
 }
 
 impl KafkaTopicTester {
@@ -63,6 +64,7 @@ impl KafkaTopicTester {
         let mut kafka: KafkaSourceFunc<(), TestData> = KafkaSourceFunc::new(
             &self.server,
             &self.topic,
+            self.group_id.clone(),
             crate::connectors::kafka::SourceOffset::Earliest,
             Format::Json(JsonFormat::default()),
             100,
@@ -192,6 +194,7 @@ async fn test_kafka() {
     let mut kafka_topic_tester = KafkaTopicTester {
         topic: "arroyo-source".to_string(),
         server: "0.0.0.0:9092".to_string(),
+        group_id: Some("test-consumer-group".to_string()),
     };
 
     let mut task_info = arroyo_types::get_test_task_info();