{"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":"Wiederverarbeitung von Ereignissen aus Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Wiederverarbeitung<\/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 Kafka Producer und Consumer verwenden, um eine garantierte Lieferung zu erreichen. In diesem Artikel m\u00f6chte ich erl\u00e4utern, wie wir die Verarbeitung von Ereignissen organisiert haben, die aus Kafka erhalten wurden, aufgrund der vor\u00fcbergehenden Nichterreichbarkeit eines externen Systems.<\/p>\n

<\/p>\n

Moderne Anwendungen arbeiten in einer sehr komplexen Umgebung. Die Gesch\u00e4ftslogik, eingeh\u00fcllt in einen modernen Technologiestack, der in einem Docker-Image l\u00e4uft, welches von einem Orchestrator wie Kubernetes oder OpenShift verwaltet wird, und das mit anderen Anwendungen oder Unternehmensl\u00f6sungen \u00fcber 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\u00e4ftsprozesse ist.<\/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. 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. <\/p>\n

<\/p>\n

Wenn der Fehler vor\u00fcbergehender 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\u00fcr Wiederholungsaufrufe in Kraft. Unabh\u00e4ngig von der Logik des Anwendungsszenarios wurde die urspr\u00fcngliche Nachricht entweder in die systemweite Warteschlange f\u00fcr verz\u00f6gerte 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\u00f6gerungsintervall 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\u00fcfung verschoben.<\/p>\n

<\/p>\n

Suche nach einer L\u00f6sung<\/h2>\n

<\/p>\n

Nach einer Internetsuche<\/a><\/noindex>, kann Folgendes gefunden werden L\u00f6sung<\/a><\/noindex>. Kurz gesagt, es wird vorgeschlagen, f\u00fcr jedes Verz\u00f6gerungsintervall ein Thema zu erstellen und auf der Anwendungsseite Consumer zu implementieren, die Nachrichten mit der ben\u00f6tigten Verz\u00f6gerung lesen. <\/p>\n

<\/p>\n

\"Wiederverarbeitung<\/p>\n

<\/p>\n

Trotz der vielen positiven Bewertungen scheint es mir nicht ganz gelungen zu sein. Vor allem weil der Entwickler, neben der Umsetzung der Gesch\u00e4ftsanforderungen, viel Zeit mit der Implementierung des beschriebenen Mechanismus verbringen muss.<\/p>\n

<\/p>\n

Dar\u00fcber hinaus, wenn auf dem Kafka-Cluster die Zugriffsverwaltung aktiviert ist, muss einige Zeit f\u00fcr die Erstellung von Topics und die Gew\u00e4hrleistung der notwendigen Zugriffsrechte auf diese aufgewendet werden. Zus\u00e4tzlich muss der richtige Parameter retention.ms f\u00fcr jedes der Retry-Topics festgelegt werden, damit die Nachrichten rechtzeitig erneut gesendet werden und nicht verloren gehen. Die Implementierung und Anforderung von Zug\u00e4ngen muss f\u00fcr jeden bestehenden oder neuen Dienst wiederholt werden.<\/p>\n

<\/p>\n

Lassen Sie uns nun betrachten, 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 von Spring-Retry, das Abstraktionen f\u00fcr die Verwaltung verschiedener BackOffPolicies bereitstellt. Es ist ein ziemlich flexibles Werkzeug, allerdings ist ein wesentlicher Nachteil die Speicherung von Nachrichten f\u00fcr die erneute Sendung 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 eine erneute Verarbeitung warten. Da dieser Punkt f\u00fcr unser System entscheidend ist, haben wir ihn nicht weiter betrachtet.<\/p>\n

<\/p>\n

Spring-Kafka selbst bietet mehrere Implementierungen von ContainerAwareErrorHandler, wie zum Beispiel SeekToCurrentErrorHandler<\/a><\/noindex>, mit dem es m\u00f6glich ist, die Offset im Fehlerfall nicht zu verschieben und die Nachricht sp\u00e4ter zu verarbeiten. Seit der Version 2.3 von Spring-Kafka gibt es die M\u00f6glichkeit, eine BackOffPolicy zu definieren.<\/p>\n

<\/p>\n

Dieser Ansatz erm\u00f6glicht es, dass erneut verarbeitete Nachrichten einen Neustart der Anwendung \u00fcberstehen, jedoch fehlt nach wie vor der Mechanismus f\u00fcr Dead Letter Queues (DLQ). Diesen Ansatz w\u00e4hlten wir Anfang 2019 in der optimistischen Annahme, dass eine DLQ nicht erforderlich sein w\u00fcrde (wir hatten das Gl\u00fcck, dass sie in den ersten Monaten des Betriebs der Anwendung mit diesem Wiederverarbeitungssystem tats\u00e4chlich nicht ben\u00f6tigt wurde). Tempor\u00e4re Fehler f\u00fchrten zur Aktivierung des SeekToCurrentErrorHandler. Andere Fehler wurden in das Protokoll geschrieben, f\u00fchrten zur Verschiebung des Offsets und die Verarbeitung wurde mit der n\u00e4chsten Nachricht fortgesetzt.<\/p>\n

<\/p>\n

Die endg\u00fcltige Entscheidung<\/h2>\n

<\/p>\n

Die auf SeekToCurrentErrorHandler basierende Implementierung hat uns dazu angeregt, einen eigenen Mechanismus zur erneuten \u00dcbermittlung von Nachrichten zu entwickeln.<\/p>\n

<\/p>\n

Zun\u00e4chst wollten wir die bereits vorhandene Erfahrung 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 Strategien f\u00fcr erneute Aufrufe festgelegt wurde, zu unterbrechen. F\u00fcr andere Anwendungen wollte man einen zentralen Punkt haben, der die Durchf\u00fchrung der Wiederholungsstrategien gew\u00e4hrleistet. Dar\u00fcber hinaus sollte dieser zentrale Punkt \u00fcber die DLQ-Funktionalit\u00e4t f\u00fcr beide Ans\u00e4tze verf\u00fcgen.<\/p>\n

<\/p>\n

Die Wiederholungsstrategie selbst sollte in der Anwendung gespeichert werden, die f\u00fcr den Empfang des n\u00e4chsten Intervalls bei einem zeitlichen Fehler verantwortlich ist.<\/p>\n

<\/p>\n

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

<\/p>\n

Beim Arbeiten mit spring-kafka k\u00f6nnte der Code zum Stoppen 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        \/\/ an DLQ\n    }<\/code><\/pre>\n
<\/p>\n
Im Beispiel ist retryAt der Zeitpunkt, zu dem das MessageListenerContainer neu gestartet werden muss, wenn es noch l\u00e4uft. Der Neustart erfolgt in einem separaten Thread, der im TaskScheduler gestartet wird, dessen Implementierung ebenfalls von spring bereitgestellt wird. <\/p>\n
<\/p>\n
Wir finden den Wert von retryAt auf die folgende Weise:<\/p>\n
<\/p>\n
    \n
  1. Der Wert des Wiederholungsz\u00e4hlers wird gesucht.<\/li>\n
  2. Entsprechend dem Wert des Z\u00e4hlers wird das aktuelle Verz\u00f6gerungsintervall in der Strategie f\u00fcr erneute Aufrufe gesucht. Die Strategie wird in der Anwendung selbst erkl\u00e4rt; f\u00fcr ihre Speicherung haben wir das JSON-Format 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 von Sekunden wird zur aktuellen Zeit addiert, um den Wert f\u00fcr retryAt zu bilden.<\/li>\n
  4. Wenn das Intervall nicht gefunden wird, ist der Wert von retryAt null und die Nachricht wird zur manuellen \u00dcberpr\u00fcfung in die DLQ gesendet.<\/li>\n<\/ol>\n
    <\/p>\n
    Bei diesem Ansatz bleibt nur die M\u00f6glichkeit, die Anzahl der Wiederholungsversuche f\u00fcr jede Nachricht, die derzeit verarbeitet wird, beispielsweise im Arbeitsspeicher der Anwendung zu speichern. Das Speichern des Z\u00e4hlerstands im Speicher ist f\u00fcr diesen Ansatz nicht kritisch, da Anwendungen mit linearer Logik die Verarbeitung insgesamt nicht durchf\u00fchren k\u00f6nnen. Im Gegensatz zu spring-retry f\u00fchrt ein Neustart der Anwendung nicht zum Verlust aller Nachrichten f\u00fcr die erneute Verarbeitung, sondern einfach zum Neustart der Strategie. <\/p>\n
    <\/p>\n
    Dieser Ansatz hilft, die Belastung f\u00fcr externe Systeme zu verringern, die aufgrund einer sehr hohen Last m\u00f6glicherweise nicht verf\u00fcgbar sind. Anders ausgedr\u00fcckt, zus\u00e4tzlich zur erneuten Verarbeitung haben wir die Implementierung des Musters erreicht. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    In unserem Fall liegt der Fehlergrenzwert nur bei 1, und um die Ausfallzeit des Systems aufgrund vor\u00fcbergehender Netzwerkunterbrechungen zu minimieren, verwenden wir eine sehr granulare Wiederholungsstrategie mit kurzen Verz\u00f6gerungsintervallen. Dies ist m\u00f6glicherweise nicht f\u00fcr alle Anwendungen der Unternehmensgruppe geeignet, daher muss das Verh\u00e4ltnis zwischen Fehlergrenzwert und Intervallgr\u00f6\u00dfe an die Besonderheiten des Systems angepasst werden.<\/p>\n
    <\/p>\n
    Eine separate Anwendung zur Verarbeitung von Nachrichten von Anwendungen mit undeterministischer Logik.<\/h3>\n
    <\/p>\n
    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\u00fchren wird:<\/p>\n
    <\/p>\n
    \npublic  void retry(ConsumerRecord record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List
     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 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 \u00fcbertragen werden. Der Wert RETRY_AT befindet sich ebenfalls wie beim Wiederholungsmechanismus \u00fcber das Stoppen des Consumers. Neben DESTINATION und RETRY_AT \u00fcbergeben wir:<\/p>\n
    <\/p>\n