`EndlessConsumer` nimmt jetzt einzelne `ConsumerRecord`s entgegen
authorKai Moritz <kai@juplo.de>
Sat, 10 Sep 2022 12:15:11 +0000 (14:15 +0200)
committerKai Moritz <kai@juplo.de>
Sat, 10 Sep 2022 16:08:45 +0000 (18:08 +0200)
commit316c89aa6aafbc339eda3727638c75f5489c0a99
tree4d523851efc26d5306f80c3ced5efcb12394f982
parentf095f71a104fcde025a63f87ba75eb5cb3136656
`EndlessConsumer` nimmt jetzt einzelne `ConsumerRecord`s entgegen

* `@KafkaHandler` von Batch-Verarbeitung auf Einzel-Verarbeitung umgestellt.
* Den `ApplicationErrorHandler` um eine passende Fehler-Verarbeitung für
  die einzelne Verarbeitung der Nachrichten ergänzt
* Da der `MessageListenerContainer` nicht dazu zu bewegen ist, die
  Offset-Commits im Fehlerfall zu unterlassen, wird explizit ein Seek auf
  die Offset-Positionen der noch nicht verarbeiteten Nachrichten
  durchgeführt.
* Dabei wurde ein von Spring Kafka abgeschauter Trick verwendet: Es genügt,
  die bisher unverarbeiteten Nachrichten durchzugehen und jeweils den
  Offset der ersten Nachricht, die zu einer Partition gesehen wird, für
  den Seek vorzumerken. Denn wenn dabei für eine Partition keine Nachricht
  gefunden wird, hat entweder das letzte `poll() keine Nachricht zu der
  Partition geliefert, oder alle Nachrichten, die zu der Partition gehört
  haben, wurden erfolgreich verarbeitet. In beiden Fällen stimmt der
  Offset bereits, den die Kafka-Bibliothek intern pflegt, so dass kein
  Seek durchgeführt werden muss!
* Der Testfall wurde entsprechend angepasst und läuft daher in dieser
  Variante auch ohne Fehler, da der gespeicherte Zustand dadurch zu den
  bestätigten Offsets passt.
src/main/java/de/juplo/kafka/ApplicationErrorHandler.java
src/main/java/de/juplo/kafka/EndlessConsumer.java
src/test/java/de/juplo/kafka/GenericApplicationTests.java