Das Herz des Clusters bilden die Broker. Anwendungen, die auf Apache Kafka aufbauen agieren als Producer und/oder Consumer und kommunizieren dafür über Nachrichten mit den Brokern. Producer senden per Push Nachrichten — im Umfeld von Kafka Records genannt — an den Broker. Ein Record besteht aus einem Schlüssel (Key) und der Nachricht selbst (Value). Außerdem können einer Nachricht anwendungsspezifische Header zugeordnet werden, die von dem Cluster nicht ausgewertet werden.

Für die Strukturierung der zu verarbeitenden Daten stellt Kafka das Konzept von Topics bereit. Ein Topic enthält i.d.R. fachlich zusammengehörige Daten. Producer/Consumer können in beliebig viele Topics schreiben bzw. aus beliebig vielen Topics lesen. Außerdem kann ein Topic von beliebig vielen (auch: keinem!) Consumer gelesen werden. Über ACLs kann dabei feingranuliert reguliert werden, welche Producer/Consumer welchen Zugriff auf ein Topic erhalten.

Kafka ermöglicht das Sharding der zu verarbeitenden Daten durch die Aufteilung der Daten eines Topics auf mehere Partitionen, die auf unterschiedlichen Brokern liegen können. Ein Topic hat immer mindestens eine Partition. Das Konzept der Consumer-Group, über das in Kafka die horizontale Skalierbarkeit über eine parallele Verarbeitung der Daten ermöglicht wird, ist eng mit diesem Sharding verknüpft: Die Anzahl der parallel einsetzbaren Instanzen einer Consumer-Group wird durch die Anzahl der Partitionen des verarbeiteten Topics vorgegeben. (Hilfreiche Tipps zu der Wahl der richtigen Anzahl an Partitionen für ein Topic finden sich z.B. einem Blogartikel von Jun Rao.)

Jede Nachricht wird beim Queuing — also vor dem Versand — mit einem Zeitstempel versehen und einer bestimmten Partition des Topics zugeordnet. Dadurch kann sie gezielt an den Broker versendet werden, der für die ausgewählte Partition zuständig ist. Dabei führt die Client-API automatisch Retries durch, wenn Netzwerkfehler auftreten. Die Producer-Anwendung wird asynchron informiert, wenn die Nachricht nach Ablauf eines konfigurierbaren Timeouts nicht zugestellt werden konnte.

Die Client-API kümmert sich im Hintergrund automatisch darum, die entgegengenommenen Nachrichten wenn möglich in Batches zu packen und diese an die zuständigen Broker zu übermitteln (Optimierung des Durchsatzes durch Vermeidung unnötiger Einzel-Requests). Die Batches werden dabei durch den Producer gepackt und ggf. komprimiert. Danach durchlaufen sie unverändert den Zustell-Prozess. D.h., die Batches werden von den Brokern nicht ausgepackt und werden so wie vom Producer gepackt gespeichert und an Consumer zugestellt.

Der für eine Partition zuständige Broker speichert die empfangenen Nachrichten in der Eingangsreihenfolge in einem Log. Nachdem der zuständige Broker den Eingang einer Nachricht bestätig hat, kann diese nicht mehr verändert (oder gelöscht!) werden. Insbesondere kann die Reihenfolge, der angenommen Nachrichten nicht mehr geändert werden.

Jede gespeicherte Nachricht bekommt dabei einen Offset zugeordnet, über den Consumer gezielt zu einer bestimmten Position in dem Datenstrom navigieren können. Der Offset einer Nachricht entspricht der Position in dem Log, in dem der zuständige Broker die Nachrichten der Partition speichert und ist daher erst bekannt, wenn die Nachricht von dem Broker als gespeichert bestätigt wird.

Kafka erzeugt bei Bedarf für jede Partition mehrer Kopien die automatisch aktualisiert werden. Diese Kopien bzw. die Broker, die diese Kopien verwalten, werden Replicas genannt. Dabei wird mit Replicas, oder dem (auf Topic-Ebene konfigurierbaren ) Replication-Factor immer die Gesamtzahl der Kopien einer Partition bezeichnet. D.h., für eine Partition existiert immer mindestens ein Replica. Kafka garantiert, dass alle Replicas einer Partition auf unterschiedlichen Brokern liegen, so dass durch den Ausfall eines Brokers immer nur eine Kopie einer Partition betroffen sein kann.

Für jede Partition übernimmt ein Broker die Rolle des Leaders (in der Grafik mit der Medallie markiert). Die restlichen Replica werden als Follower bezeichnet. Die Follower Fragen den Leader regelmäßig nach neuen Nachrichten (Polling). An diesen Polling-Anfragen erkennt der Leader automatisch, bis zu welcher Offset-Position die Follower die Partition jeweils bereits repliziert haben. Im Fehlerfall führt Kafka automatisch ein Hand-Over durch, bei dem zuerst ein neuer Leader für die Partition bestimmt und anschließend alle Clients auf den neuen Leader umgeleitet werden.

Der Cluster führt protokoll darüber welche Follower die Partition vollständig — d.h., bis zur aktuellen Offset-Position — repliziert haben. Diese Follower bilden zusammen mit dem Leader die Gruppe der In-Sync-Replica (ISR).

Jeder Replica aus dieser Gruppe (in der Grafik die Broker 1, 2 und 4) kann zu jeder Zeit die Rolle des Leaders übernehmen. Wenn der Leader heruntergefahren wird oder ausfällt, kann der Cluster daher unmittelbar einen Follower aus der Gruppe der ISR als Leader auswählen. Dieser Vorgang wird als Leader Election bezeichnet. In der Grafik sind dies die Broker 2 und 4.

Beachte: Im Fehlerfall kommt es trotzdem zu einer Unterbrechung, da der Cluster bestimmte konfigurierbare Timeouts abwarten muss, bevor er einen Broker als ausgefallen einstufen darf.

Die Leader Elections werden von einem besonderen Broker — dem Controller — durchgeführt. Die Rolle des Controllers kann von jedem Broker ausgeübt werden. Der Cluster stellt sicher, dass zu jeder Zeit genau ein Broker diese Rolle übernimmt und führt automatisch einen Fail-Over durch, wenn dieser Broker ausfällt.

Nachrichten können nur über den Leader in eine Partition geschrieben werden. Der Leader stellt dabei sicher, dass die Garantien eingehalten werden, die der Producer fordert (Parameter acks der Producer-Konfiguration). Wenn der Producer fordert, dass Kopien der Nachricht erzeugt wurden, bevor der Empfang der Nachricht bestätigt wird (acks=all), muss der Leader abwarten, bis alle Follower aus der ISR die Nachricht repliziert haben. Außerdem darf die Nachricht bis zu diesem Zeitpunkt noch nicht an die Consumer des Topics weitergeleitet werden. Ansonsten würden die Consumer in einem Fehlerfall ggf. Nachrichten zu sehen bekommen, die aus der Sicht des Producers nie versendet wurden. Die Garantie acks=all bedeutet also, dass Kafka für ein Topic mit einem Replication-Faktor von n + 1 den Ausfall von n Brokern tolerieren kann, ohne dass eine bestätigte Nachricht verloren geht. Zum Vergleich: Verteilte Systeme, die ein klassisches Quorom wie z.B. Raft verwenden, benötigen 2n + 1 Kopien, um den Ausfall von n Instanzen ohne Datenverlust zu überstehen.

Die Hohe Toleranz von Kafka gegenüber Ausfällen hat allerdings ihren Preis. Da der Leader bei der Annahme einer Nachricht auf die Bestätigung durch alle Follower warten muss, bevor er dem Producer den Eingang bestätigen und die Nachricht an wartende Consumer weiterreichen kann, wird die Zustelldauer durch die Verarbeitungsgeschwindigkeit des langsamsten Replicas bestimmt. Um diesen Effekt im Fall von Störungen möglichst gering zu halten, passt Kafka die Gruppe der ISR dynamisch an: Follower, die nicht in der Lage sind, das Log des Leaders innerhalb einer über den Broker-Parameter replica.lag.time.max.ms konfigurierbaren Zeitspanne vollständig zu replizieren, werden aus der Gruppe der ISR entfernt (ISR Shrink). Die Daten der betroffenen Partition werden dann — bezogen auf den über die Konfiguration für das Topic festgelegten Replication Factor — nicht vollständig repliziert: Die Partition ist under-replicated. Sobald die langsamen Follwer wieder aufschließen — d.h., sobald sie wieder in der Lage sind, die beim Leader eingegangenen Nachrichten in dem vorgegebenen Zeitfenster zu replizieren --, werden sie von dem Cluster automatisch wieder in die Gruppe der ISR aufgenommen (ISR Expand). Dadurch ist Kafka in der Lage, dynamisch auf (vorübergehende) Performance-Engpässe einzelner Broker zu reagieren und deren Auswirkungen auf den Datendurchsatz möglichst gering zu halten (Trade-Off zwischen Durability & Availability).

Über den Parameter min.insync.replicas lässt sich pro Topic festlegen, wie viele Kopien einer Nachricht mindestens erzeugt werden müssen, bevor der Leader eine Nachricht bestätigen darf, für die der Producer acks=all fordert. D.h., dass auch in der oben beschriebenen Überlastungssituation mindestens die über min.insync.replicas geforderte Anzahl an Kopien erstellt wird, bevor die Nachricht bestätigt wird. Andererseits bedeutet dies aber auch, dass für die betroffene Partition keine Nachrichten mehr als gespeichert bestätigt und somit auch nicht für das Lesen durch Consumer freigegeben werden können, wenn die geforderten Kopien nicht erstellt werden können. Die Partition steht dann nicht mehr für das Schreiben/Lesen neuer Nachrichten zur Verfügung.

Wenn sich nach einem Ausfall des Leaders keiner der verfügbaren Replica in der Gruppe der ISR befindet, kann die Leader Election nicht automatisch durchgeführt werden. Die Partition ist dann offline, d.h., sie kann weder gelesen werden noch können Nachrichten an die Partition gesendet werden. Wenn dieser Fehlerzustand nicht durch eine Wiederbelebung der ausgefallenen Broker-Instanzen gelöst werden kann muss eine Unclean Leader Election durchgeführt werden. Dabei gehen Nachrichten verloren, da ein Follower zum Leader ernannt wird, der nicht alle Nachrichten repliziert hat die der Leader vor dem Ausfall bestätigt hatte (in der Grafik der Broker 3).

D.h. Nachrichten, die als erfolgreich gesendet bestätigt wurden und die ggf. bereits von einem Consumer gelesen wurden, sind nicht mehr verfügbar. Ein wiederholtes Lesen des Topics, kann also zu unterschiedlichen Ergebnissen führen. Beachte: Die verlorenen Nachrichten können auch später, wenn vielleicht doch einer der ausgefallenen Follower aus der ISR neu gestartet werden kann, nicht wiederhergestellt werden, weil der neue Leader i.d.R. inzwischen andere Nachrichten für diese Offset-Positionen angenommen hat!

Für den Betrieb eines Kafka Clusters in Produktion wird zur Zeit ein Zookeeper Cluster benötigt. Die Broker greifen auf diesen zurück, um den Zustand des verteilten Systems zu überwachen und die für den Betrieb benötigten Verwaltungsdaten ausfallsicher abzulegen. Insbesondere müssen die Broker eine über Heart-Beats kontrollierte Session auf dem Zookeeper-Cluster offen halten. Wenn der Zookeeper-Cluster diese Verbindung zu einem der angemeldeten Broker verliert, wird dieser als ausgefallen betrachtet und der Controller wird dazu veranlasst neue Leader für die Partitionen zu wählen, für die der ausgefallene Broker die Rolle des Leaders übernommen hatte.

Apache Zookeeper ist ein eigenständiges Projekt, das getrennt von Apache Kafka entwickelt wird und als von Kafka benötigte Infrastruktur-Komponente separat betrieben werden muss. Um den damit verbundenen zusätzlichen Administrationsaufwand zu vermeiden, arbeiten die Entwickler von Kafka an einem neuen Betriebsmodus (KRaft), bei dem die Verwaltung der Metadaten und die Überwachung des Clusters von den Brokern selbt übernommen wird, so dass Zookeeper nicht mehr benötigt wird. Das Feature befindet sich jedoch noch in Entwicklung und ist für produktive Systeme noch nicht freigegeben. D.h. für den produktiven Betrieb eines Kafka-Clusters muss bis auf weiteres der Betrieb und das Monitoring eines Zookeeper-Clusters eingeplant werden.

In dieser Serie zeigen wir euch die Umsetzung passwortloser Authenfizierungsverfahren mit Keycloak:

Zu den Themen Kafka, und Monitoring mit Prometheus & Grafana bieten wir die folgenden Schulungen an:

Außerdem bieten wir auf den individuellen Bedarf zugeschnittene Workshops und Schulungen an. Bei Bedarf unterstützen wir Sie auch durch Beratung oder Software-/Architektur-Reviews direkt bei der Entwicklung in Ihrem Projekt. Sprechen Sie uns einfach an.