{"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\/fr\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Reprocessing des \u00e9v\u00e9nements re\u00e7us de Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Reprocessing<\/p>\n

<\/p>\n

Bonjour, Habr.<\/p>\n

<\/p>\n

R\u00e9cemment, j'ai partag\u00e9 mon exp\u00e9rience<\/a><\/noindex> sur les param\u00e8tres que nous utilisons le plus souvent dans notre \u00e9quipe pour Kafka Producer et Consumer, afin de nous rapprocher d'une livraison garantie. Dans cet article, je souhaite expliquer comment nous avons organis\u00e9 le traitement des \u00e9v\u00e9nements re\u00e7us de Kafka en raison de l'indisponibilit\u00e9 temporaire d'un syst\u00e8me externe.<\/p>\n

<\/p>\n

Les applications modernes fonctionnent dans un environnement tr\u00e8s complexe. La logique m\u00e9tier, envelopp\u00e9e dans une pile technologique moderne, fonctionne dans une image Docker g\u00e9r\u00e9e par un orchestrateur tel que Kubernetes ou OpenShift, et communique avec d'autres applications ou des solutions d'entreprise via une cha\u00eene de routeurs physiques et virtuels. Dans un tel environnement, quelque chose peut toujours casser, c'est pourquoi le traitement des \u00e9v\u00e9nements en cas d'indisponibilit\u00e9 de l'un des syst\u00e8mes externes est une partie importante de nos processus m\u00e9tier.<\/p>\n

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

Comment c'\u00e9tait avant Kafka<\/h2>\n

<\/p>\n

Auparavant, dans le projet, nous utilisions IBM MQ pour la livraison asynchrone des messages. En cas d'erreur survenant dans le processus de travail du service, le message re\u00e7u pouvait \u00eatre plac\u00e9 dans une dead-letter-queue (DLQ) pour un examen manuel ult\u00e9rieur. La DLQ \u00e9tait cr\u00e9\u00e9e \u00e0 c\u00f4t\u00e9 de la file d'attente entrante, le transfert du message se faisait au sein d'IBM MQ. <\/p>\n

<\/p>\n

Si l'erreur \u00e9tait temporaire et que nous pouvions le d\u00e9terminer (par exemple, ResourceAccessException lors d'un appel HTTP ou MongoTimeoutException lors d'une requ\u00eate \u00e0 MongoDb), alors la strat\u00e9gie des appels r\u00e9p\u00e9t\u00e9s entrait en vigueur. Peu importe la branche de la logique de l'application, le message d'origine \u00e9tait transf\u00e9r\u00e9 soit dans une file d'attente syst\u00e8me pour un envoi diff\u00e9r\u00e9, soit dans une application distincte qui avait \u00e9t\u00e9 cr\u00e9\u00e9e il y a longtemps pour renvoyer des messages. Dans ce cas, le num\u00e9ro de r\u00e9exp\u00e9dition, li\u00e9 \u00e0 l'intervalle de retard ou \u00e0 la fin de la strat\u00e9gie au niveau de l'application, \u00e9tait enregistr\u00e9 dans l'en-t\u00eate du message. Si nous atteignons la fin de la strat\u00e9gie mais que le syst\u00e8me externe reste indisponible, alors le message sera plac\u00e9 dans une DLQ pour un examen manuel.<\/p>\n

<\/p>\n

Recherche de solution<\/h2>\n

<\/p>\n

En cherchant sur Internet<\/a><\/noindex>, on peut trouver ce qui suit la solution<\/a><\/noindex>. En bref, il est propos\u00e9 de cr\u00e9er un topic par intervalle de retard et de mettre en \u0153uvre des Consumers du c\u00f4t\u00e9 de l'application, qui liront les messages avec le retard n\u00e9cessaire. <\/p>\n

<\/p>\n

\"Reprocessing<\/p>\n

<\/p>\n

Bien qu'il y ait de nombreux avis positifs, il me semble qu'il n'est pas tout \u00e0 fait r\u00e9ussi. Tout d'abord, parce que le d\u00e9veloppeur, en plus de r\u00e9pondre aux exigences commerciales, devra consacrer beaucoup de temps \u00e0 la mise en \u0153uvre du m\u00e9canisme d\u00e9crit.<\/p>\n

<\/p>\n

De plus, si la gestion des acc\u00e8s est activ\u00e9e sur le cluster Kafka, il faudra passer un certain temps \u00e0 cr\u00e9er des topics et \u00e0 garantir les acc\u00e8s n\u00e9cessaires \u00e0 ceux-ci. En outre, il sera n\u00e9cessaire de choisir le bon param\u00e8tre retention.ms pour chacun des topics de r\u00e9essai, afin que les messages puissent \u00eatre renvoy\u00e9s \u00e0 temps et ne disparaissent pas. La mise en \u0153uvre et la demande d'acc\u00e8s devront \u00eatre r\u00e9p\u00e9t\u00e9es pour chaque service existant ou nouveau.<\/p>\n

<\/p>\n

Voyons maintenant quels m\u00e9canismes de traitement des messages en double nous propose spring en g\u00e9n\u00e9ral et spring-kafka en particulier. Spring-kafka a une d\u00e9pendance transitive sur spring-retry, qui fournit des abstractions pour g\u00e9rer diff\u00e9rentes BackOffPolicy. C'est un outil assez flexible, mais son inconv\u00e9nient majeur est le stockage des messages \u00e0 renvoyer en m\u00e9moire de l'application. Cela signifie qu'un red\u00e9marrage de l'application en raison d'une mise \u00e0 jour ou d'une erreur pendant l'exploitation entra\u00eenera la perte de tous les messages en attente de traitement \u00e0 nouveau. \u00c9tant donn\u00e9 que ce point est critique pour notre syst\u00e8me, nous avons choisi de ne pas le consid\u00e9rer davantage.<\/p>\n

<\/p>\n

La biblioth\u00e8que spring-kafka propose plusieurs impl\u00e9mentations de ContainerAwareErrorHandler, par exemple SeekToCurrentErrorHandler<\/a><\/noindex>, qui permet de traiter le message plus tard sans d\u00e9placer l'offset en cas d'erreur. Depuis la version 2.3 de spring-kafka, il est possible de d\u00e9finir une BackOffPolicy.<\/p>\n

<\/p>\n

Cette approche permet aux messages en r\u00e9essai de survivre au red\u00e9marrage de l'application, mais le m\u00e9canisme DLQ est toujours absent. C'est cette option que nous avons choisie au d\u00e9but de 2019, en pensant de mani\u00e8re optimiste que le DLQ ne serait pas n\u00e9cessaire (nous avons eu de la chance et il ne fut effectivement pas n\u00e9cessaire pendant quelques mois d'exploitation de l'application avec ce syst\u00e8me de r\u00e9essai). Des erreurs temporaires provoquaient l'activation de SeekToCurrentErrorHandler. Les autres erreurs \u00e9taient enregistr\u00e9es dans les logs, entra\u00eenant un d\u00e9calage de l'offset, et le traitement se poursuivait avec le message suivant.<\/p>\n

<\/p>\n

Solution finale<\/h2>\n

<\/p>\n

La mise en \u0153uvre bas\u00e9e sur SeekToCurrentErrorHandler nous a pouss\u00e9s \u00e0 d\u00e9velopper notre propre m\u00e9canisme pour le renvoi des messages.<\/p>\n

<\/p>\n

Avant tout, nous voulions tirer parti de l'exp\u00e9rience existante et l'\u00e9largir en fonction de la logique de l'application. Pour une application \u00e0 logique lin\u00e9aire, il serait optimal d'arr\u00eater la lecture de nouveaux messages pendant un court laps de temps d\u00e9fini dans la strat\u00e9gie de r\u00e9essai. Pour les autres applications, nous souhaiterions avoir un point central qui garantirait l'application de la strat\u00e9gie de r\u00e9essai. De plus, ce point unique doit disposer d'une fonctionnalit\u00e9 DLQ pour les deux approches.<\/p>\n

<\/p>\n

La strat\u00e9gie de r\u00e9essai elle-m\u00eame doit \u00eatre stock\u00e9e dans l'application responsable de la r\u00e9cup\u00e9ration du prochain intervalle en cas d'erreur temporaire.<\/p>\n

<\/p>\n

Arr\u00eat du Consumer pour une application \u00e0 logique lin\u00e9aire<\/h3>\n

<\/p>\n

Lors de l'utilisation de spring-kafka, le code pour arr\u00eater le Consumer peut ressembler \u00e0 ceci :<\/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        \/\/ vers DLQ\n    }<\/code><\/pre>\n
<\/p>\n
Dans cet exemple, retryAt est le moment auquel il faut red\u00e9marrer le MessageListenerContainer s'il est encore en cours d'ex\u00e9cution. Le red\u00e9marrage se fera dans un thread s\u00e9par\u00e9, lanc\u00e9 par TaskScheduler, dont l'impl\u00e9mentation est \u00e9galement fournie par spring. <\/p>\n
<\/p>\n
Nous trouvons la valeur de retryAt de la mani\u00e8re suivante :<\/p>\n
<\/p>\n
    \n
  1. Nous recherchons la valeur du compteur de r\u00e9essais.<\/li>\n
  2. En fonction de la valeur du compteur, l'intervalle de d\u00e9lai actuel est recherch\u00e9 dans la strat\u00e9gie de r\u00e9essai. La strat\u00e9gie est d\u00e9clar\u00e9e dans l'application elle-m\u00eame, et nous avons choisi le format JSON pour son stockage.<\/li>\n
  3. L'intervalle trouv\u00e9 dans le tableau JSON contient le nombre de secondes apr\u00e8s lesquelles il faudra r\u00e9essayer le traitement. Ce nombre de secondes est ajout\u00e9 au temps actuel, formant ainsi la valeur pour retryAt.<\/li>\n
  4. Si l'intervalle n'est pas trouv\u00e9, la valeur de retryAt est nulle et le message sera envoy\u00e9 dans la DLQ pour un traitement manuel.<\/li>\n<\/ol>\n
    <\/p>\n
    Avec cette approche, il ne reste qu'\u00e0 conserver le nombre de tentatives pour chaque message actuellement en cours de traitement, par exemple dans la m\u00e9moire de l'application. Conserver le compteur de tentatives en m\u00e9moire n'est pas critique pour cette approche, car une application avec une logique lin\u00e9aire ne peut pas traiter l'ensemble. Contrairement \u00e0 spring-retry, le red\u00e9marrage de l'application ne conduira pas \u00e0 la perte de tous les messages \u00e0 r\u00e9it\u00e9rer, mais simplement \u00e0 un red\u00e9marrage de la strat\u00e9gie. <\/p>\n
    <\/p>\n
    Cette approche permet de r\u00e9duire la charge sur le syst\u00e8me externe, qui peut \u00eatre indisponible en raison d'une tr\u00e8s forte charge. En d'autres termes, en plus de la r\u00e9it\u00e9ration, nous avons r\u00e9ussi \u00e0 mettre en \u0153uvre le mod\u00e8le. disjoncteur<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    Dans notre cas, le seuil d'erreur est de seulement 1, et pour minimiser le temps d'arr\u00eat du syst\u00e8me en raison d'une panne de r\u00e9seau temporaire, nous utilisons une strat\u00e9gie de r\u00e9it\u00e9ration tr\u00e8s granulaire avec de courtes intervalles de d\u00e9lai. Cela peut ne pas convenir \u00e0 toutes les applications du groupe, donc le ratio entre le seuil d'erreur et la taille de l'intervalle doit \u00eatre ajust\u00e9 en fonction des caract\u00e9ristiques du syst\u00e8me.<\/p>\n
    <\/p>\n
    Une application distincte pour le traitement des messages provenant d'applications avec une logique non d\u00e9terministe<\/h3>\n
    <\/p>\n
    Voici un exemple de code qui envoie un message \u00e0 une telle application (Retryer), qui effectuera une nouvelle tentative d'envoi vers le sujet DESTINATION lorsque le temps RETRY_AT sera atteint :<\/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
    L'exemple montre qu'une grande quantit\u00e9 d'informations est transmise dans les en-t\u00eates. La valeur RETRY_AT est d\u00e9termin\u00e9e de la m\u00eame mani\u00e8re que pour le m\u00e9canisme de r\u00e9p\u00e9tition via l'arr\u00eat du Consumer. En plus de DESTINATION et RETRY_AT, nous transmettons :<\/p>\n
    <\/p>\n