{"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\/sk\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Op\u00e4tovn\u00e9 spracovanie udalost\u00ed prijat\u00fdch z Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Op\u00e4tovn\u00e9<\/p>\n

<\/p>\n

Ahoj, Habr.<\/p>\n

<\/p>\n

Ned\u00e1vno som zdie\u013eal sk\u00fasenosti<\/a><\/noindex> o tom, ak\u00e9 parametre v t\u00edme naj\u010dastej\u0161ie pou\u017e\u00edvame pre Kafka Producer a Consumer, aby sme sa pribl\u00ed\u017eili k garantovanej dod\u00e1vke. V tomto \u010dl\u00e1nku chcem poveda\u0165, ako sme zorganizovali op\u00e4tovn\u00e9 spracovanie udalosti prijatej z Kafka v d\u00f4sledku do\u010dasnej nedostupnosti vonkaj\u0161ieho syst\u00e9mu.<\/p>\n

<\/p>\n

S\u00fa\u010dasn\u00e9 aplik\u00e1cie funguj\u00fa v ve\u013emi zlo\u017eitom prostred\u00ed. Biznis logika, zabalen\u00e1 v modernom technologickom stacku, be\u017eiaca v Docker obr\u00e1zku, ktor\u00fd spravuje orchestr\u00e1tor ako Kubernetes alebo OpenShift, a komunikuj\u00faca s in\u00fdmi aplik\u00e1ciami alebo enterprise rie\u0161eniami prostredn\u00edctvom re\u0165azca fyzick\u00fdch a virtu\u00e1lnych smerova\u010dov. V takomto prostred\u00ed sa v\u017edy m\u00f4\u017ee nie\u010do pokazi\u0165, preto je op\u00e4tovn\u00e9 spracovanie udalost\u00ed v pr\u00edpade nedostupnosti jedn\u00e9ho z vonkaj\u0161\u00edch syst\u00e9mov d\u00f4le\u017eitou s\u00fa\u010das\u0165ou na\u0161ich biznis procesov.<\/p>\n

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

Ako to bolo pred Kafka<\/h2>\n

<\/p>\n

Predt\u00fdm sme v projekte pou\u017e\u00edvali IBM MQ naasynchr\u00f3nny prenos spr\u00e1v. Ak do\u0161lo k nejakej chybe po\u010das fungovania slu\u017eby, prijat\u00e1 spr\u00e1va mohla by\u0165 umiestnen\u00e1 do dead-letter-queue (DLQ) na \u010fal\u0161ie manu\u00e1lne spracovanie. DLQ sa vytv\u00e1ral ved\u013ea vstupnej fronty a presun spr\u00e1vy prebiehal v r\u00e1mci IBM MQ. <\/p>\n

<\/p>\n

Ak bola chyba do\u010dasn\u00e1 a mohli sme to ur\u010di\u0165 (napr\u00edklad ResourceAccessException pri HTTP volan\u00ed alebo MongoTimeoutException pri dotaze do MongoDb), do platnosti vst\u00fapila strat\u00e9gia opakovan\u00fdch volan\u00ed. Bez oh\u013eadu na vetvenie logiky aplik\u00e1cie bola p\u00f4vodn\u00e1 spr\u00e1va pres\u00favan\u00e1 bu\u010f do syst\u00e9movej fronty na oneskoren\u00e9 odoslanie, alebo do samostatnej aplik\u00e1cie, ktor\u00e1 bola d\u00e1vno vytvoren\u00e1 na op\u00e4tovn\u00e9 odosielanie spr\u00e1v. Pri tom sa do hlavi\u010dky spr\u00e1vy zapisuje \u010d\u00edslo op\u00e4tovn\u00e9ho odoslania, ktor\u00e9 je viazan\u00e9 na \u010dasov\u00e9 oneskorenie alebo na koniec strat\u00e9gie na \u00farovni aplik\u00e1cie. Ak sme dosiahli koniec strat\u00e9gie, ale vonkaj\u0161\u00ed syst\u00e9m je st\u00e1le nedostupn\u00fd, spr\u00e1va bude umiestnen\u00e1 do DLQ na manu\u00e1lne spracovanie.<\/p>\n

<\/p>\n

H\u013eadanie rie\u0161enia<\/h2>\n

<\/p>\n

H\u013eadanie na internete<\/a><\/noindex>, m\u00f4\u017eete n\u00e1js\u0165 nasleduj\u00face rie\u0161enia.<\/a><\/noindex>. Ak to zhrnieme, navrhuje sa zavies\u0165 po jednu t\u00e9mu na ka\u017ed\u00fd interval oneskorenia a realizova\u0165 na strane aplik\u00e1cie Consumerov, ktor\u00ed bud\u00fa \u010d\u00edta\u0165 spr\u00e1vy s potrebn\u00fdm oneskoren\u00edm. <\/p>\n

<\/p>\n

\"Op\u00e4tovn\u00e9<\/p>\n

<\/p>\n

Hoci m\u00e1 ve\u013ea pozit\u00edvnych recenzi\u00ed, zd\u00e1 sa mi, \u017ee nie je \u00faplne \u00faspe\u0161n\u00e9. Predov\u0161etk\u00fdm preto, \u017ee v\u00fdvoj\u00e1rovi bude okrem implement\u00e1cie obchodn\u00fdch po\u017eiadaviek potrebova\u0165 ve\u013ea \u010dasu na zavedenie op\u00edsan\u00e9ho mechanizmu.<\/p>\n

<\/p>\n

Okrem toho, ak je v klasteri Kafka zapnut\u00e9 riadenie pr\u00edstupu, bude potrebn\u00e9 str\u00e1vi\u0165 nejak\u00fd \u010das vytv\u00e1ran\u00edm t\u00e9m a zabezpe\u010den\u00edm potrebn\u00fdch pr\u00edstupov k nim. Navy\u0161e je potrebn\u00e9 vybra\u0165 spr\u00e1vny parameter retention.ms pre ka\u017ed\u00fa z retrial-t\u00e9m, aby spr\u00e1vy stihli by\u0165 znovu odoslan\u00e9 a nestratili sa. Implement\u00e1ciu a \u017eiados\u0165 o pr\u00edstupy bude potrebn\u00e9 zopakova\u0165 pre ka\u017ed\u00fa existuj\u00facu alebo nov\u00fa slu\u017ebu.<\/p>\n

<\/p>\n

Po\u010fme sa teraz pozrie\u0165 na mechanizmy op\u00e4tovn\u00e9ho spracovania spr\u00e1v, ktor\u00e9 n\u00e1m pon\u00faka spring ako celok a spring-kafka osobitne. Spring-kafka m\u00e1 prenosn\u00fa z\u00e1vislos\u0165 na spring-retry, ktor\u00fd poskytuje abstrakcie na spr\u00e1vu r\u00f4znych BackOffPolicy. Je to celkom flexibiln\u00fd n\u00e1stroj, ale jeho v\u00fdznamn\u00fdm nedostatkom je uchov\u00e1vanie spr\u00e1v na op\u00e4tovn\u00e9 odoslanie v pam\u00e4ti aplik\u00e1cie. To znamen\u00e1, \u017ee re\u0161tart aplik\u00e1cie kv\u00f4li aktualiz\u00e1cii alebo chybe po\u010das prev\u00e1dzky povedie k strate v\u0161etk\u00fdch spr\u00e1v \u010dakaj\u00facich na op\u00e4tovn\u00e9 spracovanie. Ke\u010f\u017ee tento bod je kritick\u00fd pre n\u00e1\u0161 syst\u00e9m, \u010falej sme sa n\u00edm nezaoberali.<\/p>\n

<\/p>\n

Sama spring-kafka poskytuje nieko\u013eko realiz\u00e1ci\u00ed ContainerAwareErrorHandler, ako napr\u00edklad SeekToCurrentErrorHandler<\/a><\/noindex>, pomocou ktor\u00e9ho je mo\u017en\u00e9 spracova\u0165 spr\u00e1vu nesk\u00f4r bez zmeny offsetu v pr\u00edpade chyby. Od verzie spring-kafka 2.3 existuje mo\u017enos\u0165 nastavi\u0165 BackOffPolicy.<\/p>\n

<\/p>\n

Tento pr\u00edstup umo\u017e\u0148uje op\u00e4tovne spracov\u00e1van\u00fdm spr\u00e1vam pre\u017ei\u0165 re\u0161tart aplik\u00e1cie, ale mechanizmus DLQ st\u00e1le ch\u00fdba. Tento variant sme si vybrali za\u010diatkom roka 2019, optimisticky sme predpokladali, \u017ee DLQ nebude potrebn\u00fd (mali sme \u0161\u0165astie a naozaj nebol potrebn\u00fd po\u010das nieko\u013ek\u00fdch mesiacov prev\u00e1dzky aplik\u00e1cie s t\u00fdmto syst\u00e9mom op\u00e4tovn\u00e9ho spracovania). Do\u010dasn\u00e9 chyby viedli k aktiv\u00e1cii SeekToCurrentErrorHandler. Ostatn\u00e9 chyby sa zapisovali do logu, viedli k posunu offsetu a spracovanie pokra\u010dovalo s nasleduj\u00facou spr\u00e1vou.<\/p>\n

<\/p>\n

Kone\u010dn\u00e9 rie\u0161enie<\/h2>\n

<\/p>\n

Implement\u00e1cia zalo\u017een\u00e1 na SeekToCurrentErrorHandler n\u00e1s podnietila k v\u00fdvoju vlastn\u00e9ho mechanizmu na opakovan\u00e9 odosielanie spr\u00e1v.<\/p>\n

<\/p>\n

Predov\u0161etk\u00fdm sme chceli vyu\u017ei\u0165 u\u017e existuj\u00face sk\u00fasenosti a roz\u0161\u00edri\u0165 ich v z\u00e1vislosti od logiky aplik\u00e1cie. Pre aplik\u00e1ciu s line\u00e1rnou logikou by bolo optim\u00e1lne na kr\u00e1tky \u010das zastavi\u0165 \u010d\u00edtanie nov\u00fdch spr\u00e1v, ktor\u00fd je ur\u010den\u00fd v r\u00e1mci strat\u00e9gie opakovan\u00fdch volan\u00ed. Pre ostatn\u00e9 aplik\u00e1cie sme chceli ma\u0165 jedin\u00fd bod, ktor\u00fd by zabezpe\u010doval vykon\u00e1vanie strat\u00e9gie opakovan\u00fdch volan\u00ed. Okrem toho by tento jedin\u00fd bod mal ma\u0165 funk\u010dnos\u0165 DLQ pre oba pr\u00edstupy.<\/p>\n

<\/p>\n

Samotn\u00e1 strat\u00e9gia opakovan\u00fdch volan\u00ed by mala by\u0165 ulo\u017een\u00e1 v aplik\u00e1cii, ktor\u00e1 je zodpovedn\u00e1 za z\u00edska\u0165 \u010fal\u0161\u00ed interval pri v\u00fdskyte do\u010dasnej chyby.<\/p>\n

<\/p>\n

Zastavenie Consumer'a pre aplik\u00e1ciu s line\u00e1rnou logikou<\/h3>\n

<\/p>\n

Pri pr\u00e1ci so spring-kafka m\u00f4\u017ee k\u00f3d na zastavenie Consumer'a vyzera\u0165 pribli\u017ene takto:<\/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
V pr\u00edklade je retryAt \u010das, kedy je potrebn\u00e9 znova spusti\u0165 MessageListenerContainer, ak e\u0161te funguje. Op\u00e4tovn\u00e9 spustenie sa uskuto\u010dn\u00ed v samostatnom vl\u00e1kne, ktor\u00e9 je spusten\u00e9 v TaskScheduler, ktor\u00e9ho implement\u00e1ciu tie\u017e poskytuje spring. <\/p>\n
<\/p>\n
Hodnotu retryAt nach\u00e1dzame nasleduj\u00facim sp\u00f4sobom:<\/p>\n
<\/p>\n
    \n
  1. H\u013ead\u00e1 sa hodnota po\u010d\u00edtadla opakovan\u00fdch volan\u00ed.<\/li>\n
  2. Pod\u013ea hodnoty po\u010d\u00edtadla sa h\u013ead\u00e1 aktu\u00e1lny interval oneskorenia v strat\u00e9gii opakovan\u00fdch volan\u00ed. Strat\u00e9gia je deklarovan\u00e1 priamo v aplik\u00e1cii, na jej uchovanie sme zvolili form\u00e1t JSON.<\/li>\n
  3. N\u00e1jden\u00fd v JSON-rozsahu interval obsahuje po\u010det sek\u00fand, po ktor\u00fdch je potrebn\u00e9 opakova\u0165 spracovanie. Tento po\u010det sek\u00fand sa prid\u00e1 k aktu\u00e1lnemu \u010dasu, \u010d\u00edm vznikne hodnota pre retryAt.<\/li>\n
  4. Ak interval nie je n\u00e1jden\u00fd, hodnota retryAt je null a spr\u00e1va bude odoslan\u00e1 do DLQ na manu\u00e1lne spracovanie.<\/li>\n<\/ol>\n
    <\/p>\n
    Pri tomto pr\u00edstupe zost\u00e1va iba zachova\u0165 po\u010det opakovan\u00fdch volan\u00ed pre ka\u017ed\u00fa spr\u00e1vu, ktor\u00e1 je moment\u00e1lne spracov\u00e1van\u00e1, napr\u00edklad v pam\u00e4ti aplik\u00e1cie. Uchov\u00e1vanie po\u010d\u00edtadla pokusov v pam\u00e4ti nie je pre tento pr\u00edstup kritick\u00e9, preto\u017ee aplik\u00e1cia s line\u00e1rnou logikou nem\u00f4\u017ee vykon\u00e1va\u0165 spracovanie celkovo. Na rozdiel od spring-retry re\u0161tartovanie aplik\u00e1cie nevedie k strate v\u0161etk\u00fdch spr\u00e1v na op\u00e4tovn\u00e9 spracovanie, ale jednoducho k re\u0161tartovaniu strat\u00e9gie. <\/p>\n
    <\/p>\n
    Tento pr\u00edstup pom\u00e1ha zn\u00ed\u017ei\u0165 z\u00e1\u0165a\u017e na extern\u00fd syst\u00e9m, ktor\u00fd m\u00f4\u017ee by\u0165 nedostupn\u00fd z d\u00f4vodu ve\u013emi ve\u013ekej z\u00e1\u0165a\u017ee. In\u00fdmi slovami, okrem op\u00e4tovn\u00e9ho spracovania sme dosiahli realiz\u00e1ciu vzoru circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    V na\u0161om pr\u00edpade prah chyby predstavuje iba hodnotu 1, a aby sme minimalizovali prestoj syst\u00e9mu z d\u00f4vodu do\u010dasnej sie\u0165ovej poruchy, pou\u017e\u00edvame ve\u013emi granularn\u00fa strat\u00e9giu opakovan\u00fdch volan\u00ed s mal\u00fdmi \u010dasov\u00fdmi intervalmi oneskorenia. To nemus\u00ed vyhovova\u0165 v\u0161etk\u00fdm aplik\u00e1ci\u00e1m skupiny, preto je potrebn\u00e9 prah chyby a ve\u013ekos\u0165 intervalu nastavi\u0165 na z\u00e1klade vlastnost\u00ed syst\u00e9mu.<\/p>\n
    <\/p>\n
    Samostatn\u00e1 aplik\u00e1cia na spracovanie spr\u00e1v z aplik\u00e1ci\u00ed s nedeterministickou logikou<\/h3>\n
    <\/p>\n
    Tu je pr\u00edklad k\u00f3du, ktor\u00fd posiela spr\u00e1vu do takejto aplik\u00e1cie (Retryer), ktor\u00e1 vykon\u00e1 opakovan\u00e9 odoslanie do topiku DESTINATION pri dosiahnut\u00ed \u010dasu 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
    Z pr\u00edkladu je vidie\u0165, \u017ee sa pren\u00e1\u0161a ve\u013ea inform\u00e1ci\u00ed v hlavi\u010dk\u00e1ch. Hodnota RETRY_AT sa nach\u00e1dza rovnako ako pre mechanizmus opakovan\u00e9ho pokusu prostredn\u00edctvom zastavenia Consumer\u2019a. Okrem DESTINATION a RETRY_AT pren\u00e1\u0161ame:<\/p>\n
    <\/p>\n