Einheit 4.3: Event-Streaming-Integration

Einführung

Sie verbinden den API-Dienst von TaskBoard mit seinem Hintergrund-Worker. Wenn ein Benutzer eine Aufgabe erstellt, verschiebt oder schließt, darf der API nicht blockieren, wenn er eine E-Mail oder einen Webhook sendet. Stattdessen erzeugt der API ein task-changed-Ereignis zu Kafka und gibt sofort zurück, während ein separates Worker-Dienst diese Ereignisse konsumiert und Benachrichtigungen auslöst. Dies entkoppelt den Anfragepfad von langsamen Nebeneffekten und ermöglicht es Ihnen, den Worker unabhängig zu skalieren.

IONOS Event Streams for Apache Kafka führt Apache Kafka 4.0.0 aus, sodass Ihre vorhandenen Kafka-Client-Bibliotheken unverändert funktionieren. Was sich von einem selbst verwalteten Broker unterscheidet, ist die Verbindung: die Daten-Ebene erfordert eine gegenseitige TLS, nicht SASL oder Klartext, und die Broker-Adressen und Zertifikate stammen von dem Cluster, den Sie in Einheit 2.5 bereitgestellt haben. Diese Einheit beginnt bei der Client-Verbindung und baut dann das Themen-Design, die zuverlässige Lieferung, die Behandlung von Toten-Briefen und die Skalierung von Consumern auf, alles gegen den TaskBoard-Ereignisfluss.

1. Verbindung von Clients mit mTLS

Die IONOS Kafka-DatenEbene ist TLS gesichert mit gegenseitiger Authentifizierung. Beide Seiten authentifizieren sich gegenseitig: Ihr Client überprüft das Broker-Zertifikat gegen die Zertifizierungsstelle des Cluster, und der Broker überprüft Ihr Client-Zertifikat, das die eigene Zertifizierungsstelle des Cluster signiert. Es gibt keinen Plaintext- oder SASL/PLAIN-Listener. Jeder Producer und Consumer stellt ein Client-Zertifikat vor.

Sie rufen die Anmeldeinformationen, die Zertifizierungsstelle, den privaten Client-Schlüssel und das Client-Zertifikat vom Cluster-Zugriffsendpunkt ab. Diese sind die gleichen Werte, die das Cluster über seine Broker-Adressen und den Zugriff auf das API bereitstellt. Jedes Client-Zertifikat ist 365 Tage gültig, sodass die Zertifikatsrotation in Ihrem operativen Runbook vor Ablauf des Jahres belongt.

1.1 Client-Verbindungskonfiguration

Eine selbstverwaltete Kafka-Client-Konfiguration verwendet security.protocol=SSL mit einem Truststore (für die Cluster-Zertifizierungsstelle) und einem Keystore (für das Client-Zertifikat und den Schlüssel). Die IONOS-Dokumentation zeigt die kanonische Client-Eigenschaftendatei: Die IONOS Kafka-Dokumentation zeigt die kanonische Client-Eigenschaftendatei, und der Client verwendet diese Werte, um eine Verbindung zum Cluster herzustellen.

security.protocol=SSL
ssl.truststore.type=PKCS12
ssl.truststore.location=ca-cert.p12
ssl.truststore.password=changeit
ssl.endpoint.identification.algorithm=

ssl.keystore.type=PKCS12
ssl.keystore.location=admin.p12
ssl.keystore.password=adminp12pass

Hinweis: ssl.endpoint.identification.algorithm ist leer. Da das Cluster ein privates Zertifizierungsstellen-CA anstelle von öffentlich signierten Zertifikaten verwendet, ist die Hostname-Überprüfung gegen die Broker-Adressen hier deaktiviert. Der Truststore validiert stattdessen die Zertifikatskette.

1.2 Producer-Verbindung in Python

Die confluent-kafka-Bibliothek akzeptiert die gleichen SSL-Schlüssel. Richten Sie den Client auf die Broker-Adressen aus Ihrer Terraform-Ausgabe aus und liefern Sie die Zertifizierungsstelle, das Zertifikat und den Schlüssel als PEM-Dateien.

from confluent_kafka import Producer

# Broker addresses come from the ionoscloud_kafka_cluster Terraform output.
conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "client.id": "taskboard-api",
}

producer = Producer(conf)

Das Cluster führt drei Broker über die Cluster-Topologie aus, daher listen Sie alle drei in bootstrap.servers für eine Verbindungsselbstheilung auf. Der Client benötigt nur einen erreichbaren Broker, um zu starten, aber das Auflisten aller drei vermeidet einen einzelnen Fehlerpunkt während des Startvorgangs.

2. Thema- und Partition-Design

Das Themen-Design ist der Bereich, in dem IONOS-spezifische Einschränkungen zum ersten Mal relevant werden. Verwenden Sie ein Thema pro Ereignistyp, anstatt ein einzelnes Firehose-Thema, da Consumer die Ereignisse abonnieren, die sie interessieren, und Sie Aufbewahrung und Partitionierung pro Anliegen festlegen können. Für TaskBoard erstellen Sie ein task-changed-Thema für den Worker und später ein task-changed.dlq-Dead-Letter-Thema.

2.1 Erstellen von Themen über das API

Das Kafka-Management-API ist regional und getrennt von der Cloud-API. Der Host folgt dem Muster https://kafka.<region>.ionos.com, und im Gegensatz zur Daten-Ebene authentifiziert das Management-API mit einem Bearer-Token und nicht mit mTLS. Erstellen Sie ein Thema mit POST /clusters/{clusterId}/topics:

curl -X POST \
  'https://kafka.de-txl.ionos.com/clusters/e69b22a5-8fee-56b1-b6fb-4a07e4205ead/topics' \
  --header 'Content-Type: application/json' \
  --header 'Authorization: Bearer '"$IONOS_TOKEN" \
  --data '{
    "properties": {
      "name": "task-changed",
      "replicationFactor": 3,
      "numberOfPartitions": 6,
      "logRetention": { "retentionTime": 604800000 }
    }
  }'

Die Body-Parameter sind name (das einzige obligatorische Feld), replicationFactor, numberOfPartitions und die logRetention-Einstellungen. Lesen Sie Themen mit GET /clusters/{clusterId}/topics und GET /clusters/{clusterId}/topics/{topicId}, die name, replicationFactor, numberOfPartitions und logRetention zurückgeben. Die Löschung ist DELETE /clusters/{clusterId}/topics/{topicId} und gibt 202 Accepted zurück, da die Entfernung asynchron ist.

2.2 Partition-Anzahl und Replikationsbeschränkungen

Wählen Sie die Anzahl der Partitionen als Vielfaches von 3 (3, 6, 9, 12 und so weiter), um eine ungleichmäßige Verteilung der Partitionen über die drei Broker zu vermeiden. Eine Partition-Anzahl, die kein Vielfaches der Broker-Anzahl ist, lässt einen Broker mehr Partitionen tragen als die anderen, was die Last verzerrt.

Der empfohlene Replikationsfaktor ist 3, der der dreibroker-Topologie entspricht, aber dies ist eine IONOS-Empfehlung und kein systemweit vorgegebener Standard; Sie müssen replicationFactor explizit angeben, wenn Sie ein Thema erstellen. Mit Replikationsfaktor 3 behält das Cluster drei Kopien jeder Nachricht, sodass das Thema den Verlust eines Brokers ohne Datenverlust übersteht. Die Standard-Thema-Aufbewahrungsfrist beträgt 604800000 ms (7 Tage). Wenn Sie retentionTime auf -1 setzen, wird keine Zeitbegrenzung angewendet, in diesem Fall ist die Aufbewahrungsfrist nur durch den von Ihnen bereitgestellten Cluster-Speicherplatz begrenzt (z. B. 750 GB auf der Größe S, 1200 GB auf der Größe M).

Die folgende Tabelle ordnet die Body-Parameter eines Themas ihren Rolle zu, wenn Sie ein Thema entwerfen.

Partitionen sind wichtig, über den Speicher hinaus. Die Partition-Anzahl ist die feste Obergrenze für die Consumer-Parallelität innerhalb einer Consumer-Gruppe, daher sollten Sie sie für den Spitzen-Durchsatz jetzt dimensionieren, da das Erhöhen der Partitionen später die Zuordnung von Schlüssel zu Partition und die Garantien für die Reihenfolge stört.

3. Produzentenzuverlässigkeit

Ein Fire-and-Forget-Produzent verliert Nachrichten bei Broker-Störungen. Für TaskBoards Task-Ereignisse, bei denen ein verlorene Ereignis eine verpasste Benachrichtigung bedeutet, konfigurieren Sie den Produzenten für die Zuverlässigkeit, bevor Sie ihn für die Durchsatzrate optimieren.

3.1 Acks, Idempotenz und Schlüssel

Setzen Sie acks=all so, dass der Produzent auf die Bestätigung aller synchronen Repliken wartet, bevor er einen Schreibvorgang als erfolgreich betrachtet. In Kombination mit einem Replikationsfaktor von 3 bedeutet dies, dass eine Nachricht über die Broker hinweg dauerhaft ist, bevor Ihr Code fortfährt. Aktivieren Sie den idempotenten Produzenten, damit Wiederholungen keine Duplikate auf der Broker-Seite erstellen.

from confluent_kafka import Producer
import json

conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "acks": "all",
    "enable.idempotence": True,
    "retries": 5,
    "linger.ms": 20,
    "batch.size": 65536,
}
producer = Producer(conf)

def on_delivery(err, msg):
    if err:
        # Persist for replay; never silently drop.
        print(f"delivery failed: {err}")
    else:
        print(f"delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}")

def emit_task_changed(task_id: str, event: dict):
    producer.produce(
        topic="task-changed",
        key=task_id.encode("utf-8"),   # same task -> same partition -> ordered
        value=json.dumps(event).encode("utf-8"),
        on_delivery=on_delivery,
    )
    producer.poll(0)

Schlüsselvergabe durch task_id leitet alle Ereignisse für eine Aufgabe zur gleichen Partition weiter, wodurch die Aufgabenreihenfolge erhalten bleibt. Ohne einen Schlüssel verteilt Kafka die Datensätze im Round-Robin-Verfahren und ein "Aufgabe geschlossen"-Ereignis könnte vor dem "Aufgabe erstellt"-Ereignis verarbeitet werden.

3.2 Batchverarbeitung und Flushing

linger.ms und batch.size opfern einige Millisekunden Latenz für einen wesentlich höheren Durchsatz, indem sie Datensätze pro Partition batchen. Es sollte immer flush() vor dem Prozessbeenden ausgeführt werden, sonst gehen die gepufferten Datensätze verloren.

# At shutdown, block until all buffered messages are delivered or fail.
remaining = producer.flush(timeout=10)
if remaining > 0:
    raise RuntimeError(f"{remaining} messages not delivered before shutdown")

4. Consumer-Gruppen, Offsets und Liefersemantik

Der TaskBoard-Arbeiter liest task-changed als Consumer-Gruppe. Jeder Consumer in der gleichen Gruppe teilt die Partitionen, sodass das Hinzufügen eines Worker-Pods die Durchsatzrate bis zur Partitionenzahl skalieren kann. Ein vierter Consumer in einem 3-Partitionen-Topic bleibt inaktiv.

4.1 Mindestens-Einmal mit manuellen Commits

Für die Benachrichtigungslieferung ist Mindestens-Einmal das richtige Standardverhalten: die Nachricht verarbeiten, dann den Offset committen. Wenn der Arbeiter nach der Verarbeitung, aber vor dem Committen abstürzt, wird die Nachricht erneut geliefert und die Benachrichtigung kann zweimal ausgelöst werden, was Sie mit einem Idempotenzschlüssel sicher machen können. Das Gegenteil, das Committen vor der Verarbeitung (Höchstens-Einmal), riskiert den Verlust von Ereignissen bei einem Absturz.

from confluent_kafka import Consumer, KafkaException
import json

conf = {
    "bootstrap.servers": "192.168.1.101:9093,192.168.1.102:9093,192.168.1.103:9093",
    "security.protocol": "ssl",
    "ssl.ca.location": "ca-cert.pem",
    "ssl.certificate.location": "client-cert.pem",
    "ssl.key.location": "client-key.pem",
    "ssl.endpoint.identification.algorithm": "none",
    "group.id": "taskboard-notifier",
    "enable.auto.commit": False,        # manual commit = control over semantics
    "auto.offset.reset": "earliest",
}
consumer = Consumer(conf)
consumer.subscribe(["task-changed"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        raise KafkaException(msg.error())
    event = json.loads(msg.value())
    send_notification(event)            # do the work first
    consumer.commit(msg)                # then commit only this offset

enable.auto.commit=False ist die Schlüsselwahl. Auto-Commit setzt die Offsets auf einem Timer unabhängig davon vor, ob die Verarbeitung erfolgreich war, was Nachrichten, deren Verarbeitung fehlgeschlagen ist, stillschweigend verwirft.

4.2 Consumer-Skalierung und Neubalancierung

Die Partitionenanzahl ist die Obergrenze der Parallelität. Um den TaskBoard-Arbeiter auf Managed Kubernetes (siehe Einheit 3.2) zu skalieren, erhöhen Sie die Anzahl der Deployment-Replikate bis zur Partitionenanzahl, und jeder neue Pod joinet die taskboard-notifier-Gruppe und erhält eine Teilmenge der Partitionen durch eine Neubalancierung. Während einer Neubalancierung werden Partitionen kurzzeitig nicht mehr konsumiert, daher sollten Sie die Verarbeitung idempotent halten und Commits häufig durchführen, um eine erneute Verarbeitung zu minimieren.

Bewerten Sie die Partitionen für Ihre maximale Consumeranzahl im Voraus. Sie können später Partitionen hinzufügen, aber dies führt zu einer Neuzuordnung der Schlüssel-zu-Partition-Zuordnung und bricht die Reihenfolge für in-flight-Schlüssel.

5. Dead-Letter-Verarbeitung und Schema-Discipline

Nicht jede Nachricht kann verarbeitet werden. Eine fehlerhafte Nutzlast oder ein Ausfall eines nachgelagerten Dienstes führt dazu, dass die Nachricht bei erneuter Zustellung weiterhin fehlschlägt, und eine Gift-Nachricht, die Sie ewig wiederholen, blockiert die gesamte Partition.

5.1 Dead-Letter-Topic-Muster

Leiten Sie Nachrichten, die ihre Wiederholungsversuche erschöpft haben, in ein dediziertes Dead-Letter-Topic um, task-changed.dlq, und committen Sie dann den ursprünglichen Offset, damit die Hauptpartition weiterhin durchläuft. Sie können den DLQ später überprüfen, die Ursache beheben und gegebenenfalls erneut abspielen.

from confluent_kafka import Producer, Consumer
import json

dlq = Producer(conf_producer)  # same mTLS settings as the main producer

def process_with_dlq(consumer, msg, max_attempts=3):
    attempts = int(dict(msg.headers() or {}).get("x-attempts", b"0") or 0)
    try:
        send_notification(json.loads(msg.value()))
        consumer.commit(msg)
    except Exception as exc:
        if attempts + 1 >= max_attempts:
            dlq.produce(
                "task-changed.dlq",
                key=msg.key(),
                value=msg.value(),
                headers={"x-error": str(exc), "x-attempts": str(attempts + 1)},
            )
            dlq.flush(5)
            consumer.commit(msg)        # advance past the poison message
        else:
            raise                        # let it redeliver for a transient error

Unterscheiden Sie zwischen transienten Fehlern (einer Zeitüberschreitung beim Benachrichtigungsdienst, die eine Wiederholung verdient) und permanenten Fehlern (einer fehlerhaften Ereignis-Nachricht, die direkt in den DLQ gelangt). Ein blindes Wiederholen permanenter Fehler blockiert die Partition, während ein blindes Dead-Lettern transienter Fehler zu verlorenen, wiederherstellbaren Ereignissen führt.

5.2 Ereignis-Verträge mit JSON-Schema

Produzenten und Konsumenten werden unabhängig voneinander bereitgestellt, sodass die Ereignis-Nutzlast ein Vertrag zwischen ihnen ist. Definieren Sie diesen mit JSON-Schema und validieren Sie ihn auf beiden Seiten. Evoluieren Sie das Schema additiv: Fügen Sie optional Felder hinzu, entfernen oder umbenennen Sie bestehende Felder nie, damit alte Konsumenten weiterhin funktionieren, während neue Produzenten ausgeliefert werden.

import jsonschema

TASK_CHANGED_V1 = {
    "type": "object",
    "required": ["task_id", "action", "occurred_at"],
    "properties": {
        "task_id": {"type": "string"},
        "action": {"enum": ["created", "moved", "closed"]},
        "occurred_at": {"type": "string", "format": "date-time"},
        "assignee": {"type": "string"},   # added in v1.1, optional = safe
    },
    "additionalProperties": False,
}

def validate_event(event: dict):
    jsonschema.validate(event, TASK_CHANGED_V1)  # raises ValidationError on breach

Übertragen Sie eine Schemaversion in der Ereignis-Nachricht oder einem Header, damit Konsumenten während eines Migrationszeitraums darauf basierend eine Verzweigung durchführen können. Diese Disziplin ermöglicht es Ihnen, die Ereignis-Form von TaskBoard zu ändern, ohne eine koordinierte, schrittweise Neu-Bereitstellung von Produzent und Arbeitsprozess durchzuführen.

API Referenz-Quick-Card

Wichtige Endpunkte für die Kafka-Verwaltungs-API. Der Host ist regional, zum Beispiel https://kafka.de-txl.ionos.com:

Verwaltungs-API-Authentifizierung: Authorization: Bearer <token> Datenflugzeug-Authentifizierung (Produzenten/Verbraucher): gegenseitige TLS mit Clientzertifikat

Code-Lab

Ziel: Erzeugen und Verarbeiten von TaskBoard task-changed Ereignissen auf einem IONOS Kafka Cluster mit einer Consumer-Gruppe, manueller Offset-Bestätigung und Dead-Letter-Routing im Fehlerfall.

Voraussetzungen:

• IONOS Cloud-Konto mit API Token (IONOS_TOKEN)
• Ein bereitgestellter Kafka Cluster aus Einheit 2.5 mit verfügbarer Broker-Adresse
• Python 3.10+ mit confluent-kafka und jsonschema installiert
• Die Cluster-Zertifizierungsstelle, Client-Zertifikat und Client-Schlüssel als PEM-Dateien gespeichert

Schritt 1: Abrufen von Broker-Adressen und Zugriffsdaten

CLUSTER_ID="e69b22a5-8fee-56b1-b6fb-4a07e4205ead"
curl -s -X GET \
  "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID" \
  -H "Authorization: Bearer $IONOS_TOKEN" | python3 -m json.tool

Das Erwartete Ergebnis:

{
  "id": "e69b22a5-8fee-56b1-b6fb-4a07e4205ead",
  "properties": { "name": "...", "version": "4.0.0", "size": "S",
    "connections": [ { "brokerAddresses": ["192.168.1.101/24", ...] } ] }
}

Schritt 2: Erstellen der Haupt- und Dead-Letter-Themen

for T in task-changed task-changed.dlq; do
  curl -s -X POST "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics" \
    -H "Content-Type: application/json" -H "Authorization: Bearer $IONOS_TOKEN" \
    --data "{\"properties\":{\"name\":\"$T\",\"replicationFactor\":3,\"numberOfPartitions\":6}}"
done

Das erwartete Ergebnis:

{"id":"ae085c4c-...","type":"topic","properties":{"name":"task-changed",
 "replicationFactor":3,"numberOfPartitions":6}}

Schritt 3: Erzeugen eines Ereignisses für eine geänderte Aufgabe

producer.produce("task-changed", key=b"task-42",
    value=b'{"task_id":"task-42","action":"created","occurred_at":"2026-06-05T10:00:00Z"}')
producer.flush(10)
print("produced")

Die Erwartete Ausgabe:

delivered to task-changed [3] @ 0
produced

Schritt 4: Verbrauchen mit einer Consumer-Gruppe und manuellem Commit

Um mit einer Consumer-Gruppe und manuellem Commit zu verbrauchen, müssen Sie die folgenden Schritte ausführen. Zunächst müssen Sie eine Consumer-Gruppe erstellen und dann die Nachrichten manuell committen. Dieser Prozess ist wichtig, um die Nachrichten korrekt zu verarbeiten und um sicherzustellen, dass die Daten nicht verloren gehen.

Die Verwendung einer Consumer-Gruppe bietet viele Vorteile, wie zum Beispiel die Skalierbarkeit und die Elastizität. Durch die manuelle Commit-Funktion können Sie die Kontrolle über den Verbrauch der Nachrichten behalten und sicherstellen, dass die Daten korrekt verarbeitet werden.

Ein Beispiel für die Verwendung einer Consumer-Gruppe und manuellem Commit ist die Verarbeitung von Daten in einem verteilten System. In diesem Fall kann die Consumer-Gruppe verwendet werden, um die Daten von verschiedenen Quellen zu sammeln und zu verarbeiten, und die manuelle Commit-Funktion kann verwendet werden, um sicherzustellen, dass die Daten korrekt gespeichert werden.

Um die Consumer-Gruppe und das manuelle Commit zu verwenden, müssen Sie die folgenden Schritte ausführen:

• Erstellen Sie eine Consumer-Gruppe
• Konfigurieren Sie die Consumer-Gruppe für die manuelle Commit-Funktion
• Verbrauchen Sie die Nachrichten mit der Consumer-Gruppe
• Committen Sie die Nachrichten manuell

Durch die Verwendung einer Consumer-Gruppe und manuellem Commit können Sie die Verarbeitung von Daten in Ihrem System verbessern und sicherstellen, dass die Daten korrekt verarbeitet werden. Dies ist besonders wichtig in Systemen, in denen die Datenverarbeitung kritisch ist, wie zum Beispiel in geschäftskritischen Datenbanken.

Wichtige Punkte:

• Die Consumer-Gruppe bietet Skalierbarkeit und Elastizität
• Die manuelle Commit-Funktion bietet Kontrolle über den Verbrauch der Nachrichten
• Die Verwendung einer Consumer-Gruppe und manuellem Commit ist besonders wichtig in Systemen, in denen die Datenverarbeitung kritisch ist

consumer.subscribe(["task-changed"])
msg = consumer.poll(5.0)
print(msg.topic(), msg.partition(), msg.offset(), msg.value())
consumer.commit(msg)

Die Erwartete Ausgabe:

task-changed 3 0 b'{"task_id":"task-42",...}'

Schritt 5: Auslösen eines Dead-Letter bei einem fehlerhaften Payload

Um einen Dead-Letter auf einem fehlerhaften Payload auszulösen, müssen Sie die folgenden Schritte befolgen. Beachten Sie, dass dies ein wichtiger Schritt im Prozess der Fehlerbehandlung in der Cloud-Computing-Umgebung ist. Ein Dead-Letter ist ein Mechanismus, der es ermöglicht, Nachrichten, die nicht verarbeitet werden können, in einer separaten Warteschlange zu speichern, um eine spätere Analyse und Behandlung zu ermöglichen.

Die Auslösung eines Dead-Letter bei einem fehlerhaften Payload ist besonders wichtig, um die Hochverfügbarkeit und die Skalierbarkeit der Systeme zu gewährleisten. Durch die Verwendung von Dead-Letters können Sie sicherstellen, dass fehlerhafte Nachrichten nicht den normalen Betrieb des Systems stören und dass sie stattdessen in einer separaten Umgebung zur weiteren Analyse und Behandlung gespeichert werden.

Um dies zu erreichen, müssen Sie die entsprechenden Einstellungen in Ihrem System vornehmen. Dies kann je nach verwendetem System und den spezifischen Anforderungen Ihrer Anwendung variieren. Es ist wichtig, dass Sie die Dokumentation Ihres Systems sorgfältig lesen und die empfohlenen Schritte befolgen, um die Auslösung von Dead-Letters bei fehlerhaften Payloads zu konfigurieren.

Durch die richtige Konfiguration und Verwendung von Dead-Letters können Sie die Zuverlässigkeit und Effizienz Ihrer Systeme in der Cloud-Computing-Umgebung erheblich verbessern. Es ist auch wichtig, dass Sie die Voraussetzungen und die empfohlene Vorgehensweise sorgfältig prüfen, bevor Sie mit der Konfiguration beginnen.

producer.produce("task-changed", key=b"task-99", value=b'{not valid json')
producer.flush(10)
# Consumer's process_with_dlq routes it to task-changed.dlq after max_attempts

Das erwartete Ergebnis:

delivered to task-changed.dlq [1] @ 0

Schritt 6: Überprüfen Sie, ob das Dead-Letter-Topic die Nachricht erhalten hat

dlq_consumer.subscribe(["task-changed.dlq"])
m = dlq_consumer.poll(5.0)
print(dict(m.headers()), m.value())

Die Erwartete Ausgabe:

{'x-error': b'...', 'x-attempts': b'3'} b'{not valid json'

Validierungs-Checkliste:

• [ ] task-changed und task-changed.dlq Themen mit Replikationsfaktor 3 erstellt
• [ ] Produzent liefert mit acks=all und Idempotenzen aktiviert
• [ ] Consumer bestätigt Offsets manuell nach der Verarbeitung
• [ ] Eine fehlerhafte Nachricht landet im Dead-Letter-Topic, nicht in einer unendlichen Wiederholungsschleife

Aufräumen:

for ID in $(curl -s "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics" \
  -H "Authorization: Bearer $IONOS_TOKEN" | python3 -c \
  "import sys,json;[print(t['id']) for t in json.load(sys.stdin)['items']]"); do
  curl -s -X DELETE "https://kafka.de-txl.ionos.com/clusters/$CLUSTER_ID/topics/$ID" \
    -H "Authorization: Bearer $IONOS_TOKEN"
done

Häufige Fallen

1. Verwendung von SASL oder Plaintext anstelle von mTLS

   • Problem: Der Client hängt beim Verbinden oder schlägt mit SSL handshake failed fehl, obwohl die Anmeldedaten korrekt erscheinen.
   • Warum es passiert: Die IONOS-DatenEbene akzeptiert nur gegenseitige TLS. Es gibt keinen SASL/PLAIN- oder Plaintext-Listener, und die Broker verwenden eine private Zertifizierungsstelle, so dass ein Standard-Truststore sie ablehnt.
   • Lösung: Setzen Sie security.protocol=ssl, liefern Sie die Cluster-Zertifizierungsstelle als ssl.ca.location, das Clientzertifikat und den Schlüssel, und deaktivieren Sie die Überprüfung des Hostnamens mit ssl.endpoint.identification.algorithm=none.

2. Auto-Commit wirft fehlgeschlagene Nachrichten stillschweigend weg

   • Problem: Einige Aufgabenbenachrichtigungen werden nie ausgelöst, aber keine Fehler erscheinen in den Protokollen und die Consumer-Lücke ist Null.
   • Warum es passiert: Mit enable.auto.commit=True, rücken die Offsets aufgrund eines Timers vor, unabhängig davon, ob send_notification erfolgreich war, so dass eine Nachricht, die eine Ausnahme ausgelöst hat, als verbraucht markiert wird.
   • Lösung: Setzen Sie enable.auto.commit=False und rufen Sie consumer.commit(msg) nur nach erfolgreichem Verarbeiten auf, leiten Sie dauerhafte Fehler zum Toten-Brief-Topic um.

3. Partitionenzahl ist kein Vielfaches der Brokeranzahl

   • Problem: Ein Broker zeigt eine höhere CPU- und Festplattenlast als die anderen beiden, und die Durchsatzleistung erreicht ein Plateau unter den Erwartungen.
   • Warum es passiert: Mit 3 Brokern verteilt sich ein Thema, das mit 4 oder 5 Partitionen erstellt wurde, ungleichmäßig, so dass ein Broker einen zusätzlichen Partitionsschlüssel hostet.
   • Lösung: Erstellen Sie Themen mit einer Partitionenzahl, die ein Vielfaches von 3 ist (3, 6, 9, 12), so dass sich die Partitionen gleichmäßig über die drei Broker verteilen.

Zusammenfassung

Sie können jetzt IONOS Event Streams for Apache Kafka in den Anwendungscodes als ersten Teil der TaskBoard-Architektur integrieren. Sie verbinden Produzenten und Konsumenten über gegenseitige TLS mithilfe von Broker-Adressen und Zertifikaten aus Ihrem bereitgestellten Cluster, erstellen Themen über die regionale Verwaltung API mit Bearer-Authentifizierung und entwerfen Partitionierungs- und Replikationseinstellungen, die die drei-Broker-Topologie respektieren. Darüber hinaus haben Sie einen zuverlässigen Produzenten mit Idempotenz und acks=all, eine Konsumentengruppe mit manueller mindestens-einmal-Commit, ein Dead-Letter-Theme für giftige Nachrichten und JSON-Schema-Verträge, die es Produzenten und Konsumenten ermöglichen, unabhängig voneinander zu evolvieren.

Wichtige Punkte:

• Die Kafka-Daten-Ebene erfordert gegenseitige TLS mit einem Client-Zertifikat; die Verwaltung API verwendet Bearer-Tokens. Es handelt sich um zwei verschiedene Authentifizierungsmodelle auf zwei verschiedenen Endpunkten.
• Die Anzahl der Themen-Partitionen sollte ein Vielfaches von 3 sein, um sie gleichmäßig über die drei Broker zu verteilen; der empfohlene Replikationsfaktor beträgt 3, aber Sie müssen ihn explizit festlegen, da er nicht standardmäßig durch das System vorgegeben ist.
• Die mindestens-einmalige Lieferung ergibt sich aus der Deaktivierung des Auto-Commits und der Commit-Anweisung der Offsets nur nach erfolgreicher Verarbeitung; machen Sie die Verarbeitung idempotent, um die erneute Lieferung zu tolerieren.
• Die Partitionierungsanzahl ist die harte Obergrenze für die Konsumenten-Parallelität; dimensionieren Sie die Partitionen für die Spitzenlast, bevor Sie bereitstellen, da das Hinzufügen später die Schlüsselordnung stört.
• Dead-Letter-Themen isolieren giftige Nachrichten, sodass eine schlechte Nutzlast nicht ihre Partition blockiert, und JSON-Schema-Verträge erzwingen den Produzenten-zu-Konsumenten-Vertrag.

Wichtige Begriffe:

• mTLS (gegenseitige TLS): Sowohl Client als auch Broker authentifizieren sich gegenseitig mit Zertifikaten; die einzige Authentifizierungsmethode der IONOS-Daten-Ebene, wobei Client-Zertifikate 365 Tage gültig sind.
• Konsumentengruppe: Eine Gruppe von Konsumenten, die die Arbeit der Partitionen eines Themas teilen, die Einheit der horizontalen Skalierung für den TaskBoard-Worker.
• Offset-Commit: Die Aufzeichnung der Position, die ein Konsument verarbeitet hat; das Commit nach der Verarbeitung ergibt die mindestens-einmalige Lieferung.
• Dead-Letter-Theme (DLQ): Ein separates Thema für Nachrichten, die nach Wiederholungen nicht verarbeitet werden können, um die Hauptpartition unblockiert zu halten.
• Replikationsfaktor: Die Anzahl der Kopien jeder Partition über die Broker; 3 bei IONOS, entsprechend der drei-Broker-Cluster, sodass das Thema einen Broker-Verlust überlebt.

Nächste Schritte

Weiterlernen: Einheit 4.4: AI Model Hub Integration

Verwandte Themen:

• Einheit 2.5: Bereitstellung von Datenbank- und Streaming-Diensten
• Einheit 3.2: Kubernetes Bereitstellung und Betrieb