GRÜN: Fehler in der Test-Logik korrigiert

* Die Assertion, dass nach einem wiederholten Versuch, den Logik-Fehler
  zu konsumieren nicht mehr Nachrichten konsumiert wurden, als für den
  Test generiert wurden ist nicht gültig, da bei einem Logik-Fehler ja
  gerade _kein_ Commit der zuletzt gelesenen Nachrichten erfolgt, da
  dies dazu führt, dass der Offset für Partitionen erhöht wird, für die
  vor dem Eintreten des Fehlers noch nicht alle Nachrichten gelesen
  wurden, wenn nicht explizti Seek's für diese Partitionen durchgeführt
  werden.
* Die Assertion, dass die Offset-Position nach einem Fehler der Offset-
  Position _vor_ der Ausführung der Fachlogik entspricht ist falsch, da
  durchaus Commits durchgeführt werden können, bevor der Fehler auftritt.
  Daher wird jetzt explizit geprüft, dass
** Die Offset-Position für keine Partition größer ist, als der Offset
   der dort zuletzt gesehenen Nachricht.
** UND mindestens eine Partition existiert, deren Offset _kleiner_ ist,
   als der Offset der zuletzt gesehenen Nachricht.

ROT: Fehler in Test-Logik aufgedeckt

* Einige Assertions in dem Test für die Offset-Position nach einem
  Logik-Fehler waren fehlerhaft.
* Dies ist bisher nicht aufgefallen, weil der Test nicht scharf genug
  war: Er hat so wenig Nachrichten gesendet, dass die fehlerhaften
  Assertions nicht aufgefallen sind, weil es nie zu einem Commit gekommen
  ist, bevor der Fehler ausgelöst wurde.
* TODO: Der Test ist wahrscheinlich immer noch in hohem Maße abhängig
  von der Ausführungsgeschwindigkeit auf dem Test-System. Besser wäre
  es, wenn die Verarbeitung künstlich gedrosselt würde, so dass die
  Timing-Annahmen zu den asynchron ablaufenden Operationen nicht auf
  das Testsystem abgestimmt werden müssen.

Methode zu prüfen der Fachlogik in `RecordGenerator` ergänzt und angebunden

Signatur und Handling des `RecordGenerator` vereinfacht/überarbeitet

* Der `RecordGenerator` darf jetzt selbst bestimmen, wie viele Nachrichten
  er erzeugt und wo wieviele Poison-Pills oder Logik-Fehler erzeugt
  werden, wenn der Test dies anfordert.
* Dafür git der `RecordGenerator` jetzt die Anzahl der tatsächlich
  erzeugten Nachrichten zurück, damit die Tests richtig reagieren können.

Anzahl der erzeugten Test-Nachrichten wird vom `RecordGenerator` bestimmt

Namen der Test-Klassen korrigiert

`GenericApplicationTest` überspring Tests, wenn Fehler nicht verfügbar

* Über eine Annotation wird für Tests, die einen bestimmten Fehler-Typ
  benötigen bei dem `RecordGenerator` nachgefragt, ob der Fehler-Typ
  erzeugt werden kann.
* Wenn der Fehler-Typ nicht zur Verfügung steht, wird der Test
  übersprungen.

Tests aus gemerged springified-consumer--serialization -> deserialization

* Es wurde nur der hinzugefügte Test übernommen.
* Der hinzugefügte Test wurde an das von Spring-Kafka abweichende
  Verhalten bei einem Logik-Fehler angepasst: Kafka führt nicht automatisch
  Seeks oder einene Commit durch. Da `EndlessConsumer` bei einem
  Logik-Fehler explizit ein `unsubscribe()` durchführt, wird kein
  Offset-Commit durchgefürt, so dass die alten Offset-Positionen gültig
  bleiben.
* Der Test wurde entsprechend umbenannt.
* `RecordGenerator` wurde um einen weiteren Integer-Set erweitert, über
  den die Indizes der zu erzeugenden Logik-Fehler gesetzt werden können.
* Der hinzugefügte Test wurde auf die überarbeitete Methode zur Erzeugung
  der Test-Nachrichten umgestellt.
* `ApplicationTest` wurde so ergänzt, dass der für den hinzugefügten Test
  benötigte Logik-Fehler erzeugt wird.

Typisierung in `GenericApplicationTest` nur noch, wo wirklich nötig

* Es wird nur noch dort mit Typisierung gearbeitet, wo dies unumgänglich
  ist, weil die typisierte Implementierung angesprochen wird.
* Das Versenden der Test-Nachrichten erfolgt als `Bytes` für Schlüssel
  und Nachricht.
* Dadurch muss der `RecordGenerator` nicht mehr typisiert werden.
* Dafür muss die typisierte Implementierung des Testfalls dann Schlüssel
  und Nachricht mit einem passenden Serializer in eine `Bytes`-Payload
  umwandeln.

`ApplicationTest` auf basis der typisierbaren Basis neu implementiert

Typisierbare Basis-Klasse `GenericApplicationTests` eingeführt

Demonstration in README.sh gepimped

Compose-Setup und README.sh für dieses Beispiel repariert

* Zuvor war in dem Setup noch ein Producer konfiguriert, der Nachrichten
  vom Typ `String` geschrieben hat, so dass der Consumer _sofort_ das
  zeitliche gesegnet hat.
* Im README-Skript wurde nicht darauf gewartet, dass der Consumer
  gemeldet hat, dass er ordentlich gestartet ist, bevor er nach der
  vermeintlichen Konsumption der Poison-Pill wieder neu gestartet wurde.

Test prüft ungültige und unbekannte Nachrichten

Verbesserungen des Testfalls gemerged (Branch 'deserialization')

Verhalten des Testfalls kontrollierbarer gemacht

* Die Awaitility-Aufrufe pollen den zu prüfenden Zustand wenn nicht anders
  angegeben so häufig, wie es die CPU zulässt - also ohne Verzögerung
  zwischen den Überprüfungen.
* Das kann den Rechner temporär so überlasten, dass der erwartete Zustand
  in der abgewarteten Zeit gar nicht eintritt!
* Z.B. aufgetreten, wenn wie hier das Commit-Interval auf 1 Sekunde
  gesetzt ist, das Polling von Awaitility aber noch ungebremst durchgeführt
  wird.
* Um diese Quelle für falsche Fehler auszuschließen, wurde jetzt
  durchgängig ein Poll-Intervall von 1 Sekunde für Awaitility gesetzt.

Testfall überarbeitet

* Abhängigkeit der Testergebnisse von Ausführreihenfolge beseitigt.
* Die Abhängigkeit bestand, da die Offset-Positionen als Zustand die
  Testausführung überdauert haben.
* Daher konnte kein weiterer Test mehr ausgeführt werden, nachdem einmal
  eine Poison-Pill in das Topic geschrieben wurde, über die der
  implementierte Consumer stolpert.
* Um das zu umgehen, werden die Offset-Positionen jetzt nach jedem Test
  auf das Ende der Partitionen verschoben. D.h., wenn in dem Test eine
  Poision-Pill geschrieben wird, über die der implementierte Consumer
  nicht hinweglesen kann, werden die Offests vor der Ausführung des
  nächsten Tests über diese Poision-Pill hinweg gesetzt.
* Dadurch ist wurde ein Fehler / eine Schwäche in der Testlogik aufgedeckt:
  In dem Test für das erfolgreiche Schreiben wurde nur deswegen ein Commit
  ausgeführt, weil zuvor noch kein Commit durchgeführt wurde, so dass der
  Default-Wert für das Commit-Interval immer überschritten war.
* Um das zu umgehen, wurde eine Konfigurations-Option für das Setzen des
  Parameters `auto.commit.interval` eingeführt, so dass im Test
  sichergestellt werden kann, dass auf jeden Fall in dem beobachteten
  Zeitraum ein automatischer Commit ausgelöst wird.
* Außerdem: Weniger verwirrende Ausgabe des Offset-Fortschritts.

Deserialisierung von Nachrichten unterschiedlichen Typs

Refaktorisierungen des Testfalls gemerged (Branch 'deserialization')

Verwendung eines weniger verwirrenden Key in Tests (fachlich irrelevant)

Erzeugung von ProducerRecord in Test refaktorisiert

Merge der überarbeiteten Compose-Konfiguration ('deserialization')

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

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

Merge branch 'deserialization' into springified-consumer--serialization

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.

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.

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

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