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

Author: sumitaich1998

build.gradle

@@ -33,7 +33,7 @@ lombok {
 }
 
 group 'io.odpf'
-version '0.7.1'
+version '0.7.2'
 
 def projName = "firehose"
 

docs/docs/advance/generic.md

@@ -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();
+
 }

src/main/java/io/odpf/firehose/config/KafkaConsumerConfig.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));