Szkolenie: Apache Kafka – Kafka Streams

Kafka Streams jako biblioteka Java – architektura pętli Consumer → Logic → Producer

Podstawowe pojęcia: strumień, partycja strumienia, topologia przetwarzania, topologia podrzędna, procesor strumienia

Dualizm strumieniowo-tabelowy (stream-table duality) – KStream jako tabela i tabela jako strumień

KStream, KTable i GlobalKTable – różnice semantyczne (INSERT vs UPDATE/DELETE)

Sześć kroków budowania aplikacji Kafka Streams: konfiguracja, konstruktor topologii, źródło, transformacje, ujście, uruchomienie

Serializacja i deserializacja (Serde) – typy wbudowane i serializacja JSON oparta na Jackson/ObjectMapper

Kafka Streams DSL – mapa transformacji stanowych i bezstanowych (KGroupedStream, KGroupedTable)

Agregacja KGroupedStream: aggregate/count/reduce → KTable, trzy funktory (inicjalizacja, dodanie, usunięcie)

Agregacja KGroupedTable: obsługa elementów dodawanych i usuwanych z grupy

Zapisywanie strumieni do tematów Kafki: operacje to() i through()

Wzorzec shutdown hook z CountDownLatch – poprawne zamykanie aplikacji

Metoda describe() obiektu Topology – wizualizacja grafu przetwarzania

Architektura rozwiązania: Kafka Connect jako ujście dostarczające wyniki do docelowego systemu

Parametry commit.interval.ms, fetch.min.bytes, fetch.max.wait.ms a widoczność wyników

Kafka Streams to zwykła aplikacja Java – brak klastra YARN ani Kubernetesa

APPLICATION_ID_CONFIG: dwie role – identyfikator aplikacji i identyfikator grupy konsumentów

Skalowanie przez uruchamianie kolejnych instancji – mechanizm Consumer Group Rebalance

Zadania (tasks), partycje i limit skalowania – jeden task na jedną partycję wejściową

Dwie drogi do równoległości: wątki (num.stream.threads) vs. osobne procesy JVM – różnice i zastosowania

Instancje nadmiarowe (powyżej liczby partycji) – rola w standby replicas

Topologie podrzędne (sub-topologies) i tematy wewnętrzne: changelog i repartition

Trzy domeny czasu: event-time (CreateTime), ingestion-time (LogAppendTime), processing-time

Własny ekstraktor znacznika czasu: implementacja TimestampExtractor, konfiguracja DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG

Typy okien w Kafka Streams: tumbling, hopping, sliding, session

Agregacja okienkowa: KGroupedStream → TimeWindowedKStream → KTable

Windowed<K> jako klucz złożony – dostęp do key.key(), key.window().startTime(), key.window().endTime()

Kafka Connect JDBC Sink Connector – konfiguracja i transformacja TimestampConverter dla atrybutów czasowych

Obserwacja wyzwalaczy powtarzalnej aktualizacji – wielokrotne wpisy w ujściu dla tego samego okna

Problem nieuporządkowania danych: utrata ponad 70% zdarzeń bez grace period

Przyczyny nieuporządkowania zdarzeń: opóźnienia sieciowe, wiele partycji Kafki, różnice zegarów, buforowanie producenta

Mechanizm utraty danych: okno uznane za kompletne i usunięte przed dotarciem spóźnionych zdarzeń

Włączenie szczegółowych metryk: METRICS_RECORDING_LEVEL_CONFIG = DEBUG

Diagnoza problemu przy użyciu metryk JMX w jconsole: dropped-records-total per partycja topologii podrzędnej

Pomiar skali nieuporządkowania: metryki record-lateness-avg i record-lateness-max

Watermark = max(event-time) − gracePeriod – wyznaczanie i semantyka

Mechanizm zamykania okna: okno zamknięte gdy koniec okna < watermark

Implementacja grace period w DSL: TimeWindows.of(...).grace(Duration.ofSeconds(40))

Grace period jako parametr aplikacji przekazywany z linii poleceń

Reset środowiska po zmianie grace period: kafka-streams-application-reset.sh, usunięcie lokalnego RocksDB, wyczyszczenie tabeli docelowej

Weryfikacja kompletności wyników: 100% rekordów uwzględnionych, dropped-records-total = 0

Koszt watermarku – wpływ grace period na rozmiar stanu RocksDB: grace=1s → ~240 wpisów; grace=40s → ~750 wpisów (~3×)

Wyjaśnienie ~3× większego stanu: grace + windowSize = liczba jednocześnie aktywnych okien × liczba kluczy

Monitorowanie stanu: metryki num-entries-active-mem-table i bytes-written-total w jconsole

Trade-off: kompletność wyniku kontra koszt stanu, czas odtwarzania i opóźnienie w dostarczaniu wyników

Filozofia Kafka Streams: brak centralnego koordynatora – zadania autonomiczne (vs. JobManager Flink, StreamExecution Spark)

Trzy filary punktu kontrolnego: State Store (RocksDB na dysku lokalnym), Changelog Topic (rozproszony WAL w Kafce), plik .checkpoint (powiązanie między nimi)

Struktura katalogów na dysku: state.dir / app.id / task.id / store.name

Task ID jako para <id_podtopologii>_<id_partycji>

Changelog Topic: konwencja nazewnicza <app.id>-<store.name>-changelog, retencja compact,delete – semantyka i konsekwencje

Plik .checkpoint: mapa storeName → offset, zapis atomowy przez plik tymczasowy i rename

Cache stanu nad RocksDB: STATESTORE_CACHE_MAX_BYTES_CONFIG (10 MB domyślnie) – wpływ semantyczny i wydajnościowy

Wpływ cache na liczbę rekordów wysyłanych do changelog: bez cache 3 rekordy, z cache 1 rekord (ostatnia wartość klucza)

Algorytm pętli przetwarzania – pięć elementów: pobranie rekordu, łańcuch procesorów, operacja stanowa, wysłanie wyniku, aktualizacja offsetu

RecordCollector i KafkaProducer: jeden kanał dla wyników i zmian changelog

Algorytm zatwierdzania – cztery podfazy: 2A opróżnienie cache (flush do RocksDB i bufora producenta); 2B flush producenta changelog (acks=all); 2C atomowy zapis .checkpoint; 2D commitSync offsetów wejściowych do __consumer_offsets

Niezmiennik kolejności 2b→2c→2d: changelog utrwalony przed commitem offsetów → semantyka at-least-once

Odtwarzanie po awarii – scenariusz 1 (ten sam węzeł): odczyt .checkpoint, wyznaczenie luki w changelog, catch-up, wznowienie

Odtwarzanie po awarii – scenariusz 2 (nowy węzeł): pełny odczyt changelog od offsetu 0, retencja compact gwarantuje spójność stanu

Repliki rezerwy (num.standby.replicas): warm standby – bieżące śledzenie changelog bez aktywnego przetwarzania partycji

Przełączenie ze standby: nadrabianie kilkunastu rekordów (sekundy) vs. pełny odczyt (minuty)

session.timeout.ms: domyślnie 45s – wpływ na czas reakcji na awarię; skrócenie do 10s i ryzyko fałszywych alarmów

Strategia StreamsPartitionAssignor: minimalizacja przeprowadzek partycji, priorytetyzacja instancji z lokalnym stanem

Identyfikacja aktorów w warsztacie: Changelog Topic (GUI, retencja, klucze z metadanymi okna), Local State Store (katalogi RocksDB, segmenty czasowe), plik .checkpoint (wartość -4 przed pierwszym zapisem)

Kafka Streams wyłącza WAL w RocksDB – rola changelog jako transakcyjnego logu

Porównanie mechanizmów checkpointów: Flink (bariery), Spark Structured Streaming (WAL offsetów + StateStore), Kafka Streams (autonomiczne zadania + changelog)

Trzy filary dojrzałej aplikacji: odporność na awarie instancji, odporność na awarie brokerów, gwarancje przetwarzania

num.standby.replicas=1: konfiguracja, koszty (dysk, CPU, sieć), zysk (natychmiastowe przełączenie)

replication.factor=3 dla automatycznie tworzonych tematów (changelog, repartition): eliminacja pojedynczego punktu awarii

Semantyka at-least-once (domyślna): trzy niezależne operacje bez atomowości → duplikaty po awarii przed commitem

Scenariusz podwójnego liczenia: awaria po wysłaniu do changelog i kafka-output, ale przed commitem offsetów wejściowych

Tryb exactly-once (EOS-v2): processing.guarantee=EXACTLY_ONCE_V2, dostępny od Kafka 2.5

Algorytm EOS-v2: flushCache → beginTransaction → send transakcyjne (changelog + output) → sendOffsetsToTransaction → commitTransaction → zapis .checkpoint

Zombie fencing: transactional.id generowany z application.id + numeru partycji, nowy producent unieważnia poprzedni

Temat __transaction_state: dziennik transakcji, stany (Ongoing, PrepareCommit, CompleteCommit), recovery koordynatora

Commit Markers: dodatkowe rekordy techniczne w tematach uczestniczących w transakcji; widoczne w GUI jako "nadmiarowe wiadomości"

isolation.level=read_committed: konsumenci widzą tylko zatwierdzone transakcje – wymagane dla Kafka Connect w trybie EOS

Konfiguracja klastra dla EOS: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3, KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=2

Dynamiczna zmiana konfiguracji: kafka-configs.sh --alter --add-config min.insync.replicas=2

Problem natychmiastowych wyzwalaczy: ~14 zbędnych UPSERT na okno dla jednego klucza – koszt dla ujść i niekompatybilność z HDFS/S3/Parquet

Operator suppress(untilWindowCloses): warunek zamknięcia okna (watermark > end_w + grace), redukcja z ~14 do 1 emisji na okno

Nowy temat wewnętrzny KTABLE-SUPPRESS-STATE-STORE-changelog: bufor okien oczekujących na emisję, wiedza o tym co już wysłano

Porównanie liczby wiadomości w tematach: 100 wejściowych → ~150 aggregate-changelog → ~16 suppress-changelog → ~16+CM kafka-output

INSERT zamiast UPSERT przy suppress – kompatybilność z dowolnym typem ujść (HDFS, S3, Parquet, Delta Lake)

Budowanie fat JAR przy użyciu maven-shade-plugin: pakowanie z zależnościami, konfiguracja ManifestResourceTransformer

Uruchamianie aplikacji przez Docker Compose z osobnymi katalogami stanu dla każdej instancji