Implementierung vereinfacht: Auf das Nötigste zusammengekürzt

* Das regelmäßige Speichern im Poll-Interval wird für die Übung nicht
  benötigt.
* Damit entfällt auch das Interface
  `PollIntervalAwareConsumerRebalanceListener`
* Die Vereinfachung hat eine Anpassung der Tests erfordert: Da in dem
  Test, der überprüft, ob die Offsets korrekt committed werde, wenn kein
  Fehler vorliegt, gar kein Rebalance auftritt, musste der Consumer
  gestoppt werden, damit die Ergebnisse für die Überprüfung sichtbar
  werden.

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index b58295f..c1bc019 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -39,9 +39,7 @@ public class ApplicationConfiguration
         recordHandler,
         adderResults,
         stateRepository,
-        properties.getClientId(),
-        Clock.systemDefaultZone(),
-        properties.getCommitInterval());
+        properties.getClientId());
   }
 
   @Bean

diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index a89c633..6d3850f 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@ -2,6 +2,7 @@ package de.juplo.kafka;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
 
 import java.time.Clock;
@@ -12,19 +13,15 @@ import java.util.*;
 
 @RequiredArgsConstructor
 @Slf4j
-public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRebalanceListener
+public class ApplicationRebalanceListener implements ConsumerRebalanceListener
 {
   private final ApplicationRecordHandler recordHandler;
   private final AdderResults adderResults;
   private final StateRepository stateRepository;
   private final String id;
-  private final Clock clock;
-  private final Duration commitInterval;
 
   private final Set<Integer> partitions = new HashSet<>();
 
-  private Instant lastCommit = Instant.EPOCH;
-
   @Override
   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
   {
@@ -73,20 +70,4 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
       stateRepository.save(new StateDocument(partition, state, results));
     });
   }
-
-
-  @Override
-  public void beforeNextPoll()
-  {
-    if (lastCommit.plus(commitInterval).isBefore(clock.instant()))
-    {
-      log.debug("Storing data, last commit: {}", lastCommit);
-      partitions.forEach(partition -> stateRepository.save(
-          new StateDocument(
-              partition,
-              recordHandler.getState(partition).getState(),
-              adderResults.getState(partition))));
-      lastCommit = clock.instant();
-    }
-  }
 }

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 0238521..00678c4 100644 (file)
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -25,7 +25,7 @@ public class EndlessConsumer<K, V> implements Runnable
   private final String id;
   private final String topic;
   private final Consumer<K, V> consumer;
-  private final PollIntervalAwareConsumerRebalanceListener rebalanceListener;
+  private final ConsumerRebalanceListener rebalanceListener;
   private final RecordHandler<K, V> recordHandler;
 
   private final Lock lock = new ReentrantLock();
@@ -67,8 +67,6 @@ public class EndlessConsumer<K, V> implements Runnable
 
           consumed++;
         }
-
-        rebalanceListener.beforeNextPoll();
       }
     }
     catch(WakeupException e)

diff --git a/src/main/java/de/juplo/kafka/PollIntervalAwareConsumerRebalanceListener.java b/src/main/java/de/juplo/kafka/PollIntervalAwareConsumerRebalanceListener.java
deleted file mode 100644 (file)
index 8abec12..0000000
--- a/src/main/java/de/juplo/kafka/PollIntervalAwareConsumerRebalanceListener.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package de.juplo.kafka;
-
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-
-
-public interface PollIntervalAwareConsumerRebalanceListener extends ConsumerRebalanceListener
-{
-  default void beforeNextPoll() {}
-}

diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
index 54137b4..595ef89 100644 (file)
--- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java
@@ -2,6 +2,7 @@ package de.juplo.kafka;
 
 import com.mongodb.client.MongoClient;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -66,7 +67,7 @@ abstract class GenericApplicationTests<K, V>
        @Autowired
        MongoProperties mongoProperties;
        @Autowired
-       PollIntervalAwareConsumerRebalanceListener rebalanceListener;
+       ConsumerRebalanceListener rebalanceListener;
        @Autowired
        RecordHandler<K, V> recordHandler;
 
@@ -91,7 +92,7 @@ abstract class GenericApplicationTests<K, V>
        /** Tests methods */
 
        @Test
-       void commitsCurrentOffsetsOnSuccess()
+       void commitsCurrentOffsetsOnSuccess() throws Exception
        {
                int numberOfGeneratedMessages =
                                recordGenerator.generate(false, false, messageSender);
@@ -114,6 +115,7 @@ abstract class GenericApplicationTests<K, V>
                                .isThrownBy(() -> endlessConsumer.exitStatus())
                                .describedAs("Consumer should still be running");
 
+               endlessConsumer.stop();
                recordGenerator.assertBusinessLogic();
        }
 
@@ -385,7 +387,6 @@ abstract class GenericApplicationTests<K, V>
        {
                try
                {
-                       endlessConsumer.stop();
                        testRecordProducer.close();
                        offsetConsumer.close();
                }