Springify: Merge der Umstellung auf die Auto-Konfiguration von Spring-Boot

Springify: Der Kafka-`Consumer` wird über die Spring-Factory erzeugt

Springify: Merge der Umstellung des Payloads auf JSON

Springify: Konfiguration erfolgt über `KafkaProperties`

Springify: Nachrichten-Typ wird über den Type-Info-Header bestimmt

Springify: Der Payload ist eine als JSON gerenderte Klasse

* Als Nachricht wird eine Instanz der Klasse `ClientMessage` erwartet
* Die Instanz wird mit Hilfe des `JsonDeserializer` von Spring Kafka
  deserialisiert.

Springify: Refactor - Aufruf der Assertions vereinfacht

* Warum auch immer: Der Compiler erkennt den Typ des Generics in der
  springifizierten Version nur dann ohne erzwungenen Cast korrekt, wenn
  der `describeAs()`-Aufruf als letzes erfolgt.
* In der nicht springifizierten Version, ist die Aufrufreihenfolge egal:
  Der Compiler erkennt den Typ des Generics unabhängig davon korrekt.
* An ggf. transitiv angezogenen Abhängigkeiten liegt es laut
  `mvn help:effective-pom` nicht. Dies zeigt für beide Versionen (abgesehen
  von dem explizit ergänzten `spring-kafka`) exakt die selben
  Abhängigkeiten.

Springify: Merge des verschärften Tests aus der Vanilla-Version

* Logik zur Abfrage der Exception wiederbelebt, an der ein über eine
  Poison Pill gestolperter `KafkaConsumer` gestorben ist, damit die
  springifizierte Version den verschärften Test bestehen kann.
* Um an die Exception zu gelangen, musste eine angepasste
  Version des `CommonContainerStoppingExceptionHandler` implementiert
  werden, die sich die Exception, über die der `KafkaConsumer` gestolpert
  ist, merkt.
* Dabei auch den Health-Endpoint wiederbelebt.
* Seltsamer Weise musste dabei der Code für die AssertJ-Assertions
  angepasst werden, obwohl sich die Logik im Testfall und die Signatur der
  getesteten Methode nicht geändert hat. Vielleicht durch eine Änderung in
  den transitiv angezogenen Abhängigkeiten durch das Einbinden von
  Spring Kafka??

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

Springify: Die Fehlerbehandlung funktioniert unabhängig vom Batch-Listener

* Es ist nicht wie zunächst vermutet nötig, einen Batch-Listener zu
  verwenden, um das gewünschte Stop-World-Verhalten für Deserialisierungs-
  Fehler zu erreichen.
* Den Batch-Listener wieder entfernt - der Test bleibt unverändert GRÜN.
* Der von Spring Kafka automatisch erzeugte Listener-Container verwendet
  dann auch automatisch den als Bean definierten Error-Handler.

Springify: GRÜN - Unerwartetes Verhalten lag an Konfigurations-Fehler

* Der `KafkaMessageListenerContainer` hatte sich nicht wegen einem
  fehlenden `ErrorHandlingDeserializer` trotz des konfigurierten
  `CommonContainerStoppingErrorHandler` in der Endlosschleife verfangen,
  so wie es die Fehlermeldung suggeriert hatte.
* Der Fehler lag eigentlich daran, dass der Error-Handler zwar erzeugt,
  aber nicht der wegen der Batch-Verarbeitung manuell erzeugten
  `KafkaListenerContainerFactory` übergeben wurde.

Springify: ROT - Merge des verschärften Tests aus der Vanilla-Version

* Ohne die Verschärfung des Tests war der Test grün, obwohl die Anwendung
  entgegen den Erwartungen trotz der eingetreuten Poison-Pill _alle_
  Nachrichten gelesen hat.
* Jetzt schlägt der Test wie erwartet fehl.

Springify: GRÜN - `start()`/`stop()` werden im Test explizit aufgerufen

Springify: Die `@PreDestroy`-Methode wird nicht benötigt

* Spring Kafka prüft schon zuvor selbst, ob der Container noch läuft und
  fährt ihn ggf. herunter.

Springify: Start/Stop prüft, ob der Container schon/noch läuft

Springify: `start()`/`stop()`/`destroy()` in EndlessConsumer wiederbelebt

* Die Logik für Start/Stop liegt jetzt wieder wie vor der Umstellung in
  dem EndlessConsumer
* Der `ApplicationRunner` kenn den `EndlessConsumer` und ruft für diese
  `start()` auf
* Entsprechende Aufrufe im `DriverController` wiederbelebt.
* Das Stoppen beim Herunterfahren der App wird wieder über eine
  `@PreDestroy`-Methode im `EndlessConsumer` realisiert.

Springify: ROT - Auto Startup in @KafkaListener deaktiviert

* Ziel, näher an die Funktion des Vanilla-EndlessConsumer heranrücken.
  Der Testfall soll möglichst unverändert funktionieren.
* Unklar, ob das für die Schulung hilfreich ist.
* Hilft aber definitiv beim Verstehen der Mechanismen von Spring Kafka
* Hier wurde jetzt erst mal der automatische Start des Containers
  unterbunden und stattdessen `Application` wieder zu einem
  `ApplicationRunner` mit `@PreDestroy`-Methode gemacht.
* TODO: Testfall ist darüber erst mal ROT, da der Container dort jetzt
  nicht gestartet wird! Das war ja aber genau das Ziel, denn bei den
  übernommenen Testfällen gab es (zumindest theoretisch) Race-Conditions,
  weil der Container - anders als bei der Vanilla-Implementierung -  immer
  schon lief.

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

Springify: BatchListener konfiguriert - Hilft nicht wirklich...

* Unerwartete aber eigentlich logische Folge:
** Es werden weiterhing alle 100 Nachrichten abgeholt
** Der Testfall schlägt trotzdem nicht fehl, da die Erwartung an die
   gespeicherten Offsets dynamisch aus den tatsächlich vom Listener
   bezogenen Nachrichten bestimmt wird.
* Der Effekt tritt ein, weil der `ErrorHandlingDeserializer` ja die
  verursachende Exception schluckt und dafür eine Behandlung des Fehlers
  vornimmt, die den Fehler für die nachfolgenden Komponenten transparenter
  machen soll.
* Es ist aber eher das Gegenteil der Fall, da die Folge ist, dass der
  `poll()`-Aufruf alle Nachrichten - _auch die nach dem Fehler!_ - abholt.
* D.h., es ist nicht mehr so simpel und elegant erkennbar, an welcher
  Stelle die `RecordDeserializationException` aufgetreten ist
* Wenn man - wie hier beabsichtigt - die Konsumption beim Auftreten des
  dieser Sorte Ausnahme-Fehler stoppen und die Offsets für alle _vor_ dem
  Auftreten des Fehlers fehlerfrei empfangenen Nachrichten dafür speichern
  will, dann ist dies ohne eine umständliche Offset-Verwaltung und vielen
  `seek()`-Aufrufen nicht mehr möglich!
* Eine einfache Abhilfe scheint nicht möglich, da der
  `KafkaMessageListenerContainer` nicht damit umgehen kann, wenn _in_ dem
  `poll()` eine Exception geworfen wird und dann - wie beobachtet - in
  einer Endlosschleife hängen bleibt.

Springify: `CommonContainerStoppingErrorHandler` für erwartetes Verhalten

* Der `CommonContainerStoppingErrorHandler` stoppt den Container beim
  ersten Auftreten eines Fehlers.
* Dadurch ist das erwartete Verhalten - soweit bisher durch die Tests
  definiert - wiederhergestellt.
* Der Handler wird den Container aber auch bei einem Fehler im Listener
  stoppen, so dass in dem Fall wahrscheinlich noch nachgebessert werden
  muss.

Springify: `ErrorHandlingDeserializer` bricht die Schleife

* Die Konfiguration eines `ErrorHandlingDeserializer` durchbricht die
  Endlosschleife.
* Das Default-Verhalten von Spring Kafka ist aber dann in dem Fall, den
  Fehler zu loggen und dann zu skippen.
* Dieses Verhalten wird von dem `DefaultErrorHandler` vorgegeben

Springify: Kernfunktion von EndlessConsumer über Spring-Kafka

* Alle weiteren Funktionen für dieses erste Experiment erst mal entfernt
* Testfall entsprechend angepasst
* Der Commit passiert hier, weil Spring Kafka per Default den eignen
  Commit-Modus `BATCH` aktiviert, der nach jedem abgearbeiteten `poll()`
  einen (synchronen!) Commit durchführt
* Der Test ist zwar grün, wenn man die App normal startet, verfängt sie
  sich jedoch in einer Endlosschleife, da kein Error-Handler konfiguriert
  ist, der die `RecordDeserializationException` korrekt behandeln kann, so
  dass sich die App in einem Life-Deadlock befindet, in dem sie immer
  wieder den Datensatz erhält, der den Fehler ausgelöst hat, dadurch
  aber nicht wie ein Vanilla-Consumer beendet wird, da Spring Kafka die
  Exception abfängt und weitermacht: neuer `poll()` für die selbe Position,
  so dass die App aus Sicht des GroupCoordinator noch lebt.
* DEBUG-Logging für `org.springframework.kafka` aktiviert, damit man die
  Commits sieht.

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

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.

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