Hallo, Habr.
Vor kurzem habe ich darüber, welche Parameter wir im Team am häufigsten für Kafka Producer und Consumer verwenden, um eine garantierte Lieferung zu erreichen. In diesem Artikel möchte ich erläutern, wie wir die Verarbeitung von Ereignissen organisiert haben, die aus Kafka erhalten wurden, aufgrund der vorübergehenden Nichterreichbarkeit eines externen Systems.
Moderne Anwendungen arbeiten in einer sehr komplexen Umgebung. Die Geschäftslogik, eingehüllt in einen modernen Technologiestack, der in einem Docker-Image läuft, welches von einem Orchestrator wie Kubernetes oder OpenShift verwaltet wird, und das mit anderen Anwendungen oder Unternehmenslösungen über eine Kette physischer und virtueller Router kommuniziert. In einer solchen Umgebung kann immer etwas kaputt gehen, weshalb die Wiederverarbeitung von Ereignissen bei Nichterreichbarkeit eines externen Systems ein wichtiger Teil unserer Geschäftsprozesse ist.
Wie es vor Kafka war
Früher haben wir im Projekt IBM MQ für die asynchrone Zustellung von Nachrichten verwendet. Wenn ein Fehler im Betrieb des Dienstes auftrat, konnte die empfangene Nachricht in eine Dead-Letter-Queue (DLQ) zur weiteren manuellen Analyse verschoben werden. Die DLQ wurde neben der eingehenden Warteschlange erstellt, die Verschiebung der Nachricht fand innerhalb von IBM MQ statt.
Wenn der Fehler vorübergehender Natur war und wir dies bestimmen konnten (zum Beispiel ResourceAccessException bei einem HTTP-Aufruf oder MongoTimeoutException bei einer Anfrage an MongoDb), trat eine Strategie für Wiederholungsaufrufe in Kraft. Unabhängig von der Logik des Anwendungsszenarios wurde die ursprüngliche Nachricht entweder in die systemweite Warteschlange für verzögerte Zustellungen verschoben oder in eine separate Anwendung, die einst zur Wiederzusendung von Nachrichten entwickelt wurde. Dabei wird in den Header der Nachricht die Anzahl der Wiederholungen geschrieben, die an den Verzögerungsintervall oder das Ende der Strategie auf Anwendungsebene gebunden ist. Wenn wir das Ende der Strategie erreicht haben, die externe Systeme jedoch weiterhin nicht erreichbar sind, wird die Nachricht in die DLQ zur manuellen Prüfung verschoben.
Suche nach einer Lösung
, kann Folgendes gefunden werden . Kurz gesagt, es wird vorgeschlagen, für jedes Verzögerungsintervall ein Thema zu erstellen und auf der Anwendungsseite Consumer zu implementieren, die Nachrichten mit der benötigten Verzögerung lesen.
Trotz der vielen positiven Bewertungen scheint es mir nicht ganz gelungen zu sein. Vor allem weil der Entwickler, neben der Umsetzung der Geschäftsanforderungen, viel Zeit mit der Implementierung des beschriebenen Mechanismus verbringen muss.
Darüber hinaus, wenn auf dem Kafka-Cluster die Zugriffsverwaltung aktiviert ist, muss einige Zeit für die Erstellung von Topics und die Gewährleistung der notwendigen Zugriffsrechte auf diese aufgewendet werden. Zusätzlich muss der richtige Parameter retention.ms für jedes der Retry-Topics festgelegt werden, damit die Nachrichten rechtzeitig erneut gesendet werden und nicht verloren gehen. Die Implementierung und Anforderung von Zugängen muss für jeden bestehenden oder neuen Dienst wiederholt werden.
Lassen Sie uns nun betrachten, welche Mechanismen für die erneute Verarbeitung von Nachrichten Spring im Allgemeinen und Spring-Kafka im Besonderen bietet. Spring-Kafka hat eine transitive Abhängigkeit von Spring-Retry, das Abstraktionen für die Verwaltung verschiedener BackOffPolicies bereitstellt. Es ist ein ziemlich flexibles Werkzeug, allerdings ist ein wesentlicher Nachteil die Speicherung von Nachrichten für die erneute Sendung im Arbeitsspeicher der Anwendung. Das bedeutet, dass ein Neustart der Anwendung aufgrund eines Updates oder eines Fehlers während des Betriebs zum Verlust aller Nachrichten führt, die auf eine erneute Verarbeitung warten. Da dieser Punkt für unser System entscheidend ist, haben wir ihn nicht weiter betrachtet.
Spring-Kafka selbst bietet mehrere Implementierungen von ContainerAwareErrorHandler, wie zum Beispiel , mit dem es möglich ist, die Offset im Fehlerfall nicht zu verschieben und die Nachricht später zu verarbeiten. Seit der Version 2.3 von Spring-Kafka gibt es die Möglichkeit, eine BackOffPolicy zu definieren.
Dieser Ansatz ermöglicht es, dass erneut verarbeitete Nachrichten einen Neustart der Anwendung überstehen, jedoch fehlt nach wie vor der Mechanismus für Dead Letter Queues (DLQ). Diesen Ansatz wählten wir Anfang 2019 in der optimistischen Annahme, dass eine DLQ nicht erforderlich sein würde (wir hatten das Glück, dass sie in den ersten Monaten des Betriebs der Anwendung mit diesem Wiederverarbeitungssystem tatsächlich nicht benötigt wurde). Temporäre Fehler führten zur Aktivierung des SeekToCurrentErrorHandler. Andere Fehler wurden in das Protokoll geschrieben, führten zur Verschiebung des Offsets und die Verarbeitung wurde mit der nächsten Nachricht fortgesetzt.
Die endgültige Entscheidung
Die auf SeekToCurrentErrorHandler basierende Implementierung hat uns dazu angeregt, einen eigenen Mechanismus zur erneuten Übermittlung von Nachrichten zu entwickeln.
Zunächst wollten wir die bereits vorhandene Erfahrung nutzen und sie je nach Logik der Anwendung erweitern. Für eine Anwendung mit linearer Logik wäre es optimal, das Lesen neuer Nachrichten für einen kurzen Zeitraum, der im Rahmen der Strategien für erneute Aufrufe festgelegt wurde, zu unterbrechen. Für andere Anwendungen wollte man einen zentralen Punkt haben, der die Durchführung der Wiederholungsstrategien gewährleistet. Darüber hinaus sollte dieser zentrale Punkt über die DLQ-Funktionalität für beide Ansätze verfügen.
Die Wiederholungsstrategie selbst sollte in der Anwendung gespeichert werden, die für den Empfang des nächsten Intervalls bei einem zeitlichen Fehler verantwortlich ist.
Anhalten des Consumers für die Anwendung mit linearer Logik
Beim Arbeiten mit spring-kafka könnte der Code zum Stoppen des Consumers etwa so aussehen:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// an DLQ
}Im Beispiel ist retryAt der Zeitpunkt, zu dem das MessageListenerContainer neu gestartet werden muss, wenn es noch läuft. Der Neustart erfolgt in einem separaten Thread, der im TaskScheduler gestartet wird, dessen Implementierung ebenfalls von spring bereitgestellt wird.
Wir finden den Wert von retryAt auf die folgende Weise:
- Der Wert des Wiederholungszählers wird gesucht.
- Entsprechend dem Wert des Zählers wird das aktuelle Verzögerungsintervall in der Strategie für erneute Aufrufe gesucht. Die Strategie wird in der Anwendung selbst erklärt; für ihre Speicherung haben wir das JSON-Format gewählt.
- Das im JSON-Array gefundene Intervall enthält die Anzahl der Sekunden, nach denen die Verarbeitung wiederholt werden muss. Diese Anzahl von Sekunden wird zur aktuellen Zeit addiert, um den Wert für retryAt zu bilden.
- Wenn das Intervall nicht gefunden wird, ist der Wert von retryAt null und die Nachricht wird zur manuellen Überprüfung in die DLQ gesendet.
Bei diesem Ansatz bleibt nur die Möglichkeit, die Anzahl der Wiederholungsversuche für jede Nachricht, die derzeit verarbeitet wird, beispielsweise im Arbeitsspeicher der Anwendung zu speichern. Das Speichern des Zählerstands im Speicher ist für diesen Ansatz nicht kritisch, da Anwendungen mit linearer Logik die Verarbeitung insgesamt nicht durchführen können. Im Gegensatz zu spring-retry führt ein Neustart der Anwendung nicht zum Verlust aller Nachrichten für die erneute Verarbeitung, sondern einfach zum Neustart der Strategie.
Dieser Ansatz hilft, die Belastung für externe Systeme zu verringern, die aufgrund einer sehr hohen Last möglicherweise nicht verfügbar sind. Anders ausgedrückt, zusätzlich zur erneuten Verarbeitung haben wir die Implementierung des Musters erreicht. .
In unserem Fall liegt der Fehlergrenzwert nur bei 1, und um die Ausfallzeit des Systems aufgrund vorübergehender Netzwerkunterbrechungen zu minimieren, verwenden wir eine sehr granulare Wiederholungsstrategie mit kurzen Verzögerungsintervallen. Dies ist möglicherweise nicht für alle Anwendungen der Unternehmensgruppe geeignet, daher muss das Verhältnis zwischen Fehlergrenzwert und Intervallgröße an die Besonderheiten des Systems angepasst werden.
Eine separate Anwendung zur Verarbeitung von Nachrichten von Anwendungen mit undeterministischer Logik.
Hier ist ein Beispielcode, der eine Nachricht an eine solche Anwendung (Retryer) sendet, die die erneute Sendung an das Thema DESTINATION zur Zeit RETRY_AT durchführen wird:
public void retry(ConsumerRecord record, String retryToTopic,
Instant retryAt, String counter, String groupId, Exception e) {
Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
List arrayOfHeaders =
new ArrayList(Arrays.asList(headers.toArray()));
updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
updateHeader(arrayOfHeaders, ORIGINAL_PARTITION,
() -> Integer.toString(record.partition()).getBytes());
if (nonNull(retryAt)) {
updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
} else {
updateHeader(arrayOfHeaders, REASON,
ExceptionUtils.getStackTrace(e)::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
}
ProducerRecord messageToSend =
new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
kafkaTemplate.send(messageToSend);
} Aus dem Beispiel geht hervor, dass viele Informationen in den Headern übertragen werden. Der Wert RETRY_AT befindet sich ebenfalls wie beim Wiederholungsmechanismus über das Stoppen des Consumers. Neben DESTINATION und RETRY_AT übergeben wir:
- GROUP_ID, nach dem wir Nachrichten für die manuelle Analyse und die Vereinfachung der Suche gruppieren.
- ORIGINAL_PARTITION, um zu versuchen, denselben Consumer für die Wiederverarbeitung beizubehalten. Dieser Parameter kann null sein, in diesem Fall wird eine neue Partition basierend auf dem Schlüssel record.key() der ursprünglichen Nachricht erhalten.
- Aktualisierter Wert COUNTER, um der Strategie der Wiederholungsaufrufe zu folgen.
- SEND_TO — eine Konstante, die angibt, ob die Nachricht zur Wiederverarbeitung gesendet werden soll, wenn RETRY_AT erreicht wird, oder ob sie in die DLQ verschoben werden soll.
- REASON — der Grund, warum die Verarbeitung der Nachricht unterbrochen wurde.
Der Retryer speichert Nachrichten zur Wiederzusendung und manuellen Analyse in PostgreSQL. Ein Timer startet eine Aufgabe, die Nachrichten mit abgelaufenem RETRY_AT findet und sie zurück in die Partition ORIGINAL_PARTITION des Ziels mit dem Schlüssel record.key() sendet.
Nach dem Versenden werden die Nachrichten aus PostgreSQL gelöscht. Die manuelle Analyse der Nachrichten erfolgt in einer einfachen Benutzeroberfläche, die über eine REST-API mit dem Retryer interagiert. Zu den Hauptfunktionen gehören das erneute Senden oder Löschen von Nachrichten aus der DLQ, das Anzeigen von Fehlermeldungen und das Suchen nach Nachrichten, z. B. nach Fehlermeldungen.
Da in unseren Clustern die Zugriffskontrolle aktiviert ist, müssen zusätzlich Zugriffsrechte für das Thema angefordert werden, das der Retryer abhört, und es muss dem Retryer ermöglicht werden, in das DESTINATION-Thema zu schreiben. Das ist unpraktisch, aber im Gegensatz zum Ansatz mit einem Intervall-Thema erhalten wir eine vollständige DLQ und eine Benutzeroberfläche zu ihrer Verwaltung.
Es gibt Fälle, in denen das eingehende Thema von mehreren verschiedenen Consumer-Gruppen gelesen wird, deren Anwendungen unterschiedliche Logik implementieren. Die Wiederverarbeitung von Nachrichten durch den Retryer für eine dieser Anwendungen führt zu Duplikaten in einer anderen. Um dies zu vermeiden, richten wir ein separates Thema für die Wiederverarbeitung ein. Das eingehende und das Retry-Thema kann von demselben Consumer ohne Einschränkungen gelesen werden.
Standardmäßig bietet dieser Ansatz keine Möglichkeit eines Circuit Breakers, allerdings kann diese Funktionalität in die Anwendung integriert werden mit oder dem neuen , indem die Aufrufe externer Dienste in die entsprechenden Abstraktionen eingebettet werden. Darüber hinaus besteht die Möglichkeit, eine Strategie für das -Muster auszuwählen, was ebenfalls nützlich sein kann. Zum Beispiel kann es in spring-cloud-netflix ein Thread-Pool oder ein Semaphore sein.
Ausgabe
Infolgedessen haben wir eine separate Anwendung entwickelt, die es ermöglicht, die Verarbeitung von Nachrichten bei vorübergehender Nichterreichbarkeit eines externen Systems zu wiederholen.
Einer der Hauptvorteile der Anwendung besteht darin, dass externe Systeme, die im selben Kafka-Cluster arbeiten, sie ohne bedeutende Änderungen auf ihrer Seite nutzen können! Diese Anwendung benötigt lediglich Zugriff auf das Retry-Topic, muss einige Kafka-Header ausfüllen und die Nachricht an den Retryer senden. Es ist keine zusätzliche Infrastruktur erforderlich. Um die Anzahl der umgeschichteten Nachrichten von der Anwendung zum Retryer und zurück zu reduzieren, haben wir Anwendungen mit linearer Logik выделили und in ihnen die Wiederverarbeitung über das Stoppen des Consumers umgesetzt.
Quelle: habr.com
