.

osskit · Nov 19, 2023 · c7b5c37 · c7b5c37
1 parent 553ba63
commit c7b5c37
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/src/main/scala/config/KafkaConfig.scala b/src/main/scala/config/KafkaConfig.scala
@@ -78,10 +78,9 @@ object KafkaConfig {
           ("bootstrap.servers", broker),
           ("linger.ms", lingerTime),
           ("max.block.ms", maxBlockMS),
-          //todo: add condition here
-          ("partitioner.class", if (useGroupPartitioner) "partitioners.GroupPartitioner" else "org.apache.kafka.clients.producer.internals.DefaultPartitioner")
-        ) ++ batchSize.map(size => Seq(("batch.size", size))).getOrElse(Seq())
-
+        ) ++
+        batchSize.map(size => Seq(("batch.size", size))).getOrElse(Seq()) ++
+        (if (useGroupPartitioner) Seq(("partitioner.class", "partitioners.GroupPartitioner")) else Seq.empty)
 
       KafkaConfig(readinessTopic, producerConfig)
     }

diff --git a/src/main/scala/partitioners/GroupPartitioner.scala b/src/main/scala/partitioners/GroupPartitioner.scala
@@ -2,6 +2,7 @@ package partitioners
 
 import org.apache.kafka.clients.producer.Partitioner
 import org.apache.kafka.common.Cluster
+import ciris.{ConfigDecoder, ConfigValue, Effect, env}
 
 class GroupPartitioner extends Partitioner {
   private val partitionsPerGroup = 5