From f95c194d45f1bc2a123b29d744a55ac7c50e56e7 Mon Sep 17 00:00:00 2001
From: Sumit Aich <aich.1998@gmail.com>
Date: Wed, 18 Jan 2023 20:41:36 +0530
Subject: [PATCH] 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
---
 build.gradle                                  |   2 +-
 docs/docs/advance/generic.md                  | Bin 5671 -> 5987 bytes
 .../firehose/config/KafkaConsumerConfig.java  |   5 +++++
 .../io/odpf/firehose/utils/KafkaUtils.java    |   3 +++
 4 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/build.gradle b/build.gradle
index 733e6e33c..09f4d0d81 100644
--- 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
index 308e339135ad48f89ebcd8c23367fd14fac5a4f5..b1823857564b5938a18e710ed6659a9a8b9bd021 100644
GIT binary patch
delta 261
zcmaitF^<A86hu3U6o4qGf49wI(NkFvp{%q45jRxuv3`P;U`Mt=P;rPHf)<H`a0j>u
z0nt#=&5Y*hoqq4^{MV|wO*l^ZOhh7@#0#Dz(K!AOW3foYjEitQUk1U5Ql-g(UlYgz
zodc7;N{QAhuZ)IzRGBswyn(GFNT0F3TUud(S}L7^^A<g22L?gVgK&?VvLM)Db?r3z
z#%2sllxt!+ra3ZMDAL|BX|&sJ$g;#MnIHIDRPpxD;5#%lrleY{cJ#?#!q%<v0F!ZE
AOaK4?

delta 12
TcmaE?w_Im~7ysrqfxBz~BqIeH

diff --git a/src/main/java/io/odpf/firehose/config/KafkaConsumerConfig.java b/src/main/java/io/odpf/firehose/config/KafkaConsumerConfig.java
index 4b8ef66f0..819d96c88 100644
--- 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
index 4654e5751..c0043d1bc 100644
--- 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));