{"id":56186,"date":"2020-02-06T00:00:00","date_gmt":"2020-02-05T21:00:00","guid":{"rendered":"https:\/\/prohoster.info\/blog\/blog_prohoster\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka"},"modified":"2020-02-18T14:04:24","modified_gmt":"2020-02-18T11:04:24","slug":"povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","status":"publish","type":"post","link":"https:\/\/prohoster.info\/de\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Ereigniswiederverarbeitung aus Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Ereigniswiederverarbeitung<\/p>\n

<\/p>\n

Hallo, Habr.<\/p>\n

<\/p>\n

Vor kurzem habe ich meine Erfahrungen geteilt<\/a><\/noindex> dar\u00fcber, welche Parameter wir im Team am h\u00e4ufigsten f\u00fcr den Kafka Producer und Consumer verwenden, um eine garantierte Lieferung zu erreichen. In diesem Artikel m\u00f6chte ich erl\u00e4utern, wie wir die erneute Verarbeitung von Ereignissen organisiert haben, die wir aus Kafka empfangen, wenn ein externes System vor\u00fcbergehend nicht erreichbar ist.<\/p>\n

<\/p>\n

Moderne Anwendungen arbeiten in einer sehr komplexen Umgebung. Die Gesch\u00e4ftsanwendungen, 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\u00f6sungen \u00fcber 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\u00fcgbarkeit eines externen Systems ein wichtiger Teil unserer Gesch\u00e4ftsprozesse.<\/p>\n

<\/a><\/noindex><\/p>\n

Wie es vor Kafka war<\/h2>\n

<\/p>\n

Fr\u00fcher haben wir im Projekt IBM MQ f\u00fcr die asynchrone Zustellung von Nachrichten verwendet. Auftretende Fehler im Service konnten dazu f\u00fchren, dass die empfangene Nachricht in eine Dead-Letter-Queue (DLQ) gelegt wurde, um sie sp\u00e4ter manuell zu bearbeiten. Die DLQ wurde neben der eingehenden Warteschlange erstellt, und das Umlegen der Nachricht fand innerhalb von IBM MQ statt. <\/p>\n

<\/p>\n

Wenn der Fehler vor\u00fcbergehender Natur war und wir dies feststellen konnten (zum Beispiel ResourceAccessException bei HTTP-Aufrufen oder MongoTimeoutException bei Anfragen an MongoDB), trat die R\u00fcckrufstrategie in Kraft. Unabh\u00e4ngig von den Verzweigungen in der Logik der Anwendung wurde die urspr\u00fcngliche Nachricht entweder in die Systemwarteschlange f\u00fcr verz\u00f6gerte Zustellungen oder in eine separate Anwendung gelegt, die einst f\u00fcr die erneute Zusendung von Nachrichten entwickelt wurde. Dabei wird der R\u00fccksendungsnummer im Nachrichtenkopf vermerkt, der an das Verz\u00f6gerungsintervall 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\u00fcgbar ist, wird die Nachricht in die DLQ zur manuellen Bearbeitung gelegt.<\/p>\n

<\/p>\n

L\u00f6sungsfindung<\/h2>\n

<\/p>\n

Nach einer Internetrecherche<\/a><\/noindex>, l\u00e4sst sich Folgendes finden L\u00f6sung<\/a><\/noindex>. Kurz gesagt, es wird vorgeschlagen, f\u00fcr jedes Delay-Intervall ein Topic einzurichten und Consumer auf der Anwendungseite zu implementieren, die die Nachrichten mit der erforderlichen Verz\u00f6gerung lesen werden. <\/p>\n

<\/p>\n

\"Ereigniswiederverarbeitung<\/p>\n

<\/p>\n

Trotz der zahlreichen positiven R\u00fcckmeldungen erscheint es mir nicht ganz gelungen. Vor allem, weil der Entwickler neben der Umsetzung der Gesch\u00e4ftsanforderungen viel Zeit in die Implementierung des beschriebenen Mechanismus investieren muss.<\/p>\n

<\/p>\n

Dar\u00fcber hinaus, wenn in einem Kafka-Cluster Zugriffsmanagement aktiviert ist, wird es einige Zeit kosten, Topics einzurichten und die erforderlichen Zugriffsrechte zu gew\u00e4hren. Hinzu kommt, dass der richtige Wert f\u00fcr retention.ms f\u00fcr 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\u00fcr jede bestehende oder neue Dienstleistung wiederholt werden.<\/p>\n

<\/p>\n

Schauen wir uns nun an, welche Mechanismen f\u00fcr die erneute Verarbeitung von Nachrichten Spring im Allgemeinen und Spring-Kafka im Besonderen bietet. Spring-Kafka hat eine transitive Abh\u00e4ngigkeit 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\u00e4hrend des Betriebs zum Verlust aller Nachrichten f\u00fchrt, die auf die erneute Verarbeitung warten. Da dieser Punkt kritisch f\u00fcr unser System ist, haben wir ihn nicht weiter untersucht.<\/p>\n

<\/p>\n

Spring-Kafka selbst bietet mehrere Implementierungen von ContainerAwareErrorHandler, zum Beispiel SeekToCurrentErrorHandler<\/a><\/noindex>, mit dem es m\u00f6glich ist, die Nachricht sp\u00e4ter zu verarbeiten, ohne den Offset im Falle eines Fehlers zu verschieben. Seit Version 2.3 von Spring-Kafka gibt es die M\u00f6glichkeit, eine BackOffPolicy festzulegen.<\/p>\n

<\/p>\n

Dieser Ansatz erm\u00f6glicht es verarbeiteten Nachrichten, einen Neustart der Anwendung zu \u00fcberstehen, jedoch fehlt nach wie vor der DLQ-Mechanismus. Genau diese Variante w\u00e4hlten wir Anfang 2019, optimistisch in der Annahme, dass wir keinen DLQ ben\u00f6tigen w\u00fcrden (wir hatten Gl\u00fcck, und tats\u00e4chlich ben\u00f6tigten wir ihn in den ersten Monaten des Betriebs dieser Wiederholungsbehandlungsanwendung nicht). Tempor\u00e4re Fehler f\u00fchrten zur Ausf\u00fchrung des SeekToCurrentErrorHandler. Andere Fehler wurden im Log protokolliert, f\u00fchrten zu einer Verschiebung des Offsets, und die Verarbeitung setzte mit der n\u00e4chsten Nachricht fort.<\/p>\n

<\/p>\n

Endg\u00fcltige Entscheidung<\/h2>\n

<\/p>\n

Die Implementierung, die auf SeekToCurrentErrorHandler basiert, hat uns dazu angesto\u00dfen, einen eigenen Mechanismus zur Wiederholung des Sendens von Nachrichten zu entwickeln.<\/p>\n

<\/p>\n

Zun\u00e4chst wollten wir die bestehenden Erfahrungen nutzen und sie je nach Logik der Anwendung erweitern. F\u00fcr eine Anwendung mit linearer Logik w\u00e4re es optimal, das Lesen neuer Nachrichten f\u00fcr einen kurzen Zeitraum, der im Rahmen der Retry-Strategie festgelegt ist, zu unterbrechen. F\u00fcr andere Anwendungen w\u00e4re es w\u00fcnschenswert, einen zentralen Punkt zu haben, der die Umsetzung der Retry-Strategie gew\u00e4hrleistet. Dar\u00fcber hinaus sollte dieser zentrale Punkt \u00fcber die Funktionalit\u00e4t einer DLQ f\u00fcr beide Ans\u00e4tze verf\u00fcgen.<\/p>\n

<\/p>\n

Die Retry-Strategie selbst sollte in der Anwendung gespeichert werden, die f\u00fcr den Empfang des n\u00e4chsten Intervalls bei Auftreten eines tempor\u00e4ren Fehlers verantwortlich ist.<\/p>\n

<\/p>\n

Anhalten des Consumers f\u00fcr eine Anwendung mit linearer Logik<\/h3>\n

<\/p>\n

Bei der Arbeit mit spring-kafka k\u00f6nnte der Code zum Anhalten des Consumers etwa so aussehen:<\/p>\n

<\/p>\n

public void pauseListenerContainer(MessageListenerContainer listenerContainer, \n                                   Instant retryAt) {\n        if (nonNull(retryAt) && listenerContainer.isRunning()) {\n            listenerContainer.stop();\n            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);\n            return;\n        }\n        \/\/ zur DLQ\n    }<\/code><\/pre>\n
<\/p>\n
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. <\/p>\n
<\/p>\n
Der Wert retryAt wird folgenderma\u00dfen ermittelt:<\/p>\n
<\/p>\n
    \n
  1. Es wird der Wert des Retry-Z\u00e4hlers gesucht.<\/li>\n
  2. Entsprechend dem Wert des Z\u00e4hlers wird das aktuelle Verz\u00f6gerungsintervall in der Retry-Strategie gesucht. Die Strategie wird in der Anwendung selbst definiert, wir haben das JSON-Format f\u00fcr ihre Speicherung gew\u00e4hlt.<\/li>\n
  3. Das im JSON-Array gefundene Intervall enth\u00e4lt die Anzahl der Sekunden, nach denen die Verarbeitung wiederholt werden muss. Diese Anzahl wird zur aktuellen Zeit hinzugef\u00fcgt, um den Wert f\u00fcr retryAt zu bilden.<\/li>\n
  4. Wenn das Intervall nicht gefunden wird, ist der Wert retryAt null und die Nachricht wird zur manuellen Analyse in die DLQ gesendet.<\/li>\n<\/ol>\n
    <\/p>\n
    Bei diesem Ansatz bleibt nur, die Anzahl der Wiederholungsaufrufe f\u00fcr jede Nachricht zu speichern, die sich derzeit in der Verarbeitung befindet, zum Beispiel im Arbeitsspeicher der Anwendung. Das Speichern des Versuchsz\u00e4hlers im Speicher ist f\u00fcr diesen Ansatz nicht kritisch, da eine Anwendung mit linearer Logik die Verarbeitung insgesamt nicht durchf\u00fchren kann. Im Gegensatz zu spring-retry f\u00fchrt ein Neustart der Anwendung nicht zum Verlust aller Nachrichten zur erneuten Verarbeitung, sondern lediglich zum Neustart der Strategie. <\/p>\n
    <\/p>\n
    Dieser Ansatz hilft, die Last von externen Systemen zu verringern, die aufgrund einer sehr hohen Auslastung m\u00f6glicherweise nicht verf\u00fcgbar sind. Anders ausgedr\u00fcckt, haben wir neben der erneuten Verarbeitung auch das Muster implementiert. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    In unserem Fall liegt der Fehlerthreshold nur bei 1, und um die Ausfallzeiten des Systems aufgrund vor\u00fcbergehender Netzwerkunterbrechungen zu minimieren, verwenden wir eine sehr granulare Wiederholungsstrategie mit kurzen Verz\u00f6gerungsintervallen. Dies k\u00f6nnte nicht f\u00fcr alle Anwendungen der Unternehmensgruppe geeignet sein, daher muss das Verh\u00e4ltnis zwischen dem Fehlerthreshold und der L\u00e4nge des Intervalls je nach den Besonderheiten des Systems angepasst werden.<\/p>\n
    <\/p>\n
    Eine separate Anwendung zur Verarbeitung von Nachrichten aus Anwendungen mit nichtdeterministischer Logik<\/h3>\n
    <\/p>\n
    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:<\/p>\n
    <\/p>\n
    \npublic <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List<Header> arrayOfHeaders = \n            new ArrayList<>(Arrays.asList(headers.toArray()));\n        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);\n        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);\n        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, \n                     () -> Integer.toString(record.partition()).getBytes());\n        if (nonNull(retryAt)) {\n            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);\n            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);\n        } else {\n            updateHeader(arrayOfHeaders, REASON, \n                         ExceptionUtils.getStackTrace(e)::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);\n        }\n        ProducerRecord<K, V> messageToSend =\n            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);\n        kafkaTemplate.send(messageToSend);\n    }<\/code><\/pre>\n
    <\/p>\n
    Aus dem Beispiel geht hervor, dass viele Informationen in den Headern \u00fcbermittelt 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 \u00fcbermitteln wir:<\/p>\n
    <\/p>\n