feat: add config for kafka consumer partition assignment strategy (#209)

* feat: add config for kafka consumer partition assignment strategy * chore: version bump * chore: bump version to 0.7.1 * chore: version bump
raystack · Jan 18, 2023 · f95c194 · f95c194
1 parent 01f688d
commit f95c194
Show file tree

Hide file tree

Showing 4 changed files with 9 additions and 1 deletion.
diff --git a/build.gradle b/build.gradle
@@ -33,7 +33,7 @@ lombok {
 }
 
 group 'io.odpf'
-version '0.7.1'
+version '0.7.2'
 
 def projName = "firehose"
 

diff --git a/docs/docs/advance/generic.md b/docs/docs/advance/generic.md
diff --git a/src/main/java/io/odpf/firehose/config/KafkaConsumerConfig.java b/src/main/java/io/odpf/firehose/config/KafkaConsumerConfig.java
@@ -52,4 +52,9 @@ public interface KafkaConsumerConfig extends AppConfig {
     @Key("SOURCE_KAFKA_CONSUMER_CONFIG_MANUAL_COMMIT_MIN_INTERVAL_MS")
     @DefaultValue("-1")
     long getSourceKafkaConsumerManualCommitMinIntervalMs();
+
+    @Key("SOURCE_KAFKA_CONSUMER_CONFIG_PARTITION_ASSIGNMENT_STRATEGY")
+    @DefaultValue("org.apache.kafka.clients.consumer.CooperativeStickyAssignor")
+    String getSourceKafkaConsumerConfigPartitionAssignmentStrategy();
+
 }
diff --git a/src/main/java/io/odpf/firehose/utils/KafkaUtils.java b/src/main/java/io/odpf/firehose/utils/KafkaUtils.java
@@ -30,6 +30,8 @@ public class KafkaUtils {
     private static final String METADATA_MAX_AGE_MS = "metadata.max.age.ms";
     private static final String MAX_POLL_RECORDS = "max.poll.records";
     private static final String SESSION_TIMEOUT_MS = "session.timeout.ms";
+    private static final String PARTITION_ASSIGNMENT_STRATEGY = "partition.assignment.strategy";
+
 
 
     /**
@@ -56,6 +58,7 @@ public static Map<String, Object> getConfig(KafkaConsumerConfig config, Map<Stri
             put(METADATA_MAX_AGE_MS, config.getSourceKafkaConsumerConfigMetadataMaxAgeMs());
             put(MAX_POLL_RECORDS, config.getSourceKafkaConsumerConfigMaxPollRecords());
             put(SESSION_TIMEOUT_MS, config.getSourceKafkaConsumerConfigSessionTimeoutMs());
+            put(PARTITION_ASSIGNMENT_STRATEGY, config.getSourceKafkaConsumerConfigPartitionAssignmentStrategy());
         }};
 
         return merge(consumerConfigurationMap, KafkaEnvironmentVariables.parse(extraParameters));