From 907fdde716b24bc352f863c06337eda647bb1a59 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 3 Sep 2022 08:25:04 +0200 Subject: [PATCH] =?utf8?q?R=C3=BCckbau=20der=20Deaktivierung=20der=20Commi?= =?utf8?q?ts,=20um=20den=20Code=20anzugleichen?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Hier wurde die Deaktivierung der Commits im Falle eines Fachlogik-Fehlers zurückgebaut, um die einzelnen Schritte der Übung leichter nachvollziehbar zu machen. * Die Deaktivierung der Commits im Falle eines Fachlogik-Fehlers (aka Exception in der Fachlogik) ist nötig, da ansonsten in dem Rebalance, das durch das `unsubscribe()` ausgelöst wird, die aktuelle Offset-Position aus sicht der Kafka-Client-Library commited wird. * *Dies führt dazu, dass der zugehörige Testfall fehl schlägt* * Dies würde dann zu einem Verlust von Nachrichten führen, da die Client-Bibliothek davon ausgeht, dass alle vom letzten `poll()` gelieferten Nachrichten erfolgreich verarbeitet wurden. * D.h., nach einem Fachlogik-Fehler darf _kein_ Commit über die Convenient-Methode des `KafkaConsumer` durchgeführt werden, die automatisch die Offsets bestätigt, die die Bibliothek für aktuell hält. --- README.sh | 2 +- .../juplo/kafka/ApplicationConfiguration.java | 3 - .../kafka/ApplicationRebalanceListener.java | 104 ++++++------------ .../java/de/juplo/kafka/EndlessConsumer.java | 4 +- .../de/juplo/kafka/RebalanceListener.java | 3 - 5 files changed, 35 insertions(+), 81 deletions(-) diff --git a/README.sh b/README.sh index 5f11d12..6be4b11 100755 --- a/README.sh +++ b/README.sh @@ -19,7 +19,7 @@ if [[ ]] then docker-compose rm -svf adder-1 adder-2 - mvn clean install || exit + mvn -D skipTests clean install || exit else echo "Using image existing images:" docker image ls $IMAGE diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index e4ac1ab..9bfd075 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -7,7 +7,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import java.time.Clock; import java.util.Optional; import java.util.Properties; import java.util.concurrent.ExecutorService; @@ -49,8 +48,6 @@ public class ApplicationConfiguration stateRepository, properties.getClientId(), properties.getTopic(), - Clock.systemDefaultZone(), - properties.getCommitInterval(), consumer); } diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java index f3e6c8a..3653197 100644 --- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java +++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java @@ -5,9 +5,6 @@ import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.common.TopicPartition; -import java.time.Clock; -import java.time.Duration; -import java.time.Instant; import java.util.*; @@ -20,15 +17,10 @@ public class ApplicationRebalanceListener implements RebalanceListener private final StateRepository stateRepository; private final String id; private final String topic; - private final Clock clock; - private final Duration commitInterval; private final Consumer consumer; private final Set partitions = new HashSet<>(); - private Instant lastCommit = Instant.EPOCH; - private boolean commitsEnabled = true; - @Override public void onPartitionsAssigned(Collection partitions) { @@ -74,31 +66,24 @@ public class ApplicationRebalanceListener implements RebalanceListener Integer partition = tp.partition(); log.info("{} - removing partition: {}", id, partition); this.partitions.remove(partition); - if (commitsEnabled) + Map state = recordHandler.removePartition(partition); + Long offset = consumer.position(tp); + log.info( + "{} - offset of next unseen message for partition {} is {}", + id, + partition, + offset); + for (String user : state.keySet()) { - Map state = recordHandler.removePartition(partition); - Long offset = consumer.position(tp); log.info( - "{} - offset of next unseen message for partition {} is {}", + "{} - Saved state for partition={}|user={}: {}", id, partition, - offset); - for (String user : state.keySet()) - { - log.info( - "{} - Saved state for partition={}|user={}: {}", - id, - partition, - user, - state.get(user)); - } - Map> results = adderResults.removePartition(partition); - stateRepository.save(new StateDocument(partition, state, results, offset)); - } - else - { - log.info("{} - Offset commits are disabled! Last commit: {}", id, lastCommit); + user, + state.get(user)); } + Map> results = adderResults.removePartition(partition); + stateRepository.save(new StateDocument(partition, state, results, offset)); }); } @@ -106,52 +91,29 @@ public class ApplicationRebalanceListener implements RebalanceListener @Override public void beforeNextPoll() { - if (!commitsEnabled) - { - log.info("{} - Offset commits are disabled! Last commit: {}", id, lastCommit); - return; - } - - if (lastCommit.plus(commitInterval).isBefore(clock.instant())) - { - partitions - .stream() - .forEach(partition -> + partitions + .stream() + .forEach(partition -> + { + log.info("{} - persisting state & offset for partition: {}", id, partition); + Map state = recordHandler.getState(partition).getState(); + Long offset = consumer.position(new TopicPartition(topic, partition)); + log.info( + "{} - offset of next unseen message for partition {} is {}", + id, + partition, + offset); + for (String user : state.keySet()) { - log.info("{} - persisting state & offset for partition: {}", id, partition); - Map state = recordHandler.getState(partition).getState(); - Long offset = consumer.position(new TopicPartition(topic, partition)); log.info( - "{} - offset of next unseen message for partition {} is {}", + "{} - Saved state for partition={}|user={}: {}", id, partition, - offset); - for (String user : state.keySet()) - { - log.info( - "{} - Saved state for partition={}|user={}: {}", - id, - partition, - user, - state.get(user)); - } - Map> results = adderResults.getState(partition); - stateRepository.save(new StateDocument(partition, state, results, offset)); - }); - - lastCommit = clock.instant(); - } - } - - @Override - public void enableCommits() - { - commitsEnabled = true; - } - - @Override - public void disableCommits() - { - commitsEnabled = false; + user, + state.get(user)); + } + Map> results = adderResults.getState(partition); + stateRepository.save(new StateDocument(partition, state, results, offset)); + }); } } diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index 8f5eab3..892c6c9 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -42,7 +42,6 @@ public class EndlessConsumer implements Runnable try { log.info("{} - Subscribing to topic {}", id, topic); - rebalanceListener.enableCommits(); consumer.subscribe(Arrays.asList(topic), rebalanceListener); while (true) @@ -92,8 +91,7 @@ public class EndlessConsumer implements Runnable } catch(Exception e) { - log.error("{} - Unexpected error: {}, disabling commits", id, e.toString(), e); - rebalanceListener.disableCommits(); + log.error("{} - Unexpected error: {}", id, e.toString(), e); shutdown(e); } finally diff --git a/src/main/java/de/juplo/kafka/RebalanceListener.java b/src/main/java/de/juplo/kafka/RebalanceListener.java index 497a111..3c1208f 100644 --- a/src/main/java/de/juplo/kafka/RebalanceListener.java +++ b/src/main/java/de/juplo/kafka/RebalanceListener.java @@ -6,7 +6,4 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; public interface RebalanceListener extends ConsumerRebalanceListener { default void beforeNextPoll() {} - - default void enableCommits() {} - default void disableCommits() {} } -- 2.20.1