{"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\/pl\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Powt\u00f3rne przetwarzanie zdarze\u0144 otrzymanych z Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Powt\u00f3rne<\/p>\n

<\/p>\n

Cze\u015b\u0107, Habr.<\/p>\n

<\/p>\n

Niedawno podzieli\u0142em si\u0119 do\u015bwiadczeniem na temat parametr\u00f3w, kt\u00f3re nasz zesp\u00f3\u0142 najcz\u0119\u015bciej wykorzystuje dla Kafka Producer i Consumer, aby osi\u0105gn\u0105\u0107 gwarantowan\u0105 dostaw\u0119. W tym artykule chcia\u0142bym opowiedzie\u0107, jak zorganizowali\u015bmy ponowne przetwarzanie zdarzenia otrzymanego z Kafka w wyniku tymczasowej niedost\u0119pno\u015bci zewn\u0119trznego systemu.<\/a><\/noindex> Nowoczesne aplikacje dzia\u0142aj\u0105 w bardzo z\u0142o\u017conym \u015brodowisku. Logika biznesowa, otoczona nowoczesnym stosem technologicznym, dzia\u0142aj\u0105ca w obrazie Docker, zarz\u0105dzanym przez orkiestratora takiego jak Kubernetes lub OpenShift, i komunikuj\u0105ca si\u0119 z innymi aplikacjami lub rozwi\u0105zaniami enterprise poprzez \u0142a\u0144cuch fizycznych i wirtualnych router\u00f3w. W takim otoczeniu zawsze co\u015b mo\u017ce si\u0119 zepsu\u0107, dlatego ponowne przetwarzanie zdarze\u0144 w przypadku niedost\u0119pno\u015bci jednego z zewn\u0119trznych system\u00f3w jest wa\u017cn\u0105 cz\u0119\u015bci\u0105 naszych proces\u00f3w biznesowych.<\/p>\n

<\/p>\n

Jak to wygl\u0105da\u0142o przed Kafka<\/p>\n

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

Wcze\u015bniej w projekcie korzystali\u015bmy z IBM MQ do asynchronicznej dostawy wiadomo\u015bci. W przypadku wyst\u0105pienia jakiegokolwiek b\u0142\u0119du w trakcie pracy serwisu otrzymana wiadomo\u015b\u0107 mog\u0142a zosta\u0107 umieszczona w kolejce dead-letter (DLQ) do dalszej r\u0119cznej analizy. DLQ by\u0142 tworzony obok kolejki przychodz\u0105cej, a przeniesienie wiadomo\u015bci nast\u0119powa\u0142o wewn\u0105trz IBM MQ.<\/h2>\n

<\/p>\n

Je\u017celi b\u0142\u0105d mia\u0142 charakter tymczasowy i mogli\u015bmy to okre\u015bli\u0107 (na przyk\u0142ad, ResourceAccessException podczas wywo\u0142ania HTTP lub MongoTimeoutException podczas zapytania do MongoDb), w\u00f3wczas wchodzi\u0142a w \u017cycie strategia powt\u00f3rze\u0144. Niezale\u017cnie od rozga\u0142\u0119zienia logiki aplikacji, pierwotna wiadomo\u015b\u0107 by\u0142a przenoszona albo do kolejki systemowej do op\u00f3\u017anionej wysy\u0142ki, albo do oddzielnej aplikacji, kt\u00f3ra kiedy\u015b zosta\u0142a stworzona do ponownej wysy\u0142ki wiadomo\u015bci. W tym przypadku w nag\u0142\u00f3wku wiadomo\u015bci zapisywano numer powt\u00f3rzenia, kt\u00f3ry by\u0142 powi\u0105zany z interwa\u0142em op\u00f3\u017anienia lub ko\u0144cem strategii na poziomie aplikacji. Je\u015bli osi\u0105gn\u0119li\u015bmy koniec strategii, a zewn\u0119trzny system wci\u0105\u017c by\u0142 niedost\u0119pny, wiadomo\u015b\u0107 zostanie umieszczona w DLQ do r\u0119cznej analizy. <\/p>\n

<\/p>\n

Szukaj\u0105c rozwi\u0105zania<\/p>\n

<\/p>\n

Szukaj\u0105c w internecie<\/h2>\n

<\/p>\n

, mo\u017cna znale\u017a\u0107 nast\u0119puj\u0105ce<\/a><\/noindex>. W skr\u00f3cie, proponuje si\u0119 za\u0142o\u017cenie po jednym temacie dla ka\u017cdego interwa\u0142u op\u00f3\u017anienia i wdro\u017cenie po stronie aplikacji Consumer\u00f3w, kt\u00f3rzy b\u0119d\u0105 odczytywa\u0107 wiadomo\u015bci z odpowiednim op\u00f3\u017anieniem. decyzj\u0119<\/a><\/noindex>. M\u00f3wi\u0105c kr\u00f3tko, proponuje si\u0119 utworzenie tematu dla ka\u017cdego interwa\u0142u op\u00f3\u017anienia i wdro\u017cenie po stronie aplikacji konsument\u00f3w, kt\u00f3rzy b\u0119d\u0105 odczytywa\u0107 wiadomo\u015bci z odpowiednim op\u00f3\u017anieniem. <\/p>\n

<\/p>\n

\"Powt\u00f3rne<\/p>\n

<\/p>\n

Mimo du\u017cej liczby pozytywnych opinii, wydaje si\u0119 mi ono nie do ko\u0144ca udane. Przede wszystkim dlatego, \u017ce deweloper, opr\u00f3cz realizacji wymaga\u0144 biznesowych, b\u0119dzie musia\u0142 po\u015bwi\u0119ci\u0107 du\u017co czasu na wdro\u017cenie opisanego mechanizmu.<\/p>\n

<\/p>\n

Ponadto, je\u015bli w klastrze Kafka w\u0142\u0105czone jest zarz\u0105dzanie dost\u0119pem, trzeba b\u0119dzie po\u015bwi\u0119ci\u0107 troch\u0119 czasu na tworzenie temat\u00f3w i zapewnienie odpowiednich dost\u0119p\u00f3w do nich. Dodatkowo, trzeba b\u0119dzie dobra\u0107 odpowiedni parametr retention.ms dla ka\u017cdego z temat\u00f3w retry, aby wiadomo\u015bci mog\u0142y by\u0107 ponownie wysy\u0142ane i nie znika\u0142y. Wdra\u017canie i \u017c\u0105danie dost\u0119p\u00f3w b\u0119dzie trzeba powt\u00f3rzy\u0107 dla ka\u017cdej istniej\u0105cej lub nowej us\u0142ugi.<\/p>\n

<\/p>\n

Przyjrzyjmy si\u0119 teraz, jakie mechanizmy ponownego przetwarzania wiadomo\u015bci oferuje nam Spring jako ca\u0142o\u015b\u0107 i Spring-Kafka w szczeg\u00f3lno\u015bci. Spring-Kafka ma transytywn\u0105 zale\u017cno\u015b\u0107 od Spring-Retry, kt\u00f3ry zapewnia abstrakcje do zarz\u0105dzania r\u00f3\u017cnymi BackOffPolicy. To do\u015b\u0107 elastyczne narz\u0119dzie, ale jego powa\u017cnym minusem jest przechowywanie wiadomo\u015bci do ponownego wys\u0142ania w pami\u0119ci aplikacji. Oznacza to, \u017ce ponowne uruchomienie aplikacji z powodu aktualizacji lub b\u0142\u0119du w trakcie eksploatacji spowoduje utrat\u0119 wszystkich wiadomo\u015bci oczekuj\u0105cych na ponowne przetworzenie. Poniewa\u017c ten punkt jest krytyczny dla naszego systemu, nie rozwa\u017cali\u015bmy go dalej.<\/p>\n

<\/p>\n

Sama Spring-Kafka oferuje kilka implementacji ContainerAwareErrorHandler, na przyk\u0142ad SeekToCurrentErrorHandler<\/a><\/noindex>, kt\u00f3ry pozwala na przetworzenie wiadomo\u015bci p\u00f3\u017aniej bez przesuwania offsetu w przypadku wyst\u0105pienia b\u0142\u0119du. Od wersji Spring-Kafka 2.3 pojawi\u0142a si\u0119 mo\u017cliwo\u015b\u0107 ustalania BackOffPolicy.<\/p>\n

<\/p>\n

To podej\u015bcie pozwala wiadomo\u015bciom do ponownego przetwarzania na przetrwanie restartu aplikacji, ale mechanizm DLQ wci\u0105\u017c jest niedost\u0119pny. W\u0142a\u015bnie ten wariant wybrali\u015bmy na pocz\u0105tku 2019 roku, optymistycznie zak\u0142adaj\u0105c, \u017ce DLQ nie b\u0119dzie potrzebny (mieli\u015bmy szcz\u0119\u015bcie i rzeczywi\u015bcie nie by\u0142 potrzebny przez kilka miesi\u0119cy eksploatacji aplikacji z takim systemem ponownego przetwarzania). B\u0142\u0119dy tymczasowe prowadzi\u0142y do uruchomienia SeekToCurrentErrorHandler. Pozosta\u0142e b\u0142\u0119dy by\u0142y rejestrowane w logach, prowadzi\u0142y do przesuni\u0119cia offsetu, a przetwarzanie by\u0142o kontynuowane z nast\u0119pn\u0105 wiadomo\u015bci\u0105.<\/p>\n

<\/p>\n

Ostateczne rozwi\u0105zanie<\/h2>\n

<\/p>\n

Realizacja oparta na SeekToCurrentErrorHandler sk\u0142oni\u0142a nas do opracowania w\u0142asnego mechanizmu do ponownego wysy\u0142ania wiadomo\u015bci.<\/p>\n

<\/p>\n

Przede wszystkim chcieli\u015bmy wykorzysta\u0107 dotychczasowe do\u015bwiadczenia i rozszerzy\u0107 je w zale\u017cno\u015bci od logiki aplikacji. Dla aplikacji o liniowej logice optymalne by\u0142oby wstrzymanie odczytu nowych wiadomo\u015bci na kr\u00f3tki czas okre\u015blony w strategii ponownych wywo\u0142a\u0144. Dla pozosta\u0142ych aplikacji chcieliby\u015bmy mie\u0107 jedn\u0105 wsp\u00f3ln\u0105 punkt, kt\u00f3ry zapewni realizacj\u0119 strategii ponownych wywo\u0142a\u0144. Dodatkowo ten wsp\u00f3lny punkt powinien mie\u0107 funkcjonalno\u015b\u0107 DLQ dla obu podej\u015b\u0107.<\/p>\n

<\/p>\n

Strategia ponownych wywo\u0142a\u0144 powinna by\u0107 przechowywana w aplikacji, kt\u00f3ra odpowiada za uzyskanie nast\u0119pnego interwa\u0142u w przypadku wyst\u0105pienia b\u0142\u0119du czasowego.<\/p>\n

<\/p>\n

Zatrzymanie Consumer\u2019a dla aplikacji o liniowej logice<\/h3>\n

<\/p>\n

Przy pracy z spring-kafka kod do zatrzymania Consumer\u2019a mo\u017ce wygl\u0105da\u0107 mniej wi\u0119cej tak:<\/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        \/\/ do DLQ\n    }<\/code><\/pre>\n
<\/p>\n
W przyk\u0142adzie retryAt to czas, w kt\u00f3rym nale\u017cy ponownie uruchomi\u0107 MessageListenerContainer, je\u015bli nadal dzia\u0142a. Ponowne uruchomienie nast\u0105pi w osobnym w\u0105tku uruchomionym w TaskScheduler, kt\u00f3rego realizacj\u0119 r\u00f3wnie\u017c zapewnia spring. <\/p>\n
<\/p>\n
Warto\u015b\u0107 retryAt znajduje si\u0119 w nast\u0119puj\u0105cy spos\u00f3b:<\/p>\n
<\/p>\n
    \n
  1. Szukane jest warto\u015b\u0107 licznika ponownych wywo\u0142a\u0144.<\/li>\n
  2. Zgodnie z warto\u015bci\u0105 licznika szukany jest bie\u017c\u0105cy interwa\u0142 op\u00f3\u017anienia w strategii ponownych wywo\u0142a\u0144. Strategia jest zadeklarowana w samej aplikacji, do jej przechowywania wybrali\u015bmy format JSON.<\/li>\n
  3. Znaleziony w tablicy JSON interwa\u0142 zawiera liczb\u0119 sekund, po kt\u00f3rej nale\u017cy powt\u00f3rzy\u0107 przetwarzanie. Ta liczba sekund jest dodawana do bie\u017c\u0105cego czasu, tworz\u0105c warto\u015b\u0107 dla retryAt.<\/li>\n
  4. Je\u015bli interwa\u0142 nie zostanie znaleziony, to warto\u015b\u0107 retryAt wynosi null, a wiadomo\u015b\u0107 zostanie wys\u0142ana do DLQ do r\u0119cznej analizy.<\/li>\n<\/ol>\n
    <\/p>\n
    W takim podej\u015bciu pozostaje tylko zachowa\u0107 liczb\u0119 ponownych wywo\u0142a\u0144 dla ka\u017cdej wiadomo\u015bci, kt\u00f3ra jest obecnie przetwarzana, na przyk\u0142ad w pami\u0119ci aplikacji. Zachowanie licznika pr\u00f3b w pami\u0119ci nie jest krytyczne dla tego podej\u015bcia, poniewa\u017c aplikacja z liniow\u0105 logik\u0105 nie mo\u017ce prowadzi\u0107 przetwarzania w ca\u0142o\u015bci. W przeciwie\u0144stwie do spring-retry, ponowne uruchomienie aplikacji nie prowadzi do utraty wszystkich wiadomo\u015bci do ponownego przetworzenia, a jedynie do ponownego uruchomienia strategii. <\/p>\n
    <\/p>\n
    To podej\u015bcie pomaga zredukowa\u0107 obci\u0105\u017cenie zewn\u0119trznego systemu, kt\u00f3ry mo\u017ce by\u0107 niedost\u0119pny z powodu bardzo wysokiego obci\u0105\u017cenia. Innymi s\u0142owy, opr\u00f3cz ponownego przetwarzania, uda\u0142o nam si\u0119 wdro\u017cy\u0107 wzorzec. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    W naszym przypadku pr\u00f3g b\u0142\u0119du wynosi tylko 1, a aby zminimalizowa\u0107 przestoje systemu spowodowane tymczasow\u0105 awari\u0105 sieci, stosujemy bardzo granularn\u0105 strategi\u0119 ponownych wywo\u0142a\u0144 z ma\u0142ymi interwa\u0142ami op\u00f3\u017anienia. Mo\u017ce to nie pasowa\u0107 do wszystkich aplikacji w grupie firm, dlatego proporcje mi\u0119dzy progiem b\u0142\u0119du a wielko\u015bci\u0105 interwa\u0142u nale\u017cy dobiera\u0107 w zale\u017cno\u015bci od cech systemu.<\/p>\n
    <\/p>\n
    Osobna aplikacja do przetwarzania wiadomo\u015bci z aplikacji o niedeterministycznej logice.<\/h3>\n
    <\/p>\n
    Oto przyk\u0142ad kodu, kt\u00f3ry wysy\u0142a wiadomo\u015b\u0107 do takiej aplikacji (Retryer), kt\u00f3ra wykona ponown\u0105 wysy\u0142k\u0119 do tematu DESTINATION po osi\u0105gni\u0119ciu czasu 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 przyk\u0142adu wida\u0107, \u017ce wiele informacji przekazywanych jest w nag\u0142\u00f3wkach. Warto\u015b\u0107 RETRY_AT znajduje si\u0119 tak samo, jak w mechanizmie powt\u00f3rze\u0144 poprzez zatrzymanie Consumer\u2019a. Opr\u00f3cz DESTINATION i RETRY_AT przekazujemy:<\/p>\n
    <\/p>\n