comments

Signed-off-by: Yupeng Fu <[email protected]>
opensearch-project · Jan 21, 2025 · 01126c5 · 01126c5
1 parent 8e7af9c
commit 01126c5
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/...ion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java b/...ion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java
@@ -85,7 +85,7 @@ public void testKafkaIngestion() {
                 .put("ingestion_source.type", "kafka")
                 .put("ingestion_source.pointer.init.reset", "earliest")
                 .put("ingestion_source.param.topic", "test")
-                .put("ingestion_source.param.bootstrapServers", kafka.getBootstrapServers())
+                .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
                 .build(),
             "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
         );

diff --git a/plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaSourceConfig.java b/plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaSourceConfig.java
@@ -25,7 +25,7 @@ public class KafkaSourceConfig {
     public KafkaSourceConfig(Map<String, Object> params) {
         // TODO: better parsing and validation
         this.topic = (String) Objects.requireNonNull(params.get("topic"));
-        this.bootstrapServers = (String) Objects.requireNonNull(params.get("bootstrapServers"));
+        this.bootstrapServers = (String) Objects.requireNonNull(params.get("bootstrap_servers"));
         assert this.bootstrapServers != null;
     }
 

diff --git a/.../ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaConsumerFactoryTests.java b/.../ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaConsumerFactoryTests.java
@@ -19,7 +19,7 @@ public void testInitialize() {
         KafkaConsumerFactory factory = new KafkaConsumerFactory();
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "test-topic");
-        params.put("bootstrapServers", "localhost:9092");
+        params.put("bootstrap_servers", "localhost:9092");
 
         factory.initialize(params);