fix: Kafka consumer not picking messages (#1441)

Signed-off-by: Benjamin Voiturier <[email protected]>
hyperledger · Nov 19, 2024 · b0effec · b0effec
1 parent b12315c
commit b0effec
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/cloud-agent/service/server/src/main/resources/logback.xml b/cloud-agent/service/server/src/main/resources/logback.xml
@@ -13,6 +13,9 @@
   <logger name="io.getquill" level="WARN">
     <appender-ref ref="STDOUT" />
   </logger>
+  <logger name="org.apache.kafka.clients" level="WARN">
+    <appender-ref ref="STDOUT" />
+  </logger>
   <root level="info">
     <appender-ref ref="STDOUT"/>
   </root>

diff --git a/...ain/scala/org/hyperledger/identus/shared/messaging/kafka/ZKafkaMessagingServiceImpl.scala b/...ain/scala/org/hyperledger/identus/shared/messaging/kafka/ZKafkaMessagingServiceImpl.scala
@@ -9,6 +9,7 @@ import zio.kafka.consumer.{
   ConsumerSettings as ZKConsumerSettings,
   Subscription as ZKSubscription
 }
+import zio.kafka.consumer.Consumer.{AutoOffsetStrategy, OffsetRetrieval}
 import zio.kafka.producer.{Producer as ZKProducer, ProducerSettings as ZKProducerSettings}
 import zio.kafka.serde.{Deserializer as ZKDeserializer, Serializer as ZKSerializer}
 
@@ -80,7 +81,7 @@ class ZKafkaConsumerImpl[K, V](
         .withMaxPollInterval(maxPollInterval) // Should be max.poll.records x 'max processing time per record'
         // 'pollTimeout' default is 50 millis. This is a ZIO Kafka property.
         .withPollTimeout(pollTimeout)
-        // .withOffsetRetrieval(OffsetRetrieval.Auto(AutoOffsetStrategy.Earliest))
+        .withOffsetRetrieval(OffsetRetrieval.Auto(AutoOffsetStrategy.Earliest))
         .withRebalanceSafeCommits(rebalanceSafeCommits)
         // .withMaxRebalanceDuration(30.seconds)
     )