{"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\/cs\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Znovu zpracov\u00e1n\u00ed ud\u00e1lost\u00ed p\u0159ijat\u00fdch z Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Znovu<\/p>\n

<\/p>\n

Ahoj, Habr.<\/p>\n

<\/p>\n

Ned\u00e1vno jsem sd\u00edlel svou zku\u0161enost<\/a><\/noindex> o tom, jak\u00e9 parametry v na\u0161em t\u00fdmu nej\u010dast\u011bji pou\u017e\u00edv\u00e1me pro Kafka Producer a Consumer, abychom dos\u00e1hli zaru\u010den\u00e9 dod\u00e1vky. V tomto \u010dl\u00e1nku chci vysv\u011btlit, jak jsme zorganizovali op\u011btovn\u00e9 zpracov\u00e1n\u00ed ud\u00e1losti, kterou jsme obdr\u017eeli z Kafka, v d\u016fsledku do\u010dasn\u00e9 nedostupnosti extern\u00edho syst\u00e9mu.<\/p>\n

<\/p>\n

Modern\u00ed aplikace funguj\u00ed ve velmi slo\u017eit\u00e9m prost\u0159ed\u00ed. Obchodn\u00ed logika, zabalen\u00e1 v modern\u00edm technologick\u00e9m z\u00e1sobn\u00edku, pracuj\u00edc\u00ed v Dockeru, kter\u00fd je \u0159\u00edzen orchestr\u00e1torem jako Kubernetes nebo OpenShift, a komunikuj\u00edc\u00ed s jin\u00fdmi aplikacemi nebo podnikov\u00fdch \u0159e\u0161en\u00edmi prost\u0159ednictv\u00edm \u0159et\u011bzce fyzick\u00fdch a virtu\u00e1ln\u00edch sm\u011brova\u010d\u016f. V takov\u00e9m prost\u0159ed\u00ed se m\u016f\u017ee v\u017edy n\u011bco pokazit, proto je op\u011btovn\u00e9 zpracov\u00e1n\u00ed ud\u00e1lost\u00ed v p\u0159\u00edpad\u011b nedostupnosti jednoho z extern\u00edch syst\u00e9m\u016f d\u016fle\u017eitou sou\u010d\u00e1st\u00ed na\u0161ich obchodn\u00edch proces\u016f.<\/p>\n

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

Jak to bylo p\u0159ed Kafka<\/h2>\n

<\/p>\n

D\u0159\u00edve jsme v projektu pou\u017e\u00edvali IBM MQ pro asynchronn\u00ed doru\u010dov\u00e1n\u00ed zpr\u00e1v. Kdy\u017e nastala n\u011bjak\u00e1 chyba v pr\u016fb\u011bhu pr\u00e1ce slu\u017eby, obdr\u017een\u00e1 zpr\u00e1va mohla b\u00fdt um\u00edst\u011bna do dead-letter-queue (DLQ) pro dal\u0161\u00ed ru\u010dn\u00ed zpracov\u00e1n\u00ed. DLQ byla vytvo\u0159ena vedle p\u0159\u00edchoz\u00ed fronty, p\u0159enesen\u00ed zpr\u00e1vy prob\u00edhalo uvnit\u0159 IBM MQ. <\/p>\n

<\/p>\n

Pokud byla chyba do\u010dasn\u00e1 a mohli jsme to ur\u010dit (nap\u0159. ResourceAccessException p\u0159i HTTP vol\u00e1n\u00ed nebo MongoTimeoutException p\u0159i dotazu v MongoDb), pak byla aktivov\u00e1na strategie opakovan\u00fdch vol\u00e1n\u00ed. Nez\u00e1visle na v\u011btven\u00ed logiky aplikace byla p\u016fvodn\u00ed zpr\u00e1va um\u00edst\u011bna bu\u010f do syst\u00e9mov\u00e9 fronty pro odlo\u017een\u00e9 odesl\u00e1n\u00ed, nebo do samostatn\u00e9 aplikace, kter\u00e1 byla kdysi vytvo\u0159ena pro op\u011btovn\u00e9 odes\u00edl\u00e1n\u00ed zpr\u00e1v. P\u0159itom je do z\u00e1hlav\u00ed zpr\u00e1vy zaps\u00e1no \u010d\u00edslo opakovan\u00e9ho odesl\u00e1n\u00ed, kter\u00e9 je v\u00e1z\u00e1no na interval zpo\u017ed\u011bn\u00ed nebo na konec strategie na \u00farovni aplikace. Pokud jsme dos\u00e1hli konce strategie, ale extern\u00ed syst\u00e9m byl st\u00e1le nedostupn\u00fd, zpr\u00e1va bude um\u00edst\u011bna do DLQ pro ru\u010dn\u00ed zpracov\u00e1n\u00ed.<\/p>\n

<\/p>\n

Hled\u00e1n\u00ed \u0159e\u0161en\u00ed<\/h2>\n

<\/p>\n

Pokud hled\u00e1me na internetu<\/a><\/noindex>, m\u016f\u017eeme naj\u00edt n\u00e1sleduj\u00edc\u00ed rozhodnut\u00ed<\/a><\/noindex>. Stru\u010dn\u011b \u0159e\u010deno, navrhuje se vytvo\u0159it jednu frontu pro ka\u017ed\u00fd interval zpo\u017ed\u011bn\u00ed a realizovat na stran\u011b aplikace Consumer\u2019y, kter\u00e9 budou \u010d\u00edst zpr\u00e1vy s pot\u0159ebn\u00fdm zpo\u017ed\u011bn\u00edm. <\/p>\n

<\/p>\n

\"Znovu<\/p>\n

<\/p>\n

Navzdory velk\u00e9mu po\u010dtu pozitivn\u00edch hodnocen\u00ed se mi zd\u00e1 pon\u011bkud ne\u00fasp\u011b\u0161n\u00e9. P\u0159edev\u0161\u00edm proto, \u017ee v\u00fdvoj\u00e1\u0159i, krom\u011b realizace obchodn\u00edch po\u017eadavk\u016f, budou muset v\u011bnovat hodn\u011b \u010dasu realizaci popsan\u00e9ho mechanismu.<\/p>\n

<\/p>\n

Krom\u011b toho, pokud je na Kafka clusteru povoleno \u0159\u00edzen\u00ed p\u0159\u00edstupu, bude nutn\u00e9 str\u00e1vit n\u011bjak\u00fd \u010das vytv\u00e1\u0159en\u00edm t\u00e9mat a zaji\u0161\u0165ov\u00e1n\u00edm pot\u0159ebn\u00fdch p\u0159\u00edstup\u016f k nim. Krom\u011b toho bude t\u0159eba vybrat spr\u00e1vn\u00fd parametr retention.ms pro ka\u017ed\u00e9 z retry t\u00e9mat, aby zpr\u00e1vy byly odes\u00edl\u00e1ny znovu a nezmizely. Realizaci a \u017e\u00e1dost o p\u0159\u00edstupy bude nutn\u00e9 opakovat pro ka\u017ed\u00fd existuj\u00edc\u00ed nebo nov\u00fd slu\u017ebu.<\/p>\n

<\/p>\n

Nyn\u00ed se pod\u00edvejme, jak\u00e9 mechanismy pro op\u011btovn\u00e9 zpracov\u00e1n\u00ed zpr\u00e1v n\u00e1m spring celkov\u011b a spring-kafka konkr\u00e9tn\u011b poskytuj\u00ed. Spring-kafka m\u00e1 tranzitivn\u00ed z\u00e1vislost na spring-retry, kter\u00fd poskytuje abstrakce pro spr\u00e1vu r\u016fzn\u00fdch BackOffPolicy. Toto je pom\u011brn\u011b flexibiln\u00ed n\u00e1stroj, ale jeho v\u00fdznamnou nev\u00fdhodou je uchov\u00e1v\u00e1n\u00ed zpr\u00e1v pro op\u011btovn\u00e9 odesl\u00e1n\u00ed v pam\u011bti aplikace. To znamen\u00e1, \u017ee restart aplikace v d\u016fsledku aktualizace nebo chyby b\u011bhem provozu povede ke ztr\u00e1t\u011b v\u0161ech zpr\u00e1v, kter\u00e9 \u010dekaj\u00ed na op\u011btovn\u00e9 zpracov\u00e1n\u00ed. Vzhledem k tomu, \u017ee tento bod je kritick\u00fd pro n\u00e1\u0161 syst\u00e9m, rozhodli jsme se d\u00e1le ho nezva\u017eovat.<\/p>\n

<\/p>\n

Samotn\u00e1 spring-kafka poskytuje n\u011bkolik implementac\u00ed ContainerAwareErrorHandler, nap\u0159\u00edklad SeekToCurrentErrorHandler<\/a><\/noindex>, s jeho\u017e pomoc\u00ed lze zpracovat zpr\u00e1vu pozd\u011bji, ani\u017e by do\u0161lo k posunut\u00ed offsetu v p\u0159\u00edpad\u011b chyby. Od verze spring-kafka 2.3 je mo\u017en\u00e9 nastavit BackOffPolicy.<\/p>\n

<\/p>\n

Tento p\u0159\u00edstup umo\u017e\u0148uje op\u011btovn\u011b zpracov\u00e1van\u00fdm zpr\u00e1v\u00e1m p\u0159e\u017e\u00edt restart aplikace, ale mechanismus DLQ st\u00e1le chyb\u00ed. Tento variant jsme zvolili na za\u010d\u00e1tku roku 2019, optimisticky p\u0159itom p\u0159edpokl\u00e1dali, \u017ee DLQ nebude pot\u0159eba (m\u011bli jsme \u0161t\u011bst\u00ed a skute\u010dn\u011b nebyl pot\u0159eba b\u011bhem n\u011bkolika m\u011bs\u00edc\u016f provozu aplikace s t\u00edmto syst\u00e9mem op\u011btovn\u00e9ho zpracov\u00e1n\u00ed). Do\u010dasn\u00e9 chyby vedly k aktivaci SeekToCurrentErrorHandler. Ostatn\u00ed chyby byly vyti\u0161t\u011bny do logu, co\u017e vedlo k posunu offsetu, a zpracov\u00e1n\u00ed pokra\u010dovalo s dal\u0161\u00ed zpr\u00e1vou.<\/p>\n

<\/p>\n

Kone\u010dn\u00e9 \u0159e\u0161en\u00ed<\/h2>\n

<\/p>\n

Implementace zalo\u017een\u00e1 na SeekToCurrentErrorHandler n\u00e1s p\u0159im\u011bla vyvinout vlastn\u00ed mechanismus pro opakovan\u00e9 odes\u00edl\u00e1n\u00ed zpr\u00e1v.<\/p>\n

<\/p>\n

P\u0159edev\u0161\u00edm jsme cht\u011bli vyu\u017e\u00edt ji\u017e existuj\u00edc\u00ed zku\u0161enosti a roz\u0161\u00ed\u0159it je v z\u00e1vislosti na logice aplikace. Pro aplikaci s line\u00e1rn\u00ed logikou by bylo optim\u00e1ln\u00ed na kr\u00e1tkou dobu pozastavit p\u0159\u00edjem nov\u00fdch zpr\u00e1v, co\u017e by bylo ur\u010deno v r\u00e1mci strategie opakovan\u00fdch vol\u00e1n\u00ed. Pro ostatn\u00ed aplikace bylo \u017e\u00e1douc\u00ed m\u00edt jednotn\u00e9 m\u00edsto, kter\u00e9 by zaji\u0161\u0165ovalo prov\u00e1d\u011bn\u00ed strategie opakovan\u00fdch vol\u00e1n\u00ed. D\u00e1le by toto jednotn\u00e9 m\u00edsto m\u011blo m\u00edt funk\u010dnost DLQ pro oba p\u0159\u00edstupy.<\/p>\n

<\/p>\n

Samotn\u00e1 strategie opakovan\u00fdch vol\u00e1n\u00ed by m\u011bla b\u00fdt ulo\u017eena v aplikaci, kter\u00e1 je zodpov\u011bdn\u00e1 za z\u00edsk\u00e1n\u00ed dal\u0161\u00edho intervalu p\u0159i v\u00fdskytu do\u010dasn\u00e9 chyby.<\/p>\n

<\/p>\n

Pozastaven\u00ed Consumer\u2019a pro aplikaci s line\u00e1rn\u00ed logikou<\/h3>\n

<\/p>\n

P\u0159i pr\u00e1ci s spring-kafka m\u016f\u017ee k\u00f3d pro zastaven\u00ed Consumer\u2019a vypadat p\u0159ibli\u017en\u011b 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 p\u0159\u00edkladu retryAt je \u010das, kdy je t\u0159eba znovu spustit MessageListenerContainer, pokud st\u00e1le b\u011b\u017e\u00ed. Op\u011btovn\u00e9 spu\u0161t\u011bn\u00ed prob\u011bhne v samostatn\u00e9m vl\u00e1kn\u011b spu\u0161t\u011bn\u00e9m v TaskScheduler, jeho\u017e implementaci tak\u00e9 poskytuje spring. <\/p>\n
<\/p>\n
Hodnotu retryAt nalezneme n\u00e1sleduj\u00edc\u00edm zp\u016fsobem:<\/p>\n
<\/p>\n
    \n
  1. Hled\u00e1 se hodnota po\u010ditadla opakov\u00e1n\u00ed.<\/li>\n
  2. Podle hodnoty po\u010ditadla se hled\u00e1 aktu\u00e1ln\u00ed interval zpo\u017ed\u011bn\u00ed ve strategii opakovan\u00fdch vol\u00e1n\u00ed. Strategie je vyhl\u00e1\u0161ena v samotn\u00e9 aplikaci, k jej\u00edmu ulo\u017een\u00ed jsme zvolili form\u00e1t JSON.<\/li>\n
  3. Nalezen\u00fd interval v JSON poli obsahuje po\u010det sekund, po kter\u00fdch je nutn\u00e9 zpracov\u00e1n\u00ed zopakovat. Tento po\u010det sekund se p\u0159i\u010dte k aktu\u00e1ln\u00edmu \u010dasu, co\u017e vytvo\u0159\u00ed hodnotu pro retryAt.<\/li>\n
  4. Pokud interval nen\u00ed nalezen, hodnota retryAt je null a zpr\u00e1va bude odesl\u00e1na do DLQ k ru\u010dn\u00edmu zpracov\u00e1n\u00ed.<\/li>\n<\/ol>\n
    <\/p>\n
    P\u0159i tomto p\u0159\u00edstupu je nutn\u00e9 uchov\u00e1vat pouze po\u010det opakovan\u00fdch pokus\u016f pro ka\u017edou zpr\u00e1vu, kter\u00e1 je aktu\u00e1ln\u011b zpracov\u00e1v\u00e1na, nap\u0159\u00edklad v pam\u011bti aplikace. Uchov\u00e1v\u00e1n\u00ed po\u010dtu pokus\u016f v pam\u011bti nen\u00ed pro tento p\u0159\u00edstup kritick\u00e9, proto\u017ee aplikace s line\u00e1rn\u00ed logikou nem\u016f\u017ee prov\u00e1d\u011bt zpracov\u00e1n\u00ed v celku. Na rozd\u00edl od spring-retry restart aplikace nevede ke ztr\u00e1t\u011b v\u0161ech zpr\u00e1v pro op\u011btovn\u00e9 zpracov\u00e1n\u00ed, ale pouze k restartu strategie. <\/p>\n
    <\/p>\n
    Tento p\u0159\u00edstup pom\u00e1h\u00e1 sn\u00ed\u017eit z\u00e1t\u011b\u017e na extern\u00ed syst\u00e9m, kter\u00fd nemus\u00ed b\u00fdt k dispozici kv\u016fli velmi vysok\u00e9 z\u00e1t\u011b\u017ei. Jin\u00fdmi slovy, krom\u011b op\u011btovn\u00e9ho zpracov\u00e1n\u00ed jsme dos\u00e1hli realizace vzoru. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    V na\u0161em p\u0159\u00edpad\u011b je prahem chyby pouze 1 a abychom minimalizovali prostoje syst\u00e9mu kv\u016fli do\u010dasn\u00e9mu s\u00ed\u0165ov\u00e9mu p\u0159eru\u0161en\u00ed, pou\u017e\u00edv\u00e1me velmi granularn\u00ed strategii opakovan\u00fdch pokus\u016f s mal\u00fdmi zpo\u017ed\u011bn\u00edmi. To nemus\u00ed vyhovovat v\u0161em aplikac\u00edm skupiny spole\u010dnost\u00ed, proto je t\u0159eba pom\u011br mezi prahem chyby a velikost\u00ed intervalu p\u0159izp\u016fsobit podle vlastnost\u00ed syst\u00e9mu.<\/p>\n
    <\/p>\n
    Samostatn\u00e1 aplikace pro zpracov\u00e1n\u00ed zpr\u00e1v z aplikac\u00ed s nedeterministickou logikou.<\/h3>\n
    <\/p>\n
    Zde je p\u0159\u00edklad k\u00f3du, kter\u00fd odes\u00edl\u00e1 zpr\u00e1vu do takov\u00e9 aplikace (Retryer), kter\u00e1 provede op\u011btovn\u00e9 odesl\u00e1n\u00ed do topiku DESTINATION p\u0159i dosa\u017een\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 p\u0159\u00edkladu je patrn\u00e9, \u017ee se do header\u016f p\u0159ed\u00e1v\u00e1 mnoho informac\u00ed. Hodnota RETRY_AT se nach\u00e1z\u00ed stejn\u011b jako pro mechanismus opakov\u00e1n\u00ed pomoc\u00ed zastaven\u00ed Consumer\u2019a. Krom\u011b DESTINATION a RETRY_AT p\u0159ed\u00e1v\u00e1me:<\/p>\n
    <\/p>\n