comments

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

Hide file tree

Showing 5 changed files with 6 additions and 6 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);
 

diff --git a/...ngestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaPartitionConsumerTests.java b/...ngestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaPartitionConsumerTests.java
@@ -42,7 +42,7 @@ public void setUp() throws Exception {
         super.setUp();
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "test-topic");
-        params.put("bootstrapServers", "localhost:9092");
+        params.put("bootstrap_servers", "localhost:9092");
 
         config = new KafkaSourceConfig(params);
         mockConsumer = mock(KafkaConsumer.class);
@@ -93,7 +93,7 @@ public void testLatestPointer() {
     public void testTopicDoesNotExist() {
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "non-existent-topic");
-        params.put("bootstrapServers", "localhost:9092");
+        params.put("bootstrap_servers", "localhost:9092");
         var kafkaSourceConfig = new KafkaSourceConfig(params);
         when(mockConsumer.partitionsFor(eq("non-existent-topic"), any(Duration.class))).thenReturn(null);
         try {

diff --git a/...ins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaSourceConfigTests.java b/...ins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaSourceConfigTests.java
@@ -19,7 +19,7 @@ public class KafkaSourceConfigTests extends OpenSearchTestCase {
     public void testConstructorAndGetters() {
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "topic");
-        params.put("bootstrapServers", "bootstrap");
+        params.put("bootstrap_servers", "bootstrap");
 
         KafkaSourceConfig config = new KafkaSourceConfig(params);