X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;ds=sidebyside;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;h=0bfee676e7140ad6bb4c4aad87deca36a2369436;hb=9fe70ad96d2b5a9ed0581057da54facba859ad1f;hp=eef0d00d73e07f8eda6053d3334498f439e6f7d7;hpb=9f8bcc8f869626efa0a258b3116582c8354a66d1;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index eef0d00..0bfee67 100644
--- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@ -2,7 +2,6 @@ package de.juplo.kafka;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
 
@@ -17,7 +16,6 @@ public class ApplicationRebalanceListener implements ConsumerRebalanceListener
   private final AdderResults adderResults;
   private final StateRepository stateRepository;
   private final String id;
-  private final Consumer consumer;
 
   private final Set<Integer> partitions = new HashSet<>();
 
@@ -50,16 +48,6 @@ public class ApplicationRebalanceListener implements ConsumerRebalanceListener
   @Override
   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
   {
-    log.info("{} - Commiting offsets for all previously assigned partitions", id);
-    try
-    {
-      consumer.commitSync();
-    }
-    catch (Exception e)
-    {
-      log.warn("{} - Could not commit offsets in onPartitionsRevoked():", id, e);
-    }
-
     partitions.forEach(tp ->
     {
       Integer partition = tp.partition();