Wie Kafka zur Realität wurde.

Hallo, Habr!

Ich arbeite im Team von Tinkoff, das für die Entwicklung unseres eigenen Benachrichtigungszentrums zuständig ist. Hauptsächlich programmiere ich in Java mit Spring Boot und löse verschiedene technische Probleme, die im Projekt auftreten.

Die meisten unserer Mikrodienste interagieren asynchron über einen Nachrichtenbroker. Zuvor verwendeten wir IBM MQ als Broker, der mit der Last nicht mehr zurechtkam, aber gleichzeitig hohe Liefersicherheit bot.

Als Ersatz wurde uns Apache Kafka vorgeschlagen, das über ein hohes Skalierungspotenzial verfügt, jedoch leider eine nahezu individuelle Anpassung der Konfiguration für verschiedene Szenarien erfordert. Außerdem erlaubte der Standardmechanismus für „at least once delivery“ in Kafka nicht, das erforderliche Konsistenzniveau sofort zu gewährleisten. Im Folgenden teile ich unsere Erfahrungen mit der Konfiguration von Kafka und erkläre, wie man „exactly once delivery“ einrichtet und damit umgeht.

Garantiezustellung und mehr

Die nachfolgenden Parameter helfen, eine Reihe von Problemen mit den Standardverbindungseinstellungen zu vermeiden. Zunächst möchten wir jedoch einem Parameter besonderes Augenmerk schenken, der das mögliche Debugging erleichtert.

Das wird unterstützt durch client.id für Producer und Consumer. Auf den ersten Blick kann als Wert der Name der Anwendung verwendet werden, was in den meisten Fällen funktioniert. Allerdings führt die Situation, in der mehrere Consumer in der Anwendung verwendet werden und Sie ihnen denselben client.id zuweisen, zu folgender Warnung:

org.apache.kafka.common.utils.AppInfoParser — Fehler bei der Registrierung von AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

Wenn Sie JMX in einer Anwendung mit Kafka verwenden möchten, könnte dies ein Problem darstellen. In diesem Fall ist es am besten, als Wert für client.id eine Kombination aus dem Namen der Anwendung und beispielsweise dem Namen des Topics zu verwenden. Das Ergebnis unserer Konfiguration kann im Output des Befehls kafka-consumer-groups aus den Utilities von Confluent angesehen werden:

Nun betrachten wir das Szenario der garantierten Nachrichtenübermittlung. Der Kafka Producer hat den Parameter acks, der es ermöglicht, einzustellen, nach wie vielen Acknowledgments der Cluster-Leiter die Nachricht als erfolgreich gespeichert betrachtet. Dieser Parameter kann folgende Werte annehmen:

• 0 — Acknowledgments werden nicht gezählt.
• 1 — Standardparameter, Acknowledgment ist nur von 1 Replikat erforderlich.
• −1 — Acknowledgments von allen synchronisierten Replikaten sind erforderlich (Cluster-Konfiguration). min.insync.replicas).

Von den genannten Werten gibt acks gleich −1 die stärksten Garantien, dass die Nachricht nicht verloren geht.

Wie wir alle wissen, sind verteilte Systeme unzuverlässig. Um sich vor vorübergehenden Ausfällen zu schützen, bietet der Kafka Producer den Parameter retries, der die Anzahl der Wiederholungsversuche während delivery.timeout.msfestlegt. Da der Parameter retries den Standardwert Integer.MAX_VALUE (2147483647) hat, kann die Anzahl der Wiederholungen der Nachricht nur durch Anpassung von delivery.timeout.ms geregelt werden.

Gehen wir zur Exactly-once-Zustellung über.

Die genannten Einstellungen ermöglichen es unserem Producer, Nachrichten mit hoher Zuverlässigkeit zu liefern. Lassen Sie uns nun darüber sprechen, wie wir sicherstellen können, dass nur eine Kopie einer Nachricht in ein Kafka-Topic geschrieben wird? Im einfachsten Fall muss der Producer dazu den Parameter enable.idempotence auf den Wert true setzen. Die Idempotenz gewährleistet, dass nur eine Nachricht in eine bestimmte Partition eines Topics geschrieben wird. Voraussetzung für die Aktivierung der Idempotenz sind die Werte acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Wenn diese Parameter vom Entwickler nicht definiert sind, werden die oben genannten Werte automatisch gesetzt.

Sobald die Idempotenz konfiguriert ist, muss sichergestellt werden, dass dieselben Nachrichten immer in dieselben Partitionen gelangen. Dies kann durch das Einrichten des Schlüssels und des Parameters partitioner.class im Producer erreicht werden. Beginnen wir mit dem Schlüssel. Für jede Sendung sollte er gleich sein. Dies lässt sich leicht erreichen, indem man eine beliebige Geschäftskennung aus der ursprünglichen Nachricht verwendet. Der Parameter partitioner.class hat den Standardwert — DefaultPartitioner. Bei dieser Standardpartitionierungsstrategie verhalten wir uns wie folgt:

• Wenn eine Partition bei der Nachrichtensendung explizit angegeben ist, verwenden wir sie.
• Wenn die Partition nicht angegeben ist, aber ein Schlüssel bereitgestellt wird, wählen wir die Partition basierend auf dem Hash des Schlüssels.
• Wenn weder die Partition noch der Schlüssel angegeben sind, wählen wir die Partitionen im Round-Robin-Verfahren aus.

Darüber hinaus ermöglicht die Verwendung eines Schlüssels und idempotenter Sendung mit dem Parameter max.in.flight.requests.per.connection = 1 eine geordnete Verarbeitung der Nachrichten auf dem Consumer. Es ist wichtig zu beachten, dass Sie, falls in Ihrem Cluster ein Zugriffsmanagement eingerichtet ist, die Berechtigungen für idempotente Aufzeichnungen im Topic benötigen.

Falls Ihnen die idempotente Sendung nach Schlüssel nicht ausreicht oder die Logik auf der Seite des Producers die Konsistenz der Daten zwischen verschiedenen Partitionen erfordert, können Transaktionen helfen. Zudem können Sie mit einer Ketten-Transaktion die Aufzeichnung in Kafka beispielsweise mit einer Datenbank-Aufzeichnung bedingt synchronisieren. Damit die transaktionale Sendung beim Producer möglich ist, muss er idempotent sein und zusätzlich transactional.id. Wenn in Ihrem Kafka-Cluster die Zugriffsverwaltung konfiguriert ist, benötigen Sie für die transaktionale Speicherung, wie auch für die idempotente Speicherung, Schreibberechtigungen, die über eine Maske mit dem Wert, der in transactional.id gespeichert ist, erteilt werden können.

Formell kann jede Zeichenfolge, z. B. der Name der Anwendung, als Transaktions-ID verwendet werden. Wenn Sie jedoch mehrere Instanzen derselben Anwendung mit identischem transactional.id ausführen, wird die zuerst gestartete Instanz mit einem Fehler gestoppt, da Kafka sie als Zombie-Prozess betrachtet.

org.apache.kafka.common.errors.ProducerFencedException: Der Producer hat einen Vorgang mit einer alten Epoch versucht. Entweder gibt es einen neueren Producer mit derselben transactionalId, oder die Transaktion des Producers wurde vom Broker abgelaufen.

Um dieses Problem zu lösen, fügen wir dem Anwendungsnamen einen Suffix in Form des Hostnamens hinzu, den wir aus den Umgebungsvariablen erhalten.

Der Producer ist konfiguriert, jedoch steuern Transaktionen in Kafka nur den Geltungsbereich der Nachricht. Unabhängig vom Status der Transaktion gelangt die Nachricht sofort in das Topic, besitzt jedoch zusätzliche systembedingte Attribute.

Damit solche Nachrichten nicht vorzeitig vom Consumer gelesen werden, muss er den Parameter isolation.level auf den Wert read_committed setzen. Ein solcher Consumer kann weiterhin nicht-transaktionale Nachrichten lesen, während er transaktionale Nachrichten erst nach dem Commit lesen kann.
Wenn Sie alle zuvor genannten Einstellungen konfiguriert haben, haben Sie genau einmalige Lieferung eingestellt. Herzlichen Glückwunsch!

Aber es gibt noch einen weiteren Punkt. Die Transactional.id, die wir oben konfiguriert haben, ist tatsächlich ein Transaktionspräfix. An den Transaktionsmanager wird eine fortlaufende Nummer angehängt. Die resultierende ID wird auf transactional.id.expiration.ms, die auf dem Kafka-Cluster konfiguriert ist und standardmäßig auf "7 Tage" gesetzt ist. Wenn innerhalb dieses Zeitraums keine Nachrichten vom Anwender empfangen werden, erhalten Sie bei dem Versuch, die nächste transaktionale Sendung durchzuführen, eine InvalidPidMappingException. Danach gibt der Transaktionskoordinator eine neue fortlaufende Nummer für die nächste Transaktion aus. Dabei kann eine Nachricht verloren gehen, wenn die InvalidPidMappingException nicht ordnungsgemäß behandelt wird.

Statt der Ergebnisse

Wie man feststellen kann, reicht es nicht aus, Nachrichten nur in Kafka zu senden. Es ist notwendig, eine Kombination von Parametern auszuwählen und bereit zu sein, schnelle Änderungen vorzunehmen. In diesem Artikel habe ich versucht, die Konfiguration von Exactly Once Delivery im Detail darzustellen und einige Probleme mit den Konfigurationen client.id und transactional.id zu beschreiben, auf die wir gestoßen sind. Unten sind die Einstellungen für Producer und Consumer in kurzer Form aufgeführt.

Producer:

1. acks = all
2. retries > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 — für die geordnete Sendung)
5. transactional.id = ${application-name}-${hostname}

Consumer:

1. isolation.level = read_committed

Um Fehler in zukünftigen Anwendungen zu minimieren, haben wir ein Wrapper über die Spring-Konfiguration erstellt, in dem bereits Werte für einige der genannten Parameter festgelegt sind.

Hier sind ein paar Materialien für das Selbststudium:

• KIP-98 — Exactly Once Delivery und Transaktionale Nachrichtenübermittlung
• Beschreibung der Einstellungen

Quelle: habr.com