> For the complete documentation index, see [llms.txt](https://docs.eximee.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.eximee.com/eksploatacja-aplikacji/obsluga-zdarzen/konsumenci-kafka.md). # Konsumenci Kafka {% hint style="info" %} Dostępność funkcjonalności zależy od licencji i może nie być dostępna we wszystkich wdrożeniach. {% endhint %} ## 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: * `-retry-0` * `-retry-1` * `-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. {% hint style="danger" %} Wymóg dotyczy absolutnie każdego topicu obsługiwanego przez low-code consumera. {% endhint %} 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..retry.max-attempts` * `script.events..retry.initial-interval` * `script.events..retry.multiplier` * `script.events..retry.max-interval` Przykład konfiguracji: ```yaml 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. {% hint style="warning" %} 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. {% endhint %} ## 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")](/files/TYFBgZUDyt8BbjOvmIYa) Na etapie projektowania procesu musimy pamiętać o uzupełnieniu nazwy wiadomości w konfiguracji. Musi spełniać ograniczenia Eximee BPMS ([Message Events](https://docs.eximeebpms.org/manual/latest/reference/bpmn20/events/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"](/files/kUoq5WwkffVNZgdbzEmm) ## 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](/files/Yziy7mc82uGrWHftQ0TP) {% hint style="info" %} 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](/budowanie-aplikacji/logika-biznesowa/scriptcode/skrypty-scriptservice/api-skryptow/operacje-i-dostep-do-danych-procesu.md) {% endhint %} Przykład użycia konsumenta Kafka: ```javascript 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; } ``` {% hint style="info" %} 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ć. {% endhint %} 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. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.eximee.com/eksploatacja-aplikacji/obsluga-zdarzen/konsumenci-kafka.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.