{"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\/ro\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Reprelucrarea evenimentelor primite din Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Reprelucrarea<\/p>\n

<\/p>\n

Bun\u0103, Habr.<\/p>\n

<\/p>\n

Recent am \u00eemp\u0103rt\u0103\u0219it experien\u021ba<\/a><\/noindex> despre ce parametrii folosim cel mai adesea \u00een echip\u0103 pentru Kafka Producer \u0219i Consumer pentru a ne apropia de livrarea garantat\u0103. \u00cen acest articol vreau s\u0103 discut despre cum am organizat procesarea repetat\u0103 a evenimentului primit din Kafka, ca urmare a indisponibilit\u0103\u021bii temporare a unei sisteme externe.<\/p>\n

<\/p>\n

Aplica\u021biile moderne func\u021bioneaz\u0103 \u00eentr-un mediu foarte complex. Logica de afaceri, \u00eenv\u0103luit\u0103 \u00een o stiv\u0103 tehnologic\u0103 modern\u0103, func\u021bioneaz\u0103 \u00eentr-o imagine Docker, gestionat\u0103 de un orchestrator precum Kubernetes sau OpenShift, \u0219i comunic\u0103 cu alte aplica\u021bii sau solu\u021bii enterprise printr-o serie de routere fizice \u0219i virtuale. \u00centr-un astfel de mediu, mereu poate ap\u0103rea ceva care se stric\u0103, astfel c\u0103 procesarea repetat\u0103 a evenimentelor \u00een cazul \u00een care una dintre sistemele externe devine indisponibil\u0103 este o parte important\u0103 a proceselor noastre de afaceri.<\/p>\n

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

Cum era \u00eenainte de Kafka<\/h2>\n

<\/p>\n

Anterior, \u00een proiect, foloseam IBM MQ pentru livrarea asincron\u0103 a mesajelor. C\u00e2nd ap\u0103rea vreo eroare \u00een procesul de func\u021bionare a serviciului, mesajul primit putea fi plasat \u00eentr-o coad\u0103 de mesaje moarte (DLQ) pentru o analiz\u0103 ulterioar\u0103 manual\u0103. DLQ era creat\u0103 l\u00e2ng\u0103 coada de intrare, mutarea mesajului av\u00e2nd loc \u00een interiorul IBM MQ. <\/p>\n

<\/p>\n

Dac\u0103 eroarea avea un caracter temporar \u0219i puteam determina asta (de exemplu, ResourceAccessException \u00een timpul unui apel HTTP sau MongoTimeoutException \u00een timpul unei interog\u0103ri \u00een MongoDb), atunci intervenea strategia de apeluri repetate. Indiferent de ramificarea logicii aplica\u021biei, mesajul original era mutat fie \u00een coada sistemului pentru trimitere \u00eent\u00e2rziat\u0103, fie \u00een o aplica\u021bie separat\u0103, care a fost creat\u0103 cu mult timp \u00een urm\u0103 pentru retrimiterea mesajelor. \u00cen acest context, \u00een antetul mesajului se \u00eenregistra num\u0103rul de retrimitere, care era legat de intervalul de \u00eent\u00e2rziere sau de sf\u00e2r\u0219itul strategiei la nivel de aplica\u021bie. Dac\u0103 am atins sf\u00e2r\u0219itul strategiei, dar sistemul extern era \u00eenc\u0103 indisponibil, mesajul era plasat \u00een DLQ pentru analiz\u0103 manual\u0103.<\/p>\n

<\/p>\n

C\u0103utarea solu\u021biei<\/h2>\n

<\/p>\n

C\u0103ut\u00e2nd pe internet<\/a><\/noindex>, po\u021bi g\u0103si urm\u0103toarele solu\u021bie<\/a><\/noindex>. Pe scurt, se propune s\u0103 se creeze o tem\u0103 pentru fiecare interval de \u00eent\u00e2rziere \u0219i s\u0103 se implementeze de partea aplica\u021biei Consumers care vor citi mesajele cu \u00eent\u00e2rzierea necesar\u0103. <\/p>\n

<\/p>\n

\"Reprelucrarea<\/p>\n

<\/p>\n

\u00cen ciuda numeroaselor recenzii pozitive, mi se pare c\u0103 nu este complet reu\u0219it. \u00cen primul r\u00e2nd, deoarece dezvoltatorul, pe l\u00e2ng\u0103 implementarea cerin\u021belor de afaceri, va trebui s\u0103 aloce mult timp pentru realizarea mecanismului descris.<\/p>\n

<\/p>\n

\u00cen plus, dac\u0103 pe clusterul Kafka este activat\u0103 gestionarea accesului, va trebui s\u0103 aloce ceva timp pentru a crea topicuri \u0219i a asigura accesul necesar la acestea. \u00cen plus, va trebui s\u0103 selecteze parametrul corect retention.ms pentru fiecare din topicurile de retry, astfel \u00eenc\u00e2t mesajele s\u0103 fie retransmise \u00een timp \u0219i s\u0103 nu dispar\u0103 din acestea. Implementarea \u0219i solicitarea accesului va trebui repetat\u0103 pentru fiecare serviciu existent sau nou.<\/p>\n

<\/p>\n

S\u0103 vedem acum ce mecanisme de procesare repetat\u0103 a mesajelor ne ofer\u0103 spring \u00een general \u0219i spring-kafka \u00een special. Spring-kafka are o dependen\u021b\u0103 tranzitiv\u0103 de spring-retry, care ofer\u0103 abstra\u021bii pentru gestionarea diferitelor BackOffPolicy. Acesta este un instrument destul de flexibil, dar dezavantajul s\u0103u semnificativ este stocarea mesajelor pentru retransmitere \u00een memoria aplica\u021biei. Asta \u00eenseamn\u0103 c\u0103 repornirea aplica\u021biei din cauza unei actualiz\u0103ri sau a unei erori \u00een timpul utiliz\u0103rii va duce la pierderea tuturor mesajelor care a\u0219teapt\u0103 procesarea repetat\u0103. Deoarece acest punct este critic pentru sistemul nostru, nu l-am mai analizat ulterior.<\/p>\n

<\/p>\n

Spring-kafka \u00een sine ofer\u0103 mai multe implement\u0103ri ale ContainerAwareErrorHandler, de exemplu, SeekToCurrentErrorHandler<\/a><\/noindex>, cu ajutorul c\u0103ruia se poate, f\u0103r\u0103 a muta offset-ul \u00een caz de eroare, procesa mesajul mai t\u00e2rziu. \u00cencep\u00e2nd cu versiunea spring-kafka 2.3, a ap\u0103rut posibilitatea de a defini BackOffPolicy.<\/p>\n

<\/p>\n

Aceast\u0103 abordare permite mesajelor retrase s\u0103 supravie\u021buiasc\u0103 repornirii aplica\u021biei, dar mecanismul DLQ este \u00eenc\u0103 absent. Acesta a fost variant\u0103 pe care am ales-o la \u00eenceputul anului 2019, fiind optim\u0219ti c\u0103 DLQ nu va fi necesar (am avut noroc \u0219i \u00eentr-adev\u0103r nu a fost nevoie \u00een c\u00e2teva luni de utilizare a aplica\u021biei cu un astfel de sistem de procesare repetat\u0103). Erorile temporare au dus la activarea SeekToCurrentErrorHandler. Celelalte erori au fost tip\u0103rite \u00een loguri, au dus la mutarea offset-ului \u0219i procesarea a continuat cu urm\u0103torul mesaj.<\/p>\n

<\/p>\n

Solu\u021bia final\u0103<\/h2>\n

<\/p>\n

Implementarea bazat\u0103 pe SeekToCurrentErrorHandler ne-a \u00eempins s\u0103 dezvolt\u0103m un mecanism propriu pentru retransmiterea mesajelor.<\/p>\n

<\/p>\n

\u00cen primul r\u00e2nd, am dorit s\u0103 folosim experien\u021ba existent\u0103 \u0219i s\u0103 o extindem \u00een func\u021bie de logica aplica\u021biei. Pentru o aplica\u021bie cu o logic\u0103 liniar\u0103, ar fi optim s\u0103 oprim citirea de noi mesaje pe o perioad\u0103 scurt\u0103 de timp, specificat\u0103 \u00een cadrul strategiei de retransmitere. Pentru celelalte aplica\u021bii, am dorit s\u0103 avem un punct unic care s\u0103 asigure implementarea strategiei de retransmitere. \u00cen plus, acest punct unic ar trebui s\u0103 dispun\u0103 de func\u021bionalitatea DLQ pentru ambele abord\u0103ri.<\/p>\n

<\/p>\n

Strategia de retransmitere propriu-zis\u0103 ar trebui s\u0103 fie stocat\u0103 \u00een aplica\u021bia responsabil\u0103 pentru ob\u021binerea urm\u0103torului interval \u00een cazul unei erori temporare.<\/p>\n

<\/p>\n

Oprirea Consumer-ului pentru o aplica\u021bie cu logic\u0103 liniar\u0103<\/h3>\n

<\/p>\n

Atunci c\u00e2nd lucr\u0103m cu spring-kafka, codul pentru oprirea Consumer-ului ar putea ar\u0103ta cam a\u0219a:<\/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        \/\/ to DLQ\n    }<\/code><\/pre>\n
<\/p>\n
\u00cen exemplu, retryAt este momentul \u00een care trebuie s\u0103 restart\u0103m MessageListenerContainer, dac\u0103 acesta este \u00eenc\u0103 activ. Restartarea va avea loc \u00eentr-un fir de execu\u021bie separat, lansat \u00een TaskScheduler, a c\u0103rei implementare o ofer\u0103 de asemenea spring. <\/p>\n
<\/p>\n
Valoarea retryAt o g\u0103sim \u00een urm\u0103torul mod:<\/p>\n
<\/p>\n
    \n
  1. Se caut\u0103 valoarea contorului de retransmiteri.<\/li>\n
  2. Conform valorii contorului, se caut\u0103 intervalul curent de \u00eent\u00e2rziere \u00een strategia de retransmitere. Strategia este declarat\u0103 \u00een cadrul aplica\u021biei, iar pentru stocarea acesteia am ales formatul JSON.<\/li>\n
  3. Intervalul g\u0103sit \u00een array-ul JSON con\u021bine num\u0103rul de secunde dup\u0103 care trebuie s\u0103 repet\u0103m procesarea. Acest num\u0103r de secunde este ad\u0103ugat la timpul curent, form\u00e2nd valoarea pentru retryAt.<\/li>\n
  4. Dac\u0103 intervalul nu este g\u0103sit, atunci valoarea retryAt este null \u0219i mesajul va fi trimis \u00een DLQ pentru o analiz\u0103 manual\u0103.<\/li>\n<\/ol>\n
    <\/p>\n
    Cu aceast\u0103 abordare, trebuie doar s\u0103 p\u0103str\u0103m num\u0103rul apelurilor repetate pentru fiecare mesaj care este acum \u00een procesare, de exemplu, \u00een memoria aplica\u021biei. P\u0103strarea unui contor de \u00eencerc\u0103ri \u00een memorie nu este critic\u0103 pentru aceast\u0103 abordare, deoarece aplica\u021bia cu o logic\u0103 liniar\u0103 nu poate efectua procesarea \u00een \u00eentregime. Spre deosebire de spring-retry, repornirea aplica\u021biei nu va duce la pierderea tuturor mesajelor pentru re-procesare, ci pur \u0219i simplu la reluarea strategiei. <\/p>\n
    <\/p>\n
    Aceast\u0103 abordare ajut\u0103 la reducerea sarcinii asupra sistemului extern, care poate fi inaccesibil din cauza unei sarcini foarte mari. Cu alte cuvinte, \u00een plus fa\u021b\u0103 de re-procesare, am realizat implementarea modelului circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    \u00cen cazul nostru, pragul de eroare este doar 1, iar pentru a minimiza timpul de nefunc\u021bionare a sistemului din cauza unei \u00eentreruperi temporare a re\u021belei, folosim o strategie de apeluri repetitive foarte granular\u0103 cu intervale de \u00eent\u00e2rzieri mici. Acest lucru poate s\u0103 nu fie potrivit pentru toate aplica\u021biile grupului de companii, a\u0219a c\u0103 raportul \u00eentre pragul de eroare \u0219i dimensiunea intervalului trebuie ajustat, baz\u00e2ndu-se pe caracteristicile sistemului.<\/p>\n
    <\/p>\n
    O aplica\u021bie separat\u0103 pentru procesarea mesajelor de la aplica\u021bii cu o logic\u0103 nedeterminist\u0103<\/h3>\n
    <\/p>\n
    Iat\u0103 un exemplu de cod care trimite un mesaj c\u0103tre o astfel de aplica\u021bie (Retryer), care va re\u00eencerca trimiterea \u00een topicul DESTINATION atunci c\u00e2nd se atinge timpul RETRY_AT:<\/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
    Din exemplu, se observ\u0103 c\u0103 se transmite multe informa\u021bii \u00een header-e. Valoarea RETRY_AT se afl\u0103 la fel ca \u0219i pentru mecanismul de re\u00eencercare prin oprirea Consumer-ului. Pe l\u00e2ng\u0103 DESTINATION \u0219i RETRY_AT, mai transmitem:<\/p>\n
    <\/p>\n