Please enable JavaScript.

Coggle requires JavaScript to display documents.

Kafka - Coggle Diagram

- - - - powinna trzymać topiki które są ze sobą powiązane use case'm oraz mają podobny workload, tzn. nie mają dużych obciążeń żeby nie wpaść w rebalancing
      - To jest jeden use-case / ten sam pipeline
        
        Np. jeden serwis buduje “Conversation Read Model” i potrzebuje eventów z: user_messages, bot_responses, conversation_state_changed
        
        Wtedy jedna consumer group ma sens, bo:
        
        te instancje wspólnie robią to samo zadanie,
        
        chcesz skalować poziomo jako jeden worker pool.
      - Mają podobne SLA i podobną “ciężkość” przetwarzania
        
        jeśli jeden topic ma “ciężkie” eventy (długie I/O), może zepsuć stabilność całej grupy (max.poll.interval, rebalance)
        
        więc łączenie jest dobre, gdy profil obciążenia jest podobny
  - - - Consumer → przetwarza i pilnuje offsetu w RAM
        Broker → przechowuje ostatni zatwierdzony offset przez commit
      - Offset należy do brokera jest: per topic, per partition.
        Oznacza pozycję rekordu w logu. Używany przez consumerów (do wiedzy „gdzie jestem”)
      - Consumer – przechowuje offset tymczasowo w pamięci - wie, który event pobrał, wie, na którym offsetcie jest, ale po restarcie wszystko to znika.
    - - Broker – przechowuje offset trwale w specjalnym wewnetrznym topicu. To topic __consumer_offsets.
      - Co zawiera?
        
        Rekordy opisujące stan consumer group, głównie:
        
        offsety:
        
        group.id
        
        topic
        
        partition
        
        offset
        
        metadata:
        
        commit timestamp
        
        (czasem) generation id
        
        (czasem) informacje o rebalansie
        
        ❌ Nie zawiera informacji o brokerach danych
        ❌ Nie mapuje partycji topicu → broker
    - - Dlaczego commit offsetu ≠ przetworzenie eventu? bo to tylko potwierdzenie otrzymania eventu a nie jego przetworzenia
      - Dlatego commit może być:
        
        I właśnie dlatego Kafka gwarantuje at-least-once z natury,a exactly-once wymaga świadomego kodu (lub transakcyjnych producerów + isolation level).
        
        za późno → duplikujesz eventy
        
        za wcześnie → tracisz eventy
      - to po prostu potwierdzenie przez consumera ze przetworzyl event i moze brac kolejny. Tym samym broker aktualizuje sobie offset i dzieki temu wie ktory event ma dac jako kolejny
    - - LEO (Log End Offset)
        
        “jak daleko leader dopisał log” (ostatni offset w logu leadera)
      - HW (High Watermark)
        
        “jak daleko dane są zreplikowane do ISR”
        
        konsumenci domyślnie nie czytają ponad HW (żeby nie czytać danych, które mogą zniknąć przy failover)
        
        HW rośnie dopiero wtedy, gdy rekord jest potwierdzony przez wszystkie repliki w ISR (czyli jest “bezpieczny” do czytania)
  - - - pauzuje konsumpcję,
      - może chwilowo zwiększyć lag,
      - i bywa źródłem „rebalance storm”
    - - consumer nie robi poll() wystarczająco często (np. bo przetwarzanie trwa za długo),
        i uderza w max.poll.interval.ms → Kafka wyrzuca go z grupy → rebalance.
  - - - Limit czasu życia to:
        session.timeout.ms domyslnie 10s
      - Consumer musi regularnie odzywać się do brokera (Group Coordinator), inaczej broker uznaje go za martwego i robi rebalancing.
        👉 heartbeat NIE jest wysyłany podczas poll()
        👉 heartbeat jest wysyłany w tle przez osobny wątek kliencki
      - Producer go nie wysyła, jego tez nie dotyczy rebalancing.
    - - Consumer MUSI wołać poll() co jakiś max czas (default: 5 minut).
        max.poll.interval.ms
      - Jesli wysyłasz heartbeat, ALE nie wywołujesz poll() w ciągu max.poll.interval.ms
        to 👉 consumer zostaje wyrzucony👉 nawet jeśli serce bije👉 Kafka zrobi rebalance👉 inny consumer przejmie partycje
      - „Skoro nie wołasz poll(), to nie jesteś gotowy na eventy.”
  - - - Lag rośnie + brak CPU/DB bottlenecku → sprawdź rebalance frequency
      - Lag rośnie + długie przetwarzanie → sprawdź max.poll.interval.ms
      - Lag skacze z zębami piły → bardzo często rebalancing
  - - - Co się dzieje z rekordami z abortowanej transakcji?
        Kafka ich nie usuwa z logu. One dalej fizycznie są w segmentach, ale: przy read_committed są filtrowane (ignorowane), broker używa metadanych transakcyjnych, żeby wiedzieć co pominąć.
    - - konsument widzi tylko dane z transakcji zatwierdzonych, ignoruje rekordy z abortowanych transakcji
      - Co się dzieje z rekordami z abortowanej transakcji?
        Kafka ich nie usuwa z logu. One dalej fizycznie są w segmentach, ale: przy read_committed są filtrowane (ignorowane), broker używa metadanych transakcyjnych, żeby wiedzieć co pominąć.
- - - - brak zewnętrznej DB na stan,
      - dokładnie zsynchronizowany stan + offsety,
      - EOSv2 (exactly-once wewnątrz Kafki),
      - bardzo szybki lokalny dostęp do stanu.
  - - - na dysku (RocksDB),
      - per instancja aplikacji,
      - per partycja.
  - - - Kafka Streams tworzy wewnętrzny topic, przepartycjonowuje dane.
      - To:
        
        kosztuje (IO, sieć),
        
        zmienia ordering,
        
        wpływa na wydajność.
  - - - → event + current state
        → „wzbogać zdarzenie”
      - Zeby KStream–KTable join działał poprawnie, obie strony muszą być skeyowane tak samo
    - - → state + state
        → „zbuduj nowy stan”*
- - - - broker odpowiedzialny za zarządzanie jedną consumer group.
      - Czym Group Coordinator NIE jest
        
        ❌ nie jest „brokerem od danych partycji”
        
        ❌ nie jest leaderem topicu
        
        ❌ nie przetwarza eventów
      - Dla danej consumer group robi 3 kluczowe rzeczy:
        
        1️⃣ Zarządza członkostwem w grupie - kto dołączył, odpadł, wysyła heartbeat, został uznany a maertwego.
        
        2️⃣ Prowadzi rebalancing - gdy consumer dołącza, odpada rozdziela partycje na nowo
        
        3️⃣ Przechowuje offsety - zapisuje offsety, udostępnia je po rebalansie, pozwala nowemu consumerowi „wiedzieć, od czego zacząć”.
      - Skąd wiadomo, który broker jest Group Coordinator?
        Mechanizm:
        Kafka robi hash po group.id
        mapuje to na partycję __consumer_offsets
        leader tej partycji = Group Coordinator
      - Group Coordinator:
        
        nie musi być brokerem, który trzyma dane topiców konsumowanych przez grupę
        
        może być zupełnie innym brokerem
    - - rola “danych”. Przyjmuje zapisy eventów
      - Dla każdej partycji:
        
        1 broker jest leaderem (przyjmuje zapisy i zwykle obsługuje odczyty)
        
        pozostałe broker-y są followerami (replikują dane)
    - - śledzi transakcje danego transactional.id, wie, czy transakcja została commit czy abort.
      - Coordinator przechowuje mapowanie transactional.id → status transakcji + epoch itp.
        Dzięki temu broker (przy read_committed) potrafi:rozpoznać, że część rekordów należy do transakcji, sprawdzić, czy transakcja została commit/abort, i odfiltrować te abortowane.
      - dotyczy: transakcyjnych producentów
        i zarządza stanem transakcji (begin/commit/abort)
    - - Jeden broker (w danym momencie) jest controllerem i odpowiada za:
        
        leader election partycji
        
        pilnowanie ISR
        
        reakcje na padnięcia brokerów
        
        zmiany w metadanych klastra
      - W nowszych Kafka (KRaft) “controller” może być nawet oddzielnym procesem/rolą (controller nodes), ale idea pozostaje ta sama: ktoś jeden koordynuje klaster.
  - - - za długo nie nadąża z replikacją,
      - jest przeciążona,
      - ma problem z IO / GC / siecią,
      - nie odpowiada w czasie.
    - - replica.lag.time.max.ms
        
        zanim wyleci z ISR.
        
        ile czasu follower może nie odpowiadać,
      - replica.lag.max.messages
        
        ile rekordów follower może być „do tyłu”, zanim wyleci z ISR.
      - min.insync.replicas
        
        Gdy np. ustawiony na 2 to Kafka odrzuci zapis, jeśli w ISR zostanie tylko 1 replika.
        
        jeśli mniejszy od replication factor, to wtedy istnieją brokery które nie muszą dawać ack przy zapisie eventów do logu i nadążać tak szybko jak in sync replicas
    - - Kafka wybiera nowego leadera z ISR nigdy spoza ISR (jeśli unclean.leader.election=false)
      - Kafka może wybrać „brudną” replikę → utrata danych. Dlatego na prod: unclean.leader.election = false (prawie zawsze)
    - - follower dogania leadera (kopiuje brakujące rekordy)
        gdy jest „wystarczająco blisko” i stabilny, wraca do ISR
    - - To nie jest domyślny tryb i zwykle robi się to świadomie.
      - Czytanie z followerów oznacza ryzyko stale reads (opóźnionych danych), bo follower może być „krok za” leaderem — czasem minimalnie, czasem znacząco przy problemach IO/sieci.
      - Ma sens, gdy:
        
        masz ciężkie read workload (dużo konsumentów / duże fetch’e),
        
        a zapis jest mniej krytyczny niż rozkład obciążenia,
        
        albo masz geograficzne klastry i chcesz czytać „bliżej”.
    - - To bardzo częsty realny problem: jedna replika raz nadąża, raz nie — ISR skacze, a przy acks=all latencja i/lub błędy zapisów potrafią “pulsować”.
      - Najczęstsze przyczyny (kontekst)
        
        wolny dysk na jednym brokerze (IO wait)
        
        GC / CPU throttling
        
        sieć z jitterem
        
        zbyt duże batch’e / zbyt agresywny load
        
        źle dobrana liczba partycji do sprzętu
  - - - Co gdy leader się zmieni?
        
        Controller robi leader election
        Broker-y aktualizują metadane
        Klient dostaje błąd typu “not leader” / “stale metadata” albo timeout
        Klient robi metadata refresh i wysyła ponownie już do właściwego brokera
        
        Szybsze odświeżanie metadanych
        
        W kliencie możesz mieć parametry, które wpływają na to, jak często/łatwo odświeża metadata (zależnie od biblioteki i wersji). Najbardziej praktyczny mechanizm to:
        
        krótsze timeouts
        
        retry/backoff (nie za agresywny, ale szybki)
        
        poleganie na tym, że błąd metadanych triggeruje refresh
      - Skąd wiadomo kto jest leaderem?
        
        producer musi wysłać rekord do leadera partycji, a consumer pobiera dane od leadera (zwykle).
        Klient (producer/consumer) ma lokalny cache:
        topic → partycje, partycja → broker-leader, lista brokerów
        Klient odświeża to przez request typu Metadata.
- - - - BROKER I TAK DOSTAJE wiadomość (jeśli połączenie żyje), ale producent nie wie, czy dotarła / została zapisana,
      - najmniejsze opóźnienia, największe ryzyko utraty.
    - - producent czeka na potwierdzenie od leader-a partycji, jeśli leader zapisze do swojego logu → sukces, jeśli leader padnie zanim repliki zdążą skopiować → możesz stracić ten event,
      - kompromis: trochę bezpieczeństwa, umiarkowane opóźnienia.
    - - producent czeka, aż leader + wszystkie in-sync repliki (ISR) potwierdzą zapis, otrzymają rekord do swojej pamięci (niekoniecznie na dysk).
      - najwyższa trwałość (przy RF=3 event jest na 2–3 brokerach),
      - większe opóźnienia.
      - nie chodzi o „wszystkie brokery w klastrze”, tylko o wszystkie repliki tej partycji, Kafka NIE czeka na wszystkie repliki — tylko na ISR.
  - - - wrzuca rekord do batcha dla danej partycji,
      - wysyła batch do właściwego brokera (leader partycji).
    - - zapisuje do logu,
      - replikuje do followerów,
      - odsyła odpowiedź („OK” / błąd).
    - - na podstawie acks + odpowiedzi → uznaje rekord za:
        
        dostarczony,
        
        do retria,
        
        albo za fail.
  - - - Kafka wtedy:
        
        numeruje wiadomości per partycja, np. seq=0,1,2,...,
        
        broker pamięta: „od tego PID dla tej partycji już widziałem seq do X”.
        
        nadaje producentowi PID (Producer ID),
      - Jeśli producent retry’uje ten sam batch (poprzedni ack od leadera mogl nie dotrzec, zaginac w sieci):
        Broker: „aha, PID ten sam, seq te same – to już mam, ignoruję duplikat.”
    - - Nawet z idempotentnym producentem może być zabawnie, jeśli: masz kilka requestów „w locie”, dochodzi do retry, a odpowiedzi przyjdą w innej kolejności.
        
        dlatego ustawia sie
        
        acks=all retries=Integer.MAX_VALUE max.in.flight.requests.per.connection=1 (lub mała wartość, np. 5)
        
        max.in.flight=1:
        zmniejsza przepustowość,
        ale upraszcza gwarancje kolejności i deduplikacji.
    - - Nie chroni przed duplikatami z powodu restartu PRODUCERA
        
        Po restarcie producent dostaje nowy PID, dla brokera to już „inny” producent. Broker nie widzi już sekwencji z poprzedniej instancji.
        
        → Eventy mogą się powtórzyć, jeśli logika aplikacji je retry’uje lub wysyła ponownie.
        
        idempotent producer chroni tylko przed duplikatami wynikającymi z retry w obrębie jednego procesu
      - Nie chroni przed duplikatami z powodu logiki biznesowej. Jeśli Twój kod wywoła send() dwa razy — idempotent producer tego nie naprawi.
      - Nie zapewnia exactly-once end-to-end. To wymaga:
        
        transakcyjnego producenta,
        
        read_committed po stronie konsumera,
        
        albo użycia Kafka Streams z EOS.
      - Nie deduplikuje eventów wysłanych z innych aplikacji
        Idempotencja jest per-producer, nie globalna.
  - - - ile producer może poczekać, zanim wyśle niepełny batch.
    - - Trade-off:
        
        większe limity/czas → throughput ↑, latency ↑
        
        mniejsze limity/czas → latency ↓, throughput ↓
      - Cena: więcej pamięci/buforów, czasem większe bursty
      - max rozmiar batcha na partycję (w bajtach).
    - - Mniej danych → mniej sieci i dysku
      - Cena: CPU na kompresję
      - zmniejsza payload, często poprawia throughput
  - - - Przy pierwszym send() broker mówi:👉 „oto Twój PID oraz starting sequence numbers per partition”
      - Producent zapisuje PID w pamięci
      - Po restarcie → producent nie ma starego PID → dostaje nowy
      - Broker traktuje go jak innego producenta
  - - - dla PID=1234 i partition=3 widziałem seq do np. 56
      - jeśli dostanie seq=56 ponownie → duplikat → ignoruje
      - jeśli dostanie seq=57 po seq=56 → OK
      - jeśli dostanie seq=100 po seq=56 → błąd (sequence out of order)
    - - Zawsze musi znać: która partycja będzie użyta, który broker jest leaderem tej partycji
      - Jak to się dzieje:
        
        Producer łączy się z klastrem.
        
        Odczytuje metadata wszystkich topiców i partycji.
        
        Default partitioner oblicza partycję → hash(key) % partitions.
        
        Producer wysyła batch bezpośrednio do leadera tej partycji.
        
        Jeśli leader padnie →producer pobiera metadane ponownie →wysyła do nowego leadera.
- - - - Partycje tego samego topicu są rozrzucone po różnych brokerach. Możesz mieć np. klaster 5 brokerów, a kafkowy kontroler automatycznie rozkłada partycje tak, żeby każdy broker był leaderem mniej więcej takiej samej liczby partycji.
      - To kontroler klastra (KRaft controller lub ZooKeeper w starych wersjach) decyduje: która replika ma być liderem, i może to zmienić dynamicznie.
      - to jedna instancja brokera w klastrze wybrana przez ZooKeepera (stare Kafka) albo KRaft (nowa Kafka)
      - zarządza:
        
        leader election
        
        ISR
        
        zmianami topologii klastra
  - - - scalenia logów z wielu partycji w jedną,
      - zmiany offsetów,
      - rozwalenia gwarancji ordering i spójności odczytów.
  - - - .log (dane),
      - .index (offset → pozycja),
      - .timeindex (timestamp → offset)
    - - ułatwiają retention,
      - przyspieszają odczyt,
      - pozwalają usuwać stare dane bez locków.
- - - - Kafka usuwa stare segmenty gdy:
        
        retention.ms
        
        Messages are deleted only after a certain time period is reached (time based retention)
        
        retention.bytes
        
        When a max message size for a partition is reached (size based retention)
      - trzymasz historię eventów przez X dni (np. 30/90/365),
      - możesz zrobić replay i odtworzyć procesy,
      - to naturalne dla eventów typu „co się stało”.
      - retencja to decyzja biznesowo-prawno-kosztowa, nie techniczna.
    - - Compaction dąży do tego, żeby w długim horyzoncie zostawić głównie ostatnią wartość dla danego key (plus ewentualnie tombstone).Czyli to bardziej pod:
        
        „co się działo po drodze?”.
        
        „jaki jest aktualny stan?”niż
      - Tombstone
        
        To rekord:
        
        z tym samym key
        
        i value = null
        
        Kafka interpretuje to jako „ten klucz ma zostać usunięty”
        
        wszyscy muszą zobaczyć delete, bo inaczej ktoś będzie trzymał „ducha” np. usera forever, dlatego ten event jest trzymany dluzej
      - to nadal jest log rekordów (append-only), tylko Kafka w tle sprząta stare wersje danego key. Nie dzieje się to natychmiast, a w tle więc czasem możesz zobaczyć pośrednie stany np. danego usera o danym key
      - Log Cleaning / Segment Rewrite
        
        Cleaner: bierze segmenty, przepisuje je do nowych segmentów tak, żeby usunąć stare wersje key, zostawia najnowsze (i tombstone’y)
  - - - nazwa ma znaczenie,
      - schemat ma znaczenie,
      - zmiany są kosztowne,
      - konsumenci nie są „wewnętrzną implementacją”.
  - - - Czyli: jeden topic payments zawiera eventy z całego cyklu życia płatności:
        
        PaymentAuthorized
        
        PaymentCompleted
        
        PaymentFailed…
        
        PaymentCreated
      - Plusy:
        
        prosta konsumpcja,
        
        naturalny ordering per encja,
        
        łatwiejszy replay „historii”.
      - Minus:
        
        topik może być „głośny” (dużo typów eventów),
        
        konsumenci muszą filtrować typy.
    - - Np. payment.completed, payment.failed, itd.
      - Plusy:
        
        konsumenci biorą tylko to, co ich interesuje,
        
        mniejszy „szum”.
      - Minus:
        
        tracisz ordering cyklu życia (bo to różne logi),
        
        rośnie liczba topików i koszty operacyjne.
  - - - checkout.payments
      - billing.payments
      - finance.payments
    - - uniknąć „płacenia za cudze paymenty” (tzn. mieszania znaczeń),
      - uniknąć konfliktu semantycznego między zespołami,
      - łatwiej robić ownership i governance.
- - - - per topic,
      - per partycja,
      - per klaster.
  - - - Replikujesz A → B
      - Failover przełącza ruch na B
      - A jest primary, B standby
    - - Produkujesz w obu klastrach
      - Replikujesz w obie strony
  - - - MM2 konsumuje z A i produkuje do B
      - lag mówi: „jak bardzo B jest do tyłu względem A”
- - - - zapis = sekwencyjny append (mega szybkie),
      - odczyt = page cache OS, nie JVM,
      - brak update/delete w miejscu.
  - - - Page cache = RAM, którego system używa jako bufora na pliki dyskowe
      - Gdy program (Kafka) zapisuje do pliku: dane nie lecą od razu na fizyczny dysk trafiają najpierw do RAM-u (page cache) OS sam decyduje, kiedy faktycznie zapisać je na dysk**
  - - - Dlatego mówimy:
        Kafka daje bardzo wysoką trwałość,
        ale nie absolutną (jak fsync-per-write).
      - Scenariusz:
        Event trafia do RAM leadera + followera (ISR)
        ACK idzie do producera
        OBIE maszyny padają jednocześnie
        Dane nie zdążyły trafić na dysk
        👉 Event może zostać utracony
    - - To jest świadomy trade-off:
        
        ogromna wydajność
        
        bardzo niskie ryzyko
        
        ale nie „zero ryzyka w fizyce wszechświata”
- - - - RF=3 → ten sam payload idzie po sieci więcej niż raz (leader→followers)
      - dużo consumerów (fetch traffic)
      - brak kompresji albo duże payloady
      - dużo danych na sekundę (MB/s, nie msg/s)
    - - słabe dyski / duży IO wait
      - compaction (przepisywanie segmentów)
      - bardzo duży retention + dużo segmentów do zarządzania
      - broker ma mało RAM (mniej page cache → więcej realnych odczytów z dysku)
    - - mocna kompresja/dekompresja (np. zstd na wysokim levelu)
      - TLS (szyfrowanie) przy dużym ruchu
      - bardzo dużo małych wiadomości (overhead per request)
      - ciężkie transformacje po stronie klienta (ale to już app-level)
  - - - narzut protokołu (nagłówki, metadane),
      - koszty syscalli i obsługi sieci,
      - koszty serializacji/enkapsulacji,
      - koszty obsługi requestu po stronie brokera (wątki, kolejki, bookkeeping).
- - - - poll() → commit() → process(message)
    - - poll() → process(message) → commit()
  - - - nie zależy od fizycznego ułożenia w logu
      - możesz zmieniać klaster / topik / partycje, a ID nadal jest ważne,
      - Producent musi ponawiać wysyłkę z tym samym eventUuid,
        a nie generować nowy przy każdym retrze. eventUuid powinien Dlatego powstać w logice biznesowej (np. przy tworzeniu płatności), a nie „ad hoc” w miejscu, gdzie robisz producer.send().
    - - mocno wiąże idempotencję z fizyczną strukturą klastra,
      - przy migracji / reprocessingu / kopiowaniu danych między klastrami offsety mogą się zmienić,
- - - - pozwala wysłać wiele eventów do wielu partycjii na końcu powiedzieć:
        
        abortTransaction() → nic nie istnieje
        
        commitTransaction() → wszystko widoczne
      - Dla konsumentów z isolation.level=read_committed:
        
        abortowane eventy są niewidoczne
        
        nigdy ich nie zobaczą
    - - Co jest wymagane, żeby to działało?
        
        konsument czyta z isolation.level=read_committed
        
        retries włączone
        
        acks=all (wymuszone)
        
        enable.idempotence=true (wymuszone)
        
        transactional.id ustawiony (stały, unikalny per instancja logiczna)
      - Czego transakcje NIE rozwiązują?
        
        Bardzo ważne:
        
        ❌ nie spinają DB + Kafka w jedną transakcję
        
        ❌ nie chronią przed złym modelem domeny
        
        ❌ nie zastępują outboxa
        
        ❌ nie rozwiązują duplikatów po restarcie aplikacji biznesowej
        
        ❌ nie robi exactly-once w całym systemie
        
        One rozwiązują spójność wewnątrz Kafki.Dają exactly once wewnatrz Kafki.
    - - Czyli log → log, a nie DB → log.
      - „czytam z Kafki i piszę z powrotem do Kafki”
      - fan-out / fan-in
      - agregacje
      - enrich → publish
      - transformacje eventów
      - stream processing
  - - - Na końcu transakcji broker dopisuje specjalne rekordy typu:
        COMMIT marker albo ABORT marker Te markery mówią: “rekordy transakcji X w tej partycji są zatwierdzone/odrzucone”.
- - - - Przykłady:
        
        timeout DB,
        
        chwilowy 503,
        
        chwilowa awaria sieci,
        
        deadlock
      - ✅ Retry z backoff
      - Dla zachowania kolejności per payment_id masz dwa główne podejścia:
        
        A) Retry inline (na miejscu)
        
        jak timeout → śpisz np. 100ms/500ms/1s i próbujesz jeszcze raz
        
        plus: prostota, ordering naturalnie zachowany
        
        minus: blokujesz partycję, rośnie lag, ryzyko rebalance jeśli za długo
        
        B) Retry topic (z opóźnieniem)
        
        przerzucasz event do payments.retry.5s, potem payments.retry.1m itd. - delayed retry topics pattern
        Retry Escalation Pattern
        
        plus: nie blokujesz partycji głównej, kontrolujesz backoff
        
        minus: trzeba pilnować ordering per key (da się, ale wymaga dyscypliny w key)
        
        key powinien byc ten sam co na glownym topicu
        
        retry “obok” (oddzielne retry topiki + retry-worker) jest najbezpieczniejsze,
        główny topic zostawiasz jako “źródło faktów”, a nie „mechanizm retry”.
        
        Czy na retry topic idzie dokładnie ten sam event, czy wzbogacony?
        
        ✅ Ten sam event + metadane retry
        
        dokładamy tylko metadane:
        headers są idealne na takie techniczne rzeczy,
        payload zostaje domenowy i “czysty”.
        
        attempt
        
        firstSeenAt
        
        lastErrorType
        
        nextRetryAt (opcjonalnie)
        
        originalTopic/partition/offset (super do debugowania)
        
        kto pilnuje czasu?
        
        **Retry-worker konsumuje od razu, ALE: jeśli now < notBefore → nie przetwarza jeszcze.
        zamiast tego robi: pause(partition) i śpi do tego czasu (albo krótkie sleep/poll). Gdy czas minie → wznawia i przetwarza.
        Kto ustawia notBefore?
        👉 ten, kto decyduje o retry (czyli Twój kod: consumer, który wrzuca na retry topic).
        To nie jest „producent biznesowy” — to jest „retry publisher”.
        
        Opcja A2: Retry topic + dedykowany „delay processor”
        payments.retry.5m konsumuje mały serwis, który tylko czeka do notBefore i dopiero wtedy publikuje na payments.retry.ready (albo bezpośrednio do „processor topicu”).
        Plusy:
        główny procesor nie musi umieć czekać,
        lepsza separacja.
        Minus:
        dodatkowy serwis.
        
        Opcja A3: Używasz gotowego mechanizmu (Kafka Streams / Connect / framework)
        istnieją gotowe podejścia do opóźnień (np. streamowe „schedule”), które zdejmują z Ciebie ręczne pause/sleep.
    - - Przykłady:
        
        niezgodny schemat,
        
        null w polu wymaganym,
        
        błąd walidacji domenowej
      - ❌ Retry nic nie da → fallback do DLQ / parking lot + alarm + commit offsetu zeby nei blokowac topicu
      - DLQ vs Retry topic
        
        DLQ (Dead Letter Queue)
        
        Zwykle nie jest konsumowany automatycznie przez ten sam flow biznesowy.
        
        To „parking” do:
        
        analizy,
        
        ręcznej naprawy (albo automatu naprawiającego),
        
        audytu.
        
        Często ma:
        
        pełny kontekst (payload + headers + wyjątek + stacktrace + timestamp),
        
        alert/metrykę, żeby ktoś w ogóle zauważył, że rośnie.
        
        Retry topic
        
        Są 2 popularne sposoby:
        
        B) Łańcuch retry topików z opóźnieniami (np. z connectorami / stream processing)payments.retry.30s → payments.retry.5m → wraca do payments albo idzie dalej.
        
        A) Delay przez czas w wiadomości + scheduler/workerKonsument retry-topiku sprawdza notBefore i jeśli za wcześnie → odkłada / pauzuje / requeue.
        
        Ważne: retry topic ma sens, bo nie blokuje głównej partycji.
    - - Przykłady:
        
        DB down,
        
        brak połączeń w poolu,
        
        zły indeks,
        
        throttling
      - Retry bez kontroli zrobi lawinę
        ✅ Circuit breaker / pause / throttling + naprawa przyczyny