Cześć, Habr.
Niedawno podzieliłem się doświadczeniem Nowoczesne aplikacje działają w bardzo złożonym środowisku. Logika biznesowa, otoczona nowoczesnym stosem technologicznym, działająca w obrazie Docker, zarządzanym przez orkiestratora takiego jak Kubernetes lub OpenShift, i komunikująca się z innymi aplikacjami lub rozwiązaniami enterprise poprzez łańcuch fizycznych i wirtualnych routerów. W takim otoczeniu zawsze coś może się zepsuć, dlatego ponowne przetwarzanie zdarzeń w przypadku niedostępności jednego z zewnętrznych systemów jest ważną częścią naszych procesów biznesowych.
Jak to wyglądało przed Kafka
Wcześniej w projekcie korzystaliśmy z IBM MQ do asynchronicznej dostawy wiadomości. W przypadku wystąpienia jakiegokolwiek błędu w trakcie pracy serwisu otrzymana wiadomość mogła zostać umieszczona w kolejce dead-letter (DLQ) do dalszej ręcznej analizy. DLQ był tworzony obok kolejki przychodzącej, a przeniesienie wiadomości następowało wewnątrz IBM MQ.
Jeżeli błąd miał charakter tymczasowy i mogliśmy to określić (na przykład, ResourceAccessException podczas wywołania HTTP lub MongoTimeoutException podczas zapytania do MongoDb), wówczas wchodziła w życie strategia powtórzeń. Niezależnie od rozgałęzienia logiki aplikacji, pierwotna wiadomość była przenoszona albo do kolejki systemowej do opóźnionej wysyłki, albo do oddzielnej aplikacji, która kiedyś została stworzona do ponownej wysyłki wiadomości. W tym przypadku w nagłówku wiadomości zapisywano numer powtórzenia, który był powiązany z interwałem opóźnienia lub końcem strategii na poziomie aplikacji. Jeśli osiągnęliśmy koniec strategii, a zewnętrzny system wciąż był niedostępny, wiadomość zostanie umieszczona w DLQ do ręcznej analizy.
Szukając rozwiązania
Szukając w internecie
. W skrócie, proponuje się założenie po jednym temacie dla każdego interwału opóźnienia i wdrożenie po stronie aplikacji Consumerów, którzy będą odczytywać wiadomości z odpowiednim opóźnieniem. . Mówiąc krótko, proponuje się utworzenie tematu dla każdego interwału opóźnienia i wdrożenie po stronie aplikacji konsumentów, którzy będą odczytywać wiadomości z odpowiednim opóźnieniem.
Mimo dużej liczby pozytywnych opinii, wydaje się mi ono nie do końca udane. Przede wszystkim dlatego, że deweloper, oprócz realizacji wymagań biznesowych, będzie musiał poświęcić dużo czasu na wdrożenie opisanego mechanizmu.
Ponadto, jeśli w klastrze Kafka włączone jest zarządzanie dostępem, trzeba będzie poświęcić trochę czasu na tworzenie tematów i zapewnienie odpowiednich dostępów do nich. Dodatkowo, trzeba będzie dobrać odpowiedni parametr retention.ms dla każdego z tematów retry, aby wiadomości mogły być ponownie wysyłane i nie znikały. Wdrażanie i żądanie dostępów będzie trzeba powtórzyć dla każdej istniejącej lub nowej usługi.
Przyjrzyjmy się teraz, jakie mechanizmy ponownego przetwarzania wiadomości oferuje nam Spring jako całość i Spring-Kafka w szczególności. Spring-Kafka ma transytywną zależność od Spring-Retry, który zapewnia abstrakcje do zarządzania różnymi BackOffPolicy. To dość elastyczne narzędzie, ale jego poważnym minusem jest przechowywanie wiadomości do ponownego wysłania w pamięci aplikacji. Oznacza to, że ponowne uruchomienie aplikacji z powodu aktualizacji lub błędu w trakcie eksploatacji spowoduje utratę wszystkich wiadomości oczekujących na ponowne przetworzenie. Ponieważ ten punkt jest krytyczny dla naszego systemu, nie rozważaliśmy go dalej.
Sama Spring-Kafka oferuje kilka implementacji ContainerAwareErrorHandler, na przykład , który pozwala na przetworzenie wiadomości później bez przesuwania offsetu w przypadku wystąpienia błędu. Od wersji Spring-Kafka 2.3 pojawiła się możliwość ustalania BackOffPolicy.
To podejście pozwala wiadomościom do ponownego przetwarzania na przetrwanie restartu aplikacji, ale mechanizm DLQ wciąż jest niedostępny. Właśnie ten wariant wybraliśmy na początku 2019 roku, optymistycznie zakładając, że DLQ nie będzie potrzebny (mieliśmy szczęście i rzeczywiście nie był potrzebny przez kilka miesięcy eksploatacji aplikacji z takim systemem ponownego przetwarzania). Błędy tymczasowe prowadziły do uruchomienia SeekToCurrentErrorHandler. Pozostałe błędy były rejestrowane w logach, prowadziły do przesunięcia offsetu, a przetwarzanie było kontynuowane z następną wiadomością.
Ostateczne rozwiązanie
Realizacja oparta na SeekToCurrentErrorHandler skłoniła nas do opracowania własnego mechanizmu do ponownego wysyłania wiadomości.
Przede wszystkim chcieliśmy wykorzystać dotychczasowe doświadczenia i rozszerzyć je w zależności od logiki aplikacji. Dla aplikacji o liniowej logice optymalne byłoby wstrzymanie odczytu nowych wiadomości na krótki czas określony w strategii ponownych wywołań. Dla pozostałych aplikacji chcielibyśmy mieć jedną wspólną punkt, który zapewni realizację strategii ponownych wywołań. Dodatkowo ten wspólny punkt powinien mieć funkcjonalność DLQ dla obu podejść.
Strategia ponownych wywołań powinna być przechowywana w aplikacji, która odpowiada za uzyskanie następnego interwału w przypadku wystąpienia błędu czasowego.
Zatrzymanie Consumer’a dla aplikacji o liniowej logice
Przy pracy z spring-kafka kod do zatrzymania Consumer’a może wyglądać mniej więcej tak:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// do DLQ
}W przykładzie retryAt to czas, w którym należy ponownie uruchomić MessageListenerContainer, jeśli nadal działa. Ponowne uruchomienie nastąpi w osobnym wątku uruchomionym w TaskScheduler, którego realizację również zapewnia spring.
Wartość retryAt znajduje się w następujący sposób:
- Szukane jest wartość licznika ponownych wywołań.
- Zgodnie z wartością licznika szukany jest bieżący interwał opóźnienia w strategii ponownych wywołań. Strategia jest zadeklarowana w samej aplikacji, do jej przechowywania wybraliśmy format JSON.
- Znaleziony w tablicy JSON interwał zawiera liczbę sekund, po której należy powtórzyć przetwarzanie. Ta liczba sekund jest dodawana do bieżącego czasu, tworząc wartość dla retryAt.
- Jeśli interwał nie zostanie znaleziony, to wartość retryAt wynosi null, a wiadomość zostanie wysłana do DLQ do ręcznej analizy.
W takim podejściu pozostaje tylko zachować liczbę ponownych wywołań dla każdej wiadomości, która jest obecnie przetwarzana, na przykład w pamięci aplikacji. Zachowanie licznika prób w pamięci nie jest krytyczne dla tego podejścia, ponieważ aplikacja z liniową logiką nie może prowadzić przetwarzania w całości. W przeciwieństwie do spring-retry, ponowne uruchomienie aplikacji nie prowadzi do utraty wszystkich wiadomości do ponownego przetworzenia, a jedynie do ponownego uruchomienia strategii.
To podejście pomaga zredukować obciążenie zewnętrznego systemu, który może być niedostępny z powodu bardzo wysokiego obciążenia. Innymi słowy, oprócz ponownego przetwarzania, udało nam się wdrożyć wzorzec. .
W naszym przypadku próg błędu wynosi tylko 1, a aby zminimalizować przestoje systemu spowodowane tymczasową awarią sieci, stosujemy bardzo granularną strategię ponownych wywołań z małymi interwałami opóźnienia. Może to nie pasować do wszystkich aplikacji w grupie firm, dlatego proporcje między progiem błędu a wielkością interwału należy dobierać w zależności od cech systemu.
Osobna aplikacja do przetwarzania wiadomości z aplikacji o niedeterministycznej logice.
Oto przykład kodu, który wysyła wiadomość do takiej aplikacji (Retryer), która wykona ponowną wysyłkę do tematu DESTINATION po osiągnięciu czasu RETRY_AT:
public void retry(ConsumerRecord record, String retryToTopic,
Instant retryAt, String counter, String groupId, Exception e) {
Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
List arrayOfHeaders =
new ArrayList(Arrays.asList(headers.toArray()));
updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
updateHeader(arrayOfHeaders, ORIGINAL_PARTITION,
() -> Integer.toString(record.partition()).getBytes());
if (nonNull(retryAt)) {
updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
} else {
updateHeader(arrayOfHeaders, REASON,
ExceptionUtils.getStackTrace(e)::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
}
ProducerRecord messageToSend =
new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
kafkaTemplate.send(messageToSend);
} Z przykładu widać, że wiele informacji przekazywanych jest w nagłówkach. Wartość RETRY_AT znajduje się tak samo, jak w mechanizmie powtórzeń poprzez zatrzymanie Consumer’a. Oprócz DESTINATION i RETRY_AT przekazujemy:
- GROUP_ID, według którego grupujemy wiadomości do ręcznej analizy i uproszczenia wyszukiwania.
- ORIGINAL_PARTITION, aby spróbować zachować tego samego Consumer do ponownego przetwarzania. Ten parametr może mieć wartość null, w takim przypadku nowa partycja zostanie uzyskana na podstawie klucza record.key() oryginalnej wiadomości.
- Zaktualizowana wartość COUNTER, aby śledzić strategię ponownych wywołań.
- SEND_TO — stała wskazująca, czy wysłać wiadomość do ponownego przetworzenia po osiągnięciu RETRY_AT, czy umieścić ją w DLQ.
- REASON — powód, dla którego przetwarzanie wiadomości zostało przerwane.
Retryer przechowuje wiadomości do ponownego wysłania i ręcznej analizy w PostgreSQL. Na podstawie timera uruchamiana jest zadanie, które znajduje wiadomości z osiągniętym RETRY_AT i wysyła je z powrotem do partycji ORIGINAL_PARTITION tematu DESTINATION z kluczem record.key().
Po wysłaniu wiadomości są one usuwane z PostgreSQL. Ręczna analiza wiadomości odbywa się w prostym interfejsie użytkownika, który współdziała z Retryer za pomocą REST API. Jego główne cechy to ponowne wysyłanie lub usuwanie wiadomości z DLQ, przeglądanie informacji o błędach oraz wyszukiwanie wiadomości, na przykład według nazwy błędu.
Ponieważ w naszych klastrach włączone jest zarządzanie dostępem, konieczne jest dodatkowe żądanie dostępu do tematu, który nasłuchuje Retryer, oraz umożliwienie Retryerowi pisania do tematu DESTINATION. Jest to niewygodne, ale w odróżnieniu od podejścia z tematem w interwałach, mamy pełnoprawny DLQ i interfejs użytkownika do zarządzania nim.
Zdarzają się sytuacje, gdy przychodzący temat jest odczytywany przez kilka różnych grup consumer, których aplikacje implementują różną logikę. Ponowne przetwarzanie wiadomości przez Retryer dla jednej z takich aplikacji spowoduje duplikat w innej. Aby się przed tym zabezpieczyć, tworzymy osobny temat do ponownego przetwarzania. Temat przychodzący i temat retry mogą być odczytywane przez tego samego Consumer bez żadnych ograniczeń.
Domyślnie podejście to nie zapewnia możliwości circuit breaker’a, jednak można go dodać do aplikacji za pomocą lub nowego , owijając miejsca wywołań zewnętrznych serwisów w odpowiednie abstrakcje. Ponadto pojawia się możliwość wyboru strategii dla wzoru, co również może być przydatne. Na przykład, w spring-cloud-netflix może to być pula wątków lub semafor.
Wnioski
W efekcie uzyskaliśmy osobną aplikację, która umożliwia powtórzenie przetwarzania wiadomości w przypadku tymczasowej niedostępności jakiegoś zewnętrznego systemu.
Jedną z głównych zalet aplikacji jest to, że mogą z niej korzystać zewnętrzne systemy działające w tym samym klastrze Kafka, bez większych modyfikacji po swojej stronie! Takiej aplikacji wystarczy tylko uzyskać dostęp do tematu retry, wypełnić kilka nagłówków Kafka i wysłać wiadomość do Retryera. Nie trzeba uruchamiać żadnej dodatkowej infrastruktury. Aby zredukować liczbę wiadomości przenoszonych z aplikacji do Retryera i z powrotem, wyodrębniliśmy aplikacje z liniową logiką i wprowadziliśmy w nich powtórne przetwarzanie poprzez zatrzymanie Konsumenta.
Źródło: habr.com
