Support custom kafka consumer configs

Signed-off-by: Varun Bharadwaj <[email protected]>

Author: varunbharadwaj

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/RemoteStoreKafkaIT.java

@@ -57,6 +57,9 @@ public void testSegmentReplicationWithRemoteStore() throws Exception {
                 .put("ingestion_source.param.topic", topicName)
                 .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
                 .put("index.replication.type", "SEGMENT")
+                // set custom kafka consumer properties
+                .put("ingestion_source.param.fetch.min.bytes", 30000)
+                .put("ingestion_source.param.enable.auto.commit", false)
                 .build(),
             mapping
         );

plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java

@@ -9,7 +9,6 @@
 package org.opensearch.plugin.kafka;
 
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -99,9 +98,10 @@ protected static Consumer<byte[], byte[]> createConsumer(String clientId, KafkaS
         Properties consumerProp = new Properties();
         consumerProp.put("bootstrap.servers", config.getBootstrapServers());
         consumerProp.put("client.id", clientId);
-        if (config.getAutoOffsetResetConfig() != null && !config.getAutoOffsetResetConfig().isEmpty()) {
-            consumerProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, config.getAutoOffsetResetConfig());
-        }
+
+        // add additional kafka configurations if provided
+        consumerProp.putAll(config.getAdditionalConfigurations());
+
         // TODO: why Class org.apache.kafka.common.serialization.StringDeserializer could not be found if set the deserializer as prop?
         // consumerProp.put("key.deserializer",
         // "org.apache.kafka.common.serialization.StringDeserializer");

plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaSourceConfig.java

@@ -10,6 +10,7 @@
 
 import org.opensearch.core.util.ConfigurationUtils;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -18,21 +19,24 @@
 public class KafkaSourceConfig {
     private final String PROP_TOPIC = "topic";
     private final String PROP_BOOTSTRAP_SERVERS = "bootstrap_servers";
-    // TODO: support pass any generic kafka configs
-    private final String PROP_AUTO_OFFSET_RESET = "auto.offset.reset";
 
     private final String topic;
     private final String bootstrapServers;
-    private final String autoOffsetResetConfig;
+
+    private final Map<String, Object> additionalConfigMap;
 
     /**
-     * Constructor
+     * Extracts and look for required and optional kafka consumer configurations.
      * @param params the configuration parameters
      */
     public KafkaSourceConfig(Map<String, Object> params) {
         this.topic = ConfigurationUtils.readStringProperty(params, PROP_TOPIC);
         this.bootstrapServers = ConfigurationUtils.readStringProperty(params, PROP_BOOTSTRAP_SERVERS);
-        this.autoOffsetResetConfig = ConfigurationUtils.readOptionalStringProperty(params, PROP_AUTO_OFFSET_RESET);
+        this.additionalConfigMap = new HashMap<>(params);
+
+        // remove required configurations initialized above from the additionalConfigMap
+        additionalConfigMap.remove(PROP_TOPIC);
+        additionalConfigMap.remove(PROP_BOOTSTRAP_SERVERS);
     }
 
     /**
@@ -60,4 +64,8 @@ public String getBootstrapServers() {
     public String getAutoOffsetResetConfig() {
         return autoOffsetResetConfig;
     }
+
+    public Map<String, Object> getAdditionalConfigurations() {
+        return additionalConfigMap;
+    }
 }

plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaSourceConfigTests.java

@@ -16,10 +16,12 @@
 
 public class KafkaSourceConfigTests extends OpenSearchTestCase {
 
-    public void testConstructorAndGetters() {
+    public void testKafkaSourceConfig() {
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "topic");
         params.put("bootstrap_servers", "bootstrap");
+        params.put("fetch.min.bytes", 30000);
+        params.put("enable.auto.commit", false);
 
         KafkaSourceConfig config = new KafkaSourceConfig(params);
 
@@ -29,5 +31,7 @@ public void testConstructorAndGetters() {
             "bootstrap",
             config.getBootstrapServers()
         );
+        Assert.assertEquals("Incorrect fetch.min.bytes", 30000, config.getAdditionalConfigurations().get("fetch.min.bytes"));
+        Assert.assertEquals("Incorrect enable.auto.commit", false, config.getAdditionalConfigurations().get("enable.auto.commit"));
     }
 }