Konsumenci Kafka

Dostępność funkcjonalności zależy od licencji i może nie być dostępna we wszystkich wdrożeniach.

Cel

Procesy w Eximee mogą składać się z kroków, które wymagają asynchronicznej komunikacji z systemami zewnętrznymi. Przykładowo: wysyłamy żądanie o weryfikację dokumentów, jednak ta jest czasochłonna, więc na odpowiedź będziemy musieli poczekać. Zadanie to może być wykonane asynchronicznie w zewnętrznym systemie i może on powiadomić proces Eximee za pośrednictwem Kafki. Konsumenci Kafka służą właśnie do obsługi takich odpowiedzi. Używamy ich do odebrania wiadomości, wstępnego przetworzenia zawartych w nich danych oraz powiadamiania Eximee BPMS, by kontynuować wykonywanie procesu.

Powiązanie topika Kafki z Konsumentem Kafki

Każdy konsument Kafki musi być powiązany z topikiem, na który będzie nasłuchiwał. Do ustalenia takiego mapowania potrzebujemy:

• nazwy topicu - powinniśmy ją dostać odgórnie,
• nazwy konsumenta Kafki - ustalamy ją sami, analogicznie do nazwy innych artefaktów,
• mając obie nazwy, możemy zgłosić administratorom prośbę o dodanie do konfiguracji mapowania topicu na konsumenta Kafki (podając obie nazwy). Bez mapowania, będziemy w stanie napisać konsumenta Kafki, jednak nie będzie on odbierał wiadomości.

Wymagania konfiguracyjne po stronie Kafka

Dla każdego topicu używanego przez low-code consumera administratorzy muszą przygotować nie tylko topic podstawowy, ale też kompletny zestaw topiców do ponowień i wiadomości odrzuconych:

• <nazwa-topicu>-retry-0
• <nazwa-topicu>-retry-1
• <nazwa-topicu>-dlt

Jeżeli którykolwiek z tych topiców nie istnieje, konfiguracja jest niepełna i mechanizm ponowień oraz obsługi DLT nie będzie działał poprawnie.

Wymóg dotyczy absolutnie każdego topicu obsługiwanego przez low-code consumera.

Dodatkowo konsument Eximee musi mieć nadane uprawnienie idempotentWrite, ponieważ w ramach obsługi błędów zapisuje wiadomości na topicach retry i DLT.

Przykład kompletnej konfiguracji topiców

Jeżeli low-code consumer korzysta z topicu eximee-mortgage.fct.email-status, przygotuj cały zestaw:

• eximee-mortgage.fct.email-status
• eximee-mortgage.fct.email-status-retry-0
• eximee-mortgage.fct.email-status-retry-1
• eximee-mortgage.fct.email-status-dlt

Przy zgłoszeniu do administratorów podaj również informację, że dla konsumenta Eximee trzeba nadać uprawnienie idempotentWrite.

Obsługa błędów i ponawianie dostarczenia wiadomości

Konsument low-code może zakończyć obsługę wiadomości na dwa sposoby:

1. Nie zgłasza wyjątku - wiadomość zostaje uznana za poprawnie obsłużoną.
2. Zgłasza wyjątek MessageNotReadyToProcessException i wymusza ponowną próbę dostarczenia.

Domyślnie low-code consumer korzysta z następującej konfiguracji ponowień:

• maxAttempts = 3
• initialInterval = 60000 ms
• multiplier = 5.0
• maxInterval = 300000 ms

Oznacza to, że standardowo wykonywane są 3 próby łącznie:

1. pierwsza próba na topicu podstawowym,
2. druga próba po 1 minucie na topicu -retry-0,
3. trzecia próba po 5 minutach na topicu -retry-1.

Jeżeli trzecia próba również zakończy się wyjątkiem, komunikat trafia do topicu -dlt.

Przebieg obsługi błędu wygląda więc następująco:

1. Wiadomość trafia na topic podstawowy.
2. Jeżeli konsument zgłosi wyjątek, po 1 minucie wiadomość trafia na topic z przyrostkiem -retry-0.
3. Jeżeli druga próba również zakończy się wyjątkiem, po 5 minutach wiadomość trafia na topic z przyrostkiem -retry-1.
4. Jeżeli trzecia próba także zakończy się wyjątkiem, wiadomość trafia na topic z przyrostkiem -dlt.

Konfiguracja własnych czasów retry

Czasy retry są konfigurowane per kafkaId za pomocą właściwości:

• script.events.<kafkaId>.retry.max-attempts
• script.events.<kafkaId>.retry.initial-interval
• script.events.<kafkaId>.retry.multiplier
• script.events.<kafkaId>.retry.max-interval

Przykład konfiguracji:

script:
  events:
    kafkaCentralna:
      retry:
        max-attempts: 4
        initial-interval: 30000
        multiplier: 2.0
        max-interval: 180000

W tym przykładzie:

• po pierwszym błędzie kolejna próba nastąpi po 30 sekundach,
• następne opóźnienia będą wyliczane z użyciem mnożnika 2.0,
• pojedyncze opóźnienie nie przekroczy 180 sekund,
• łączna liczba prób wyniesie 4.

Po przeniesieniu wiadomości na Dead Letter Topic komunikat nie jest już automatycznie przekazywany dalej ani przetwarzany ponownie. Wymagana jest reakcja administratora, który powinien przeanalizować przyczynę błędu i zdecydować o dalszych działaniach.

Obecność topicu DLT nie rozwiązuje problemu automatycznie. W aktualnej implementacji nie ma dedykowanego wyjątku typu "nie próbuj więcej" oraz nie ma automatycznego dalszego przetwarzania komunikatów z DLT. Jest to miejsce, do którego trafiają komunikaty wymagające ręcznej analizy i interwencji administracyjnej.

Konsumenci Kafka w Procesie

Konsumenci Kafka mogą działać niezależnie, jednak zazwyczaj będą powiązani z procesem w Eximee BPMS. Konsumenci Kafka w procesie biorą udział w zdarzeniach typu "Message". Na kroku tego typu Eximee BPMS oczekuje na wiadomość, by móc kontynuować wykonywanie procesu. Wiadomość może zostać wysłana z konsumenta Kafki.

Ilustracja 1. Przykładowy proces z krokiem typu "Message Intermediate Catch Event" (zdarzenie "Oczekuj na powiadomienie z banku")

Na etapie projektowania procesu musimy pamiętać o uzupełnieniu nazwy wiadomości w konfiguracji. Musi spełniać ograniczenia Eximee BPMS (Message Events) i będzie nam potrzebna później, przy powiązaniu kroku procesu z konsumentem Kafki.

Ilustracja 2. Przykładowa konfiguracja kroku typu "Message Intermediate Catch Event"

Tworzenie Konsumenta Kafka

W Eximee Designer wybieramy odpowiednio Biblioteka → Konsumenci Kafka. Tutaj, podobnie do pozostałych artefaktów, znajduje się przycisk Dodaj Konsumenta Kafka:

Ilustracja 3. Zakładka z typem artefaktów: Konsumenci Kafka

Konsumenci Kafka mają ograniczony dostęp do API i mogą korzystać tylko z funkcji dostępnych w api.process.*. Więcej informacji o API: Operacje i dostęp do danych procesu

Przykład użycia konsumenta Kafka:

function callService(context) {
    // pobranie treści wiadomości z kontekstu
    let message = context.getMessage().value;
    // przetwarzanie wiadomości
    let jsonValue = JSON.parse(stringValue);
    let processInstanceId = jsonValue.processInstanceId;
    // wysyłka wiadomości do Eximee BPMS z informacją, by kontynuować proces
    api.process.v1.byInstanceId(messageValue.processInstanceId).correlateMessage('get-notification', null);
    return;
}

Format wiadomości nie jest narzucany, więc struktura i typ zawartych w niej danych powinny być ustalone odgórnie, by być w stanie je poprawnie przetworzyć.

Pisząc konsumenta Kafki, niemal zawsze będziemy zaczynać od pobrania treści wiadomości, używając context.getMessage().value;. To jak później możemy przetworzyć wiadomość, zależy od ustalonego formatu wiadomości, więc zestaw i typ pól w każdym przypadku mogą być inne. Ważne jest, aby na etapie ustaleń pamiętać o identyfikatorze instancji procesu. W przykładzie jest to pole processInstanceId. Dzięki tej wartości wiemy, której instancji procesu dotyczy dany komunikat.

Pobranie danych z wiadomości

Używając context.getMessage() pobieramy wiadomość i jej metadane. Obiekt, który jest zwracany przez tę funkcję, ma następujące pola:

Nazwa	Typ	Opis
value	string	Treść (payload) wiadomości.
headers	object	Nagłówki wiadomości Kafka jako mapa klucz–wartość
key	string	Klucz wiadomości Kafka. Może być null, jeśli nie został uzupełniony.
topic	string	Nazwa topiku Kafka, z którego pochodzi wiadomość
partition	number	Numer partycji topiku, z której pochodzi wiadomość
offset	number	Znacznik czasu (epoch ms) utworzenia wiadomości
timestamp	number	Czas utworzenia wiadomości

Należy mieć na uwadze, że:

• context.getMessage() - pobiera cały obiekt wiadomości wraz z metadanymi,
• context.getMessage().value - pobiera samą treść wiadomości. Najczęściej będziemy korzystać właśnie z tej formy.