Hallo, Habr.
Vor kurzem habe ich darüber, welche Parameter wir im Team am häufigsten für den Kafka Producer und Consumer verwenden, um eine garantierte Lieferung zu erreichen. In diesem Artikel möchte ich erläutern, wie wir die erneute Verarbeitung von Ereignissen organisiert haben, die wir aus Kafka empfangen, wenn ein externes System vorübergehend nicht erreichbar ist.
Moderne Anwendungen arbeiten in einer sehr komplexen Umgebung. Die Geschäftsanwendungen, eingebettet in moderne Technologiestacks, laufen in Docker-Containern, die von einem Orchestrator wie Kubernetes oder OpenShift verwaltet werden, und kommunizieren mit anderen Anwendungen oder Unternehmenslösungen über eine Kette von physischen und virtuellen Routern. In einer solchen Umgebung kann immer etwas schiefgehen, daher ist die erneute Verarbeitung von Ereignissen im Falle der Nichtverfügbarkeit eines externen Systems ein wichtiger Teil unserer Geschäftsprozesse.
Wie es vor Kafka war
Früher haben wir im Projekt IBM MQ für die asynchrone Zustellung von Nachrichten verwendet. Auftretende Fehler im Service konnten dazu führen, dass die empfangene Nachricht in eine Dead-Letter-Queue (DLQ) gelegt wurde, um sie später manuell zu bearbeiten. Die DLQ wurde neben der eingehenden Warteschlange erstellt, und das Umlegen der Nachricht fand innerhalb von IBM MQ statt.
Wenn der Fehler vorübergehender Natur war und wir dies feststellen konnten (zum Beispiel ResourceAccessException bei HTTP-Aufrufen oder MongoTimeoutException bei Anfragen an MongoDB), trat die Rückrufstrategie in Kraft. Unabhängig von den Verzweigungen in der Logik der Anwendung wurde die ursprüngliche Nachricht entweder in die Systemwarteschlange für verzögerte Zustellungen oder in eine separate Anwendung gelegt, die einst für die erneute Zusendung von Nachrichten entwickelt wurde. Dabei wird der Rücksendungsnummer im Nachrichtenkopf vermerkt, der an das Verzögerungsintervall oder das Ende der Strategie auf Anwendungsebene gebunden ist. Wenn wir das Ende der Strategie erreicht haben, aber das externe System immer noch nicht verfügbar ist, wird die Nachricht in die DLQ zur manuellen Bearbeitung gelegt.
Lösungsfindung
, lässt sich Folgendes finden . Kurz gesagt, es wird vorgeschlagen, für jedes Delay-Intervall ein Topic einzurichten und Consumer auf der Anwendungseite zu implementieren, die die Nachrichten mit der erforderlichen Verzögerung lesen werden.
Trotz der zahlreichen positiven Rückmeldungen erscheint es mir nicht ganz gelungen. Vor allem, weil der Entwickler neben der Umsetzung der Geschäftsanforderungen viel Zeit in die Implementierung des beschriebenen Mechanismus investieren muss.
Darüber hinaus, wenn in einem Kafka-Cluster Zugriffsmanagement aktiviert ist, wird es einige Zeit kosten, Topics einzurichten und die erforderlichen Zugriffsrechte zu gewähren. Hinzu kommt, dass der richtige Wert für retention.ms für jedes der Retry-Topics gefunden werden muss, um sicherzustellen, dass die Nachrichten rechtzeitig erneut gesendet werden und nicht verloren gehen. Die Implementierung und Beantragung von Zugriffsrechten muss für jede bestehende oder neue Dienstleistung wiederholt werden.
Schauen wir uns nun an, 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 zu Spring-Retry, das Abstraktionen zur Verwaltung verschiedener BackOffPolicies bereitstellt. Dies ist ein ziemlich flexibles Werkzeug, aber ein wesentlicher Nachteil ist die Speicherung von Nachrichten zur erneuten Zustellung 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 die erneute Verarbeitung warten. Da dieser Punkt kritisch für unser System ist, haben wir ihn nicht weiter untersucht.
Spring-Kafka selbst bietet mehrere Implementierungen von ContainerAwareErrorHandler, zum Beispiel , mit dem es möglich ist, die Nachricht später zu verarbeiten, ohne den Offset im Falle eines Fehlers zu verschieben. Seit Version 2.3 von Spring-Kafka gibt es die Möglichkeit, eine BackOffPolicy festzulegen.
Dieser Ansatz ermöglicht es verarbeiteten Nachrichten, einen Neustart der Anwendung zu überstehen, jedoch fehlt nach wie vor der DLQ-Mechanismus. Genau diese Variante wählten wir Anfang 2019, optimistisch in der Annahme, dass wir keinen DLQ benötigen würden (wir hatten Glück, und tatsächlich benötigten wir ihn in den ersten Monaten des Betriebs dieser Wiederholungsbehandlungsanwendung nicht). Temporäre Fehler führten zur Ausführung des SeekToCurrentErrorHandler. Andere Fehler wurden im Log protokolliert, führten zu einer Verschiebung des Offsets, und die Verarbeitung setzte mit der nächsten Nachricht fort.
Endgültige Entscheidung
Die Implementierung, die auf SeekToCurrentErrorHandler basiert, hat uns dazu angestoßen, einen eigenen Mechanismus zur Wiederholung des Sendens von Nachrichten zu entwickeln.
Zunächst wollten wir die bestehenden Erfahrungen 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 Retry-Strategie festgelegt ist, zu unterbrechen. Für andere Anwendungen wäre es wünschenswert, einen zentralen Punkt zu haben, der die Umsetzung der Retry-Strategie gewährleistet. Darüber hinaus sollte dieser zentrale Punkt über die Funktionalität einer DLQ für beide Ansätze verfügen.
Die Retry-Strategie selbst sollte in der Anwendung gespeichert werden, die für den Empfang des nächsten Intervalls bei Auftreten eines temporären Fehlers verantwortlich ist.
Anhalten des Consumers für eine Anwendung mit linearer Logik
Bei der Arbeit mit spring-kafka könnte der Code zum Anhalten 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;
}
// zur DLQ
}In dem Beispiel retryAt ist dies der Zeitpunkt, an dem der MessageListenerContainer neu gestartet werden muss, falls er noch aktiv ist. Der Neustart erfolgt in einem separaten Thread, der im TaskScheduler gestartet wird, dessen Implementierung ebenfalls von Spring bereitgestellt wird.
Der Wert retryAt wird folgendermaßen ermittelt:
- Es wird der Wert des Retry-Zählers gesucht.
- Entsprechend dem Wert des Zählers wird das aktuelle Verzögerungsintervall in der Retry-Strategie gesucht. Die Strategie wird in der Anwendung selbst definiert, wir haben das JSON-Format für ihre Speicherung gewählt.
- Das im JSON-Array gefundene Intervall enthält die Anzahl der Sekunden, nach denen die Verarbeitung wiederholt werden muss. Diese Anzahl wird zur aktuellen Zeit hinzugefügt, um den Wert für retryAt zu bilden.
- Wenn das Intervall nicht gefunden wird, ist der Wert retryAt null und die Nachricht wird zur manuellen Analyse in die DLQ gesendet.
Bei diesem Ansatz bleibt nur, die Anzahl der Wiederholungsaufrufe für jede Nachricht zu speichern, die sich derzeit in der Verarbeitung befindet, zum Beispiel im Arbeitsspeicher der Anwendung. Das Speichern des Versuchszählers im Speicher ist für diesen Ansatz nicht kritisch, da eine Anwendung mit linearer Logik die Verarbeitung insgesamt nicht durchführen kann. Im Gegensatz zu spring-retry führt ein Neustart der Anwendung nicht zum Verlust aller Nachrichten zur erneuten Verarbeitung, sondern lediglich zum Neustart der Strategie.
Dieser Ansatz hilft, die Last von externen Systemen zu verringern, die aufgrund einer sehr hohen Auslastung möglicherweise nicht verfügbar sind. Anders ausgedrückt, haben wir neben der erneuten Verarbeitung auch das Muster implementiert. .
In unserem Fall liegt der Fehlerthreshold nur bei 1, und um die Ausfallzeiten des Systems aufgrund vorübergehender Netzwerkunterbrechungen zu minimieren, verwenden wir eine sehr granulare Wiederholungsstrategie mit kurzen Verzögerungsintervallen. Dies könnte nicht für alle Anwendungen der Unternehmensgruppe geeignet sein, daher muss das Verhältnis zwischen dem Fehlerthreshold und der Länge des Intervalls je nach den Besonderheiten des Systems angepasst werden.
Eine separate Anwendung zur Verarbeitung von Nachrichten aus Anwendungen mit nichtdeterministischer Logik
Hier ist ein Beispielcode, der eine Nachricht an eine solche Anwendung (Retryer) sendet, die bei Erreichen der Zeit RETRY_AT erneut an das Thema DESTINATION sendet:
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 übermittelt werden. Der Wert RETRY_AT ist ebenfalls vorhanden, wie es beim Wiederholungsmechanismus durch die Unterbrechung des Consumers der Fall ist. Neben DESTINATION und RETRY_AT übermitteln wir:
- GROUP_ID, nach dem wir Nachrichten für die manuelle Analyse gruppieren und die Suche vereinfachen.
- ORIGINAL_PARTITION, um zu versuchen, denselben Consumer für die Wiederverarbeitung zu verwenden. Dieser Parameter kann null sein, in diesem Fall wird eine neue Partition anhand des Schlüssels record.key() der ursprünglichen Nachricht erhalten.
- Das aktualisierte COUNTER-Wert, 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 wird.
- REASON — der Grund, warum die Verarbeitung der Nachricht unterbrochen wurde.
Der Retryer speichert Nachrichten zur Wiederzusendung und manuellen Analyse in PostgreSQL. Über einen Timer wird eine Aufgabe gestartet, die Nachrichten mit dem erreichten RETRY_AT findet und sie zurück in die ORIGINAL_PARTITION des Ziel-Topics mit dem Schlüssel record.key() sendet.
Nach dem Senden werden die Nachrichten aus PostgreSQL gelöscht. Die manuelle Verarbeitung der Nachrichten erfolgt über eine einfache UI, die über REST API mit dem Retryer interagiert. Zu den Hauptfunktionen gehören die erneute Zustellung oder Löschung von Nachrichten aus der DLQ, die Anzeige von Fehlerinformationen und die Suche nach Nachrichten, zum Beispiel nach dem Fehlernamen.
Da in unseren Clustern die Zugriffssteuerung aktiviert ist, müssen zusätzlich Zugriffsrechte für das Topic, das der Retryer hört, angefordert werden, und der Retryer muss die Berechtigung erhalten, in das DESTINATION Topic zu schreiben. Das ist unpraktisch, aber im Gegensatz zu dem Ansatz mit einem Intervall-Topic erhalten wir eine vollständige DLQ und eine UI zur Verwaltung.
Es gibt Fälle, in denen ein eingehendes Topic von mehreren Consumer-Gruppen gelesen wird, deren Anwendungen unterschiedliche Logik implementieren. Die erneute Verarbeitung einer Nachricht über den Retryer für eine dieser Anwendungen führt zu einem Duplikat bei einer anderen. Um sich davor zu schützen, richten wir ein separates Topic für die erneute Verarbeitung ein. Das eingehende und das Retry-Topic können ohne Einschränkungen von demselben Consumer gelesen werden.
Standardmäßig bietet dieser Ansatz keine Möglichkeit für einen Circuit Breaker, jedoch kann dieser mit Hilfe von oder neu , indem die Aufrufstellen externer Dienste in die entsprechenden Abstraktionen gekapselt werden. Zudem besteht die Möglichkeit, eine Strategie für Muster auszuwählen, was ebenfalls nützlich sein kann. Beispielsweise kann es in Spring Cloud Netflix ein Thread-Pool oder ein Semaphore sein.
Fazit
So entstand eine eigenständige Anwendung, die es ermöglicht, die Verarbeitung einer Nachricht bei vorübergehender Nichterreichbarkeit eines externen Systems zu wiederholen.
Ein wesentlicher Vorteil der Anwendung ist, dass externe Systeme, die im selben Kafka-Cluster arbeiten, sie ohne wesentliche Anpassungen nutzen können! Diese Anwendung benötigt lediglich den Zugang zum Retry-Topic, muss einige Kafka-Header ausfüllen und die Nachricht an den Retryer senden. Es ist keine zusätzliche Infrastruktur notwendig. Um die Anzahl der übermittelten Nachrichten zwischen der Anwendung und dem Retryer zu minimieren, haben wir Anwendungen mit linearer Logik getrennt und die Wiederverarbeitung über das Stoppen des Consumers realisiert.
Quelle: habr.com
