Das Speichern der Daten und Offsets erfolgt nicht mehr nach jedem `poll()`

* Statdessen kann eine `Duration` konfiguriert werden.
* Ähnlich wie in der Client-Library von Kafka, wird ein Zeitstempel für
  den letzten Commit gespeichert und die Daten werden immer dann
  gespeichert, wenn dieser weiter als das eingestellte
  `consumer.commit-interval` in der Vergangenheit liegt.

Wenn kein gespeicherter Offset vorliegt, auto.offset.reset von Kafka nutzen

* Es wird jetzt nur noch dann ein expliziter Seek durchgeführt, wenn eine
  gespeicherte Offset-Position gefunden wurde.
* Andernfalls wird der von Kafka initialisierte Ausgansgs-Offset verwendet.
* Welchen Offset Kafka vorgibt, hängt von `auto.offset.rest` ab!

Fehler im Shutdown-Code korrigiert: Shutdown von `EndlessConsumer` zu spät

mongo-express sollte erst nach der MongoDB gestartet werden

Ausgabe der verarbeiteten Nachrichten im Revoke-Callback entfernt

* Es musste allein für diese Ausgabe eine Map mit den zuletzt eingelesenen
  Offset-Positionen gepflegt werden.
* Das ist zu viel Overhead, für die Randmeldung im Log.

Fehler in Logging-Ausgabe korrigiert

* Der über den Merge hinzugefügt Test hat einen Fehler aufgedeckt.
* In onPartitionsRevoked() wurde bei der Berechnung der verarbeiteten
  Nachrichten für die Log-Ausgabe ein Nullzeiger dereferenziert.
* Ursache dafür war, dass die Map `offsets` in der Version, die die Offsets
  speichert gar nicht mehr gepflegt wurde.

Merge der Refaktorisierung des EndlessConsumer (Branch 'stored-state')

* Die `commtSync()`-Aufrufe machen beim Speichern der Offsets außerhalb
  von Kafka keinen Sinn mehr.
* Der Testfall musste an die extern gespeicherten Offsets angepasst
  werden: Die gesehenen Offsets müssen aus der MongoDB gelesen werden,
  anstatt über einen separaten Consumer aus Kafka.
* Der mit dem Merge hinzugefügte Test schlägt fehl, da er einen Fehler
  aufdeckt (NPE bei einer Log-Ausgabe zur Offset-Verarbeitung).

Merge der Refaktorisierung des EndlessConsumer (Branch 'deserialization')

* Um die Implementierung besser testen zu können, wurde die Anwendung
  in dem Branch 'deserialization' refaktorisiert.
* Diese Refaktorisierung werden hier zusammen mit den eingeführten
  Tests gemerged.
* Der so verfügbar gemachte Test wurde so angepasst, dass er das Speichern
  des Zustands in einer MongoDB berücksichtigt.

Merge der überarbeiteten Compose-Konfiguration ('rebalance-listener')

Merge der überarbeiteten Compose-Konfiguration ('stored-state')

README.sh startet MongoDB und Mongo-Express

Merge der überarbeiteten Compose-Konfiguration ('rebalance-listener')

* Dabei auch die letzten Verbesserungen aus 'rebalance-listener' übernommen.

Merge der überarbeiteten Compose-Konfiguration ('counting-consumer')

Merge der überarbeiteten Compose-Konfiguration ('endless-stream-consumer')

* Die letzten Änderungen an 'endless-stream-consumer' sind länger nicht
  mehr gemerged worden.

Compose-Konfiguration unabhängig von Default-Konfiguration gemacht

* Damit Instanzen parallel über die IDE (mit voreingestelltem Default-Port)
  und Compose gestartet werden können, wurden den einzelnen Komponenten
  (Producer, Consumer etc.) jeweils unterschiedliche explizite
  Default-Ports zugewiesen.
* Dies führt leicht zu fehlern, in den Compose-Setups, da dort i.d.R.
  Port-Mappings für die gestarteten Instanzen definiert werden.
* Daher werden die Compose-Setups jetzt so umgestellt, dass sie den
  einkompilierten Default-Port der Komponenten explizit mit dem Port `8080`
  überschreiben, so dass alle Komponenten _innerhalb_ von Compose
  einheitlich (und so wie bei Spring-Boot standard) über `8080` ansprechbar
  sind.

Merge der Upgrades für Confluent/Spring-Boot (Branch 'first-contact')

Upgrade von Spring Boot und den Confluent-Kafka-Images

* Upgrade der Kafk-Images von Confluent 7.0.2 auf 7.1.3
** Unterstützt Kafka 3.1.x (siehe https://docs.confluent.io/platform/current/installation/versions-interoperability.html[Versions-Matrix])
* Upgrade für Spring Boot von 2.6.5 auf 2.7.2
** Enthält Kafka: 3.1.1
** Enthält Spring Kafka: 2.8.8

Fehler in Shutdown-Logik für den `ExecutorService` korrigiert

Merge branch 'rebalance-listener' into deserialization

Merge branch 'endless-stream-consumer' into rebalance-listener

HealthIndicator implementiert: Implementierung selbst vergessen :/

Refactor: `EndlessConsumer` ist selbst ein `ConsumerRebalanceListener`

* Vorbereitung für die Übernahme der Funktionalität in der springifizierten
  Version.

Tests: Refaktorisiert - Durcheinander bei Assertions aufgeräumt

* Zuvor wurden die Assertions von JUnit Jupiter und AssertJ durcheinander
  gewürfelt verwendet.
* Jetzt werden stringent nur noch die Assertions von AssertJ verwendet.

Tests: Refaktorisiert - Serialisierung des Payloads konfigurierbar gemacht

Tests: Tests prüfen den Status des Consumers

Tests: Refaktorisiert - Nachrichten werden für alle Tests aufgezeichnet

Tests: Fehlerfall-Test prüft, dass nicht alle Nachrichten gelesen wurden

Tests: Assert-Beschreibungen korrigiert/ergänzt

Tests: Geprüft, dass der Fehler nach einem Neustart neu vorliegt

Tests: Umbau für einen Commit im Fehlerfall und Anpassung des Tests

Tests: Refaktorisiert - Methoden-Reihenfolge aufgeräumt und kommentiert

Tests: Der Test wartet, bis die Offsets regulär committed wurden

* Zuvor wurde der Offset-Commit erzwungen, indem der EndlessConsumer
  über den Aufruf von `stop()` beendet wurde.
* Jetzt wird die Überprüfung der Erwartungen über awaitility aufgeschoben,
  bis die Erwartungen beobachtet werden können - oder eine Zeitschranke
  gerissen wird.

Tests: Der EndlessConsumer wird jetzt doch asynchron ausgeführt

* Der Test-Code wird verständlicher, wenn der Consumer asynchron läuft
* Für die Überprüfung der Erwartungen wird dann awaitility benötigt
** Da der Consumer in einem separaten Thread läuft, muss auf diesen
   gezielt gewartet werden
** Dafür wird es deutlicher/verständlicher, auf welche der auch aus dem
   regulären Betrieb des EndlessConsumer bekannten Zustandsänderungen
   der Test wartet
* Da die Offsets für die Controlle (ggf.) abgefragt werden, während der
  EndlessConsumer noch läuft, wird ein separater KafkaConsumer benötigt,
  um die Offsets abzurufen.
  BEACHTE: Um die Änderungen in dem Offsets-Topic zu greifen zu bekommen,
  muss dieser KafkaConsumer für jede Abfrage ein `assign()` durchführen,
  damit er gezwungen ist, die Offsets neu vom Broker zu erfragen.

Tests: Refaktorisiert - DRY für Test auf Offset-Fortschritt

Tests: Überprüft, ob überhaupt ein Offset-Fortschritt vorliegt

Tests: Offsets werden unter TopicPartition abgelegt

Tests: Test-Reihenfolge definiert, da das Topic nicht geleert wird

Tests: Erste Version eines synchronen Integration-Test implementiert

Demonstration der RecordDeserializationException

* EndlessConsumer wird mit dem Value-Type `Long` anstatt `String` erzeugt
* Setup für die Demonstration der DeserializationException überarbeitet

Refaktorisierung für Tests - EndlessConsumer typisiert

Refaktorisierung für Tests - Record-Handler als Bean konfigurierbar

* Ein Handler für die Verarbeitung der einzelnen ConsumerRecord's kann
  jetzt ein `java.util.function.Consumer` als Bean definiert werden
* Die Nachricht wird erst nach dem Aufruf des Handlers als konsumiert
  gezählt, damit der Handler die Nachricht über eine Exception ablehnen
  kann.

Die Spring-Boot App wird nun sauber herunter gefahren

* Zuvor wurde die App nicht richtig beendet, weil der ExecutorService nicht
  beendet wurde und sich die JVM deswegen nicht beenden konnte.
* Dies ist erst durch die Aktivierung des shutdown-Endpoints aufgefallen.
* Jetzt wurde in `Application` eine `@PreDestroy`-Methode ergänzt, die
  den ExecutorService nach allen Regeln der Kunst ordentlich beendet.

Refaktorisierung für Tests - ExecutorService als separate Bean erzeugt

Refaktorisierung für Tests - Start des EndlessConsumer in ApplicationRunner

* Bisher wurde der EndlessConsumer beim erzeugen der Bean gestartet
* Dies ist für das Aufsetzen von Tests ungünstig, da die erzeugte Bean
  dort nicht unbedingt direkt gestartet werden soll
* Daher wurde der Start des EndlessConsumer in einen ApplicationRunner
  ausgelagert

Refaktorisierung für Tests - Bean-Definition in Config-Klasse verschoben

Refaktorisierung für Tests - KafkaConsumer als eigenständige Bean

* Der KafakConsumer wird als eigenständige Bean erzeugt
* Die Bean wird dem EndlessConsumer im Konstruktor übergeben
* Dafür muss der Lebenszyklus der KafkaConsumer-Bean von dem der
  EndlessConsumer-Bean getrennt werden:
** close() darf nicht mehr im finally-Block im EndlessConsumer aufgerufen
   werden
** Stattdessen muss close() als Destry-Methode der Bean definiert werden
** Für start/stop muss stattdessen unsubscribe() im finally-Block aufgerufen
   werden
** Da unsubscribe() die Offset-Position nicht commited, muss explizit
   ein Offsset-Commit beauftragt werden, wenn der Consumer regulär
   gestoppt wird (WakeupException)

Die Offsets werden ausgegeben, wenn eine Partition zugeteilt/entzogen wird

Gesehene Schlüssel sollten als long gezählt werden

Merge branch 'endless-stream-consumer' into rebalance-listener

HealthIndicator implementiert

shutdown-Endpoint aktiviert

Informationen zur Kafka-Konfiguration im info-Endpoint sichtbar gemacht

Informationen zur Java-Umgebung im info-Endpoint aktiviert

Das Git-Commit-Id-Plugin generiert eine git.properties

Das Spring-Boot-Maven-Plugin generiert Build-Info

Default-Konfiguration überarbeitet

* Eine über die IDE bzw. Maven gestartete Instanz soll klar als solche
  erkennbar sein (`client.id` = DEV), darf dafür aber im Compose-Setup
  mitspielen (`group.id` = my-group).
* Bisher gab es häufig Port-Konflikte, wenn über die IDE bzw. über Maven
  parallel zu einem Compose-Setup eine Instanz gestartet wurde. Daher wird
  jetzt hier explizit auf einen abweichenden Port (8881) ausgewichen.
* Im Compose-Setup auch den neuen Port für den Producer übernommen

Falschen Status-Code bei start/stop-Fehler korrigiert

Verständliche Fehlermeldungen für DriverController

Merge branch 'endless-stream-consumer' into rebalance-listener

Thread-Synchronisation bei start/stop überarbeitet

Fehlerbehandlung in EndlessConsumer.destroy() korrigiert

Validierung erfolgt über spring-boot-starter-validation

Merge branch rebalance-listener into counting-consumer

* Nur wegen der Rück-Umbenennung des counting-consumer in endless-consumer
* Die anderen Änderungen aus dem Branch wurden entfernt

Merge branch 'stored-state' into stored-offsets

MongoDB Express zum Compose-Setup hinzugefügt

Consumer aus der IDE spielt mit

* Sonst läuft bei Experimenten die MongoDB auseinander
* Denn in die schreiben die Docker-Instanzen und die IDE-Instanz
* Deswegen gehört jetzt auch die IDE-Instanz zu der Consumer-Group, die
  in Docker-Compose verwendet wird.
* *ANMERKUNG:*
** Da die gezählten Keys jetzt in der MongoDB gespeichert werden, wird
   jetzt die Störung der Partitionierung durch die Änderung der Anzahl
   der Partitionen wieder sichtbar, _obwohl_ zwischendurch ein Wechsel
   der zuständigen Consumer-Instanz erfolgt
** D.h., es wird deutlich, wie sehr Zustand, der aus einer Partition
   aufgebaut wird, unter einer Änderung der Partitions-Anzahl leidet!

Merge branch 'rebalance-listener' into stored-state

Rückbau auf verschachtelte Maps

* Die zuvor erfundenen fachlichen Klassen passen nicht zu dazu, dass man
  sie - wie gedacht - direkt in MongoDB stopfen könnte.
* Daher hier erst mal der Rückbau auf Maps, da das dan für die Übungen
  einfacher ist.

Offset-Position in der MongoDB wird vor jedem `poll()` aktualisiert

Setup für die Übung "Offsets" überarbeitet

Offset-Position wird in der MongoDB gespeichert

Die Key-Statistiken werden in einer MongoDB gespeichert

counting-consumer ist eine Verbesserung von endless-consumer

Benennung der neuen fachlichen Klassen verbessert

Verschachtelte Map gegen fachliche Datenstruktur ausgetauscht

Report über gesehene Schlüssel wiederbelebt

* An der alten Stelle war die Map ja jetzt nur noch leer, da dem Consumer
  zu dem Zeitpunkt, an dem die gesehen Schlüssel ausgegeben wurden, bereits
  alle Partitionen entzogen worden sind.
* Daher werden die gesehenen Schlüssel jetzt ausgegeben, wenn eine
  Partition entzogen wird.

Rebalance-Listener anstatt Wegwerfen der Map

HTTPie gibt nicht nur den Response, sondern auch den Request aus

Merge des Upgrades der Confluent-Images auf 7.0.2

Merge des Upgrades der Confluent-Images auf 7.0.2

Upgrade der Images von Confluent 6.2.0 auf 7.0.2

Der Consumer erkennt die Änderung der Partitionierung schneller

* `metadata.max.age.ms` auf 1000 heruntergesetzt (Default 5 min)
* Dadurch wird die Änderung der Anzahl an Partitionen zeitnah erkannt

Für null-Keys wird der String NULL gezählt

README.sh zeigt die Auswirkung einer geänderten Partitions-Anzahl

Der Consumer zählt jetzt die Nachrichten pro Key für jedes Topic

* Die Ergebnisse werden beim Beenden des Consumer ausgegeben
* Wenn der Consumer neu gestartet wird, werden die Ergebnisse zurückgesetzt
* Über /seen können Zwischenstände abgefragt werden

`auto.offset.reset` konfigurierbar gemacht

Fehler bei der Erzeugung des KafkaConsumer werden nicht mehr verschluckt

* Beim Erzeugen der Properties-Instanz können Exceptions fliegen
* Beim Erzeugen der KafkaConsumer-Instanz können Exception fliegen
* Daher wurden diese Schritte in den try/catch-Block verlegt
* Neben der Nachricht wird jetzt auch der ganze Stack-Trace gelogged
* Da die Erzeugung des KafkaConsumer jetzt im try/catch-Block geschieht,
  wird der EndlessConsumer im Fehlerfall korrekt als beendet markiert

Endless Consumer: a simple consumer, implemented as Spring-Boot-App

SimpleConsumer in EndlessConsumer umbenannt

* Ohne die Umbenennung _vor_ der Veränderung würde die Versions-Historie
  der Klasse verloren gehen!

README.sh verwendet den cli-Service für Kommandos

Fälschlich hartkodiertes Topic gegen Variable getauscht

Code reorganisiert, um Änderungen vergleichbarer zu machen

Upgrade für Spring Boot 2.6.0 -> 2.6.5

* Kafka: 3.0.1
* Spring Kafka: 2.8.4

First Contact: Simple Producer & Consumer