🥇Ako sa Kafka stala skutočnosťou

Ako sa Kafka stala realitou

Ahoj, Habr!

Pracujem v tíme Tinkoff, ktorý sa zaoberá vývojom vlastného centra notifikácií. Väčšinou vyvíjam v Jave pomocou Spring boot a riešim rôzne technické problémy, ktoré sa v projekte objavujú.

Väčšina našich mikroservisov asynchrónne interaguje navzájom prostredníctvom správy. Predtým sme ako broker používali IBM MQ, ktorý už nezvládal zaťaženie, ale mal vysoké záruky dodania.

Na jeho nahradenie nám navrhli Apache Kafka, ktorá má vysoký potenciál na škálovanie, ale bohužiaľ si vyžaduje prakticky individuálny prístup k konfigurácii pre rôzne scenáre. Okrem toho mechanzimus at least once delivery, ktorý v Kafke pracuje ako predvolené nastavenie, neumožnil udržiavať potrebnú úroveň konzistencie priamo z krabice. Ďalej sa podelím o naše skúsenosti s konfiguráciou Kafky, konkrétne o tom, ako nastaviť a žiť s exactly once delivery.

Zaručené dodanie a nielen to

Parametre, o ktorých sa bude hovoriť ďalej, pomôžu zabrániť množstvu problémov s predvolenými pripojovacími nastaveniami. Ale najprv chcem venovať pozornosť jednému parametru, ktorý uľahčí možný debugging.

S týmto pomôže client.id pre Producer a Consumer. Na prvý pohľad môže byť ako hodnota použité meno aplikácie, a vo väčšine prípadov to bude fungovať. Hoci situácia, keď aplikácia používa viacero Consumerov a vy im nastavíte rovnaký client.id, vedie k nasledujúcemu varovaniu:

org.apache.kafka.common.utils.AppInfoParser — Chyba pri registrácii AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

Ak chcete v aplikácii s Kafkou používať JMX, môže to byť problém. Pre tento prípad je najlepšie použiť ako hodnotu client.id kombináciu mena aplikácie a napríklad mena topiku. Výsledok našej konfigurácie si môžete pozrieť vo výstupu príkazu kafka-consumer-groups z nástrojov od Confluent:

Ako sa Kafka stala realitou

Teraz si prejdime scenár zaručeného dodania správy. Kafka Producer má parameter acks, ktorý umožňuje nastaviť, po koľkých potvrdeniach by mal líder klastru považovať správu za úspešne zapísanú. Tento parameter môže nadobúdať nasledujúce hodnoty:

0 — potvrdenia sa nebudú brať do úvahy.
1 — predvolené nastavenie, potvrdenie je potrebné len od 1 repliky.
−1 — je potrebné potvrdenie od všetkých synchronizovaných replik (nastavenie klastera min.insync.replicas).

Z uvedených hodnôt je vidieť, že acks rovný −1 poskytuje najsilnejšie záruky, že správa nebude stratená.

Ako všetci vieme, distribuované systémy nie sú spoľahlivé. Aby sme sa chránili pred dočasnými poruchami, Kafka Producer poskytuje parameter retries, ktorý umožňuje nastaviť počet pokusov o opätovné odoslanie počas delivery.timeout.ms. Keďže parameter retries má predvolenú hodnotu Integer.MAX_VALUE (2147483647), počet opätovných odoslaní správy sa dá regulovať, zmenou len delivery.timeout.ms.

Prechádzame k presne raz dodávke

Uvedené nastavenia umožňujú nášmu Producerovi dodávať správy s vysokou zárukou. Poďme teraz hovoriť o tom, ako zabezpečiť zápis len jednej kópie správy do Kafka-topic? V najjednoduchšom prípade je potrebné na Produceri nastaviť parameter enable.idempotence na hodnotu true. Idempotencia zaručuje zápis len jednej správy do konkrétnej partície jedného topicu. Predpokladom pre zapnutie idempotencie sú hodnoty acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Ak tieto parametre nie sú nastavené vývojárom, automaticky sa nastavia uvedené vyššie hodnoty.

Keď je idempotencia nastavená, je potrebné dosiahnuť, aby rovnaké správy prichádzali zakaždým do tých istých partícií. Toho možno dosiahnuť nastavením kľúča a parametru partitioner.class na Producer. Poďme začať s kľúčom. Pre každé odoslanie by mal byť rovnaký. To sa ľahko dosiahne použitím nejakého obchodného identifikátora z pôvodnej správy. Parameter partitioner.class má predvolenú hodnotu — DefaultPartitioner. Pri tejto stratégii paritizovania v predvolenom nastavení postupujeme takto:

Ak je partícia Explicitne určená pri odoslaní správy, použijeme ju.
Ak partícia nie je určená, ale kľúč je určený — vyberieme partíciu podľa hash kľúča.
Ak nie sú určené ani partícia, ani kľúč — vyberáme partície podľa poradia (round-robin).

Okrem toho, použitie kľúča a idempotentného odosielania s parametrom max.in.flight.requests.per.connection = 1 poskytuje usporiadané spracovanie správ na Consumer. Je dôležité si uvedomiť, že ak máte na svojom klastri nastavené riadenie prístupu, potrebujete práva na idempotentné zapisovanie do témy.

Ak náhodou potrebujete viac možností idempotentného odosielania podľa kľúča alebo logika na strane Producer vyžaduje zachovanie konzistencie dát medzi rôznymi partiami, transakcie prichádzajú na pomoc. Okrem toho pomocou reťazovej transakcie môžete podmienečne synchronizovať zápis do Kafka, napríklad so zápisom do DB. Na zapnutie transakčného odosielania na Producer musí byť idempotentný a navyše pridať transactional.id. Ak máte vo svojom Kafka klastri nastavené riadenie prístupu, potom pre transakčné zapisovanie, rovnako ako pre idempotentné, budete potrebovať práva na zápis, ktoré môžu byť poskytnuté pomocou masky s použitím hodnoty uložené v transactional.id.

Formálne ako identifikátor transakcie môžete použiť akýkoľvek reťazec, napríklad názov aplikácie. Ale ak spúšťate niekoľko inštancií tej istej aplikácie s rovnakým transactional.id, prvá spustená inštancia sa zastaví s chybou, pretože Kafka ju považuje za zombie proces.

org.apache.kafka.common.errors.ProducerFencedException: Producer sa pokúsil o operáciu so starým epochom. Buď existuje novší producer so rovnakým transactionalId, alebo transakcia producer bola vypršaná brokerom.

Aby sme tento problém vyriešili, pridáme k názvu aplikácie suffix vo forme názvu hostiteľa, ktorý získame z premenných prostredia.

Producer je nastavený, ale transakcie v Kafka riadia iba oblasť viditeľnosti správy. Bez ohľadu na stav transakcie správa okamžite putuje do témy, ale má ďalšie systémové atribúty.

Aby takéto správy neboli prečítané Consumerom predčasne, musí si nastaviť parameter isolation.level na hodnotu read_committed. Takýto Consumer môže čítať netransakčné správy ako predtým, ale transakčné iba po commite.
Ak ste nastavili všetky predchádzajúce nastavenia, nastavili ste exactly once delivery. Gratulujem!

Ale existuje ešte jeden detail. Transactional.id, ktorý sme nastavili vyššie, je v skutočnosti prefix transakcie. K nemu sa v manažéri transakcií dopisuje poradové číslo. Získaný identifikátor sa vydáva na transactional.id.expiration.ms, ktorý sa konfiguruje na Kafka klasteri a má predvolenú hodnotu „7 dní“. Ak počas tohto obdobia aplikácia nedostáva žiadne správy, pri pokuse o nasledujúce transakčné odoslanie dostanete InvalidPidMappingException. Po tom koordinátor transakcií pridelí nové poradové číslo pre nasledujúcu transakciu. V tomto prípade môže byť správa stratená, ak nebude InvalidPidMappingException správne spracovaný.

Namiesto súhrnov

Ako môžete vidieť, nestačí len posielať správy do Kafka. Je potrebné vybrať kombináciu parametrov a byť pripravený na rýchle zmeny. V tomto článku som sa podrobne snažil ukázať nastavenie exactly once delivery a opísal niekoľko problémov konfigurácií client.id a transactional.id, s ktorými sme sa stretli. Nižšie v stručnej forme sú uvedené nastavenia Producer a Consumer.

Producer:

acks = all
retries > 0
enable.idempotence = true
max.in.flight.requests.per.connection ≤ 5 (1 — pre usporiadané odosielanie)
transactional.id = ${application-name}-${hostname}

Consumer:

isolation.level = read_committed

Aby sme minimalizovali chyby v budúcich aplikáciách, vytvorili sme si vlastný wrapper nad spring konfiguráciou, kde sú už nastavené hodnoty pre niektoré z uvedených parametrov.

A tu je pár materiálov na samostatné štúdium:

Zdroj: habr.com