Kako je Kafka postala stvarnost

Bok, Habr!

Radim u timu Tinkoff koji se bavi razvojem vlastitog centra za obavijesti. Većinom programiram u Javi koristeći Spring Boot i rješavam razne tehničke probleme koji se pojavljuju u projektu.

Većina naših mikroservisa asinkrono međusobno komunicira putem posrednika poruka. Ranije smo koristili IBM MQ kao posrednika, koji je prestao odolijevati opterećenju, iako je imao visoka jamstva isporuke.

Kao zamjenu nam je predložen Apache Kafka, koja ima visoki potencijal za skaliranje, ali nažalost zahtijeva gotovo individualni pristup konfiguraciji za različite scenarije. Osim toga, mehanizam 'at least once delivery', koji radi u Kafki prema defaultu, nije omogućavao održavanje potrebnog nivoa konzistentnosti iz kutije. U nastavku ću podijeliti naše iskustvo s konfiguracijom Kafke, a posebno ću opisati kako postaviti i živjeti s 'exactly once delivery'.

Jamčena isporuka i ne samo to

Parametri o kojima ćemo govoriti dalje će pomoći u sprječavanju niza problema s postavkama povezivanja prema zadanim postavkama. No prvo bih želio obratiti pažnju na jedan parametar koji će olakšati moguću dijagnostiku.

U tome će pomoći client.id za Producer i Consumer. Na prvi pogled, kao vrijednost se može koristiti ime aplikacije, i u većini slučajeva to će funkcionirati. Međutim, situacija kada aplikacija koristi više Consumer-a i za njih postavljate isti client.id dovodi do sljedeće upozorenja:

org.apache.kafka.common.utils.AppInfoParser — Greška pri registraciji AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

Ako želite koristiti JMX u aplikaciji s Kafkom, to može predstavljati problem. U tom slučaju najbolje je koristiti kombinaciju imena aplikacije i, na primjer, imena topica kao vrijednost client.id. Rezultat naše konfiguracije možete vidjeti u izlazu komande kafka-consumer-groups iz Confluentovih alata:

Sada ćemo razmotriti scenarij jamčene isporuke poruke. Kafka Producer ima parametar acks, koji omogućava postavljanje nakon koliko potvrda lideru klastera treba smatrati poruku uspješno zapisanu. Ovaj parametar može imati sljedeće vrijednosti:

• 0 — potvrde se neće računati.
• 1 — parametar po defaultu, potrebna je potvrda samo od 1 replike.
• −1 — zahtijevaju se acknowledge od svih sinkroniziranih replika (konfiguracija klastera min.insync.replicas).

Iz navedenih vrijednosti jasno je da acks jednako −1 pruža najjače garancije da poruka neće biti izgubljena.

Kao što svi znamo, distribuirani sustavi nisu pouzdani. Kako bismo se zaštitili od povremenih kvarova, Kafka Producer nudi parametar retries, koji omogućuje postavljanje broja pokušaja ponovnog slanja tijekom delivery.timeout.ms. Budući da parametar retries ima zadanu vrijednost Integer.MAX_VALUE (2147483647), broj ponovnih slanja poruke može se regulirati promjenom samo delivery.timeout.ms.

Idemo prema exactly once delivery

Navedene postavke omogućuju našem Produceru da dostavlja poruke s visokom garancijom. Sada razgovarajmo o tome kako osigurati da se samo jedna kopija poruke zapiše u Kafka-topik? U najjednostavnijem slučaju, za to treba postaviti parametar enable.idempotence na true. Idempotentnost jamči da se samo jedna poruka zapise u određenu particiju jednog topika. Preduvjet za uključivanje idempotentnosti su vrijednosti acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Ako ovi parametri nisu postavljeni od strane programera, automatski će se postaviti navedene vrijednosti.

Kada je idempotentnost postavljena, potrebno je osigurati da iste poruke svaki put završe u istim particijama. To se može postići konfiguriranjem ključa i parametra partitioner.class na Produceru. Započnimo s ključem. Za svako slanje mora biti isti. To se lako postiže korištenjem bilo kojeg poslovnog identifikatora iz originalne poruke. Parametar partitioner.class ima zadanu vrijednost — DefaultPartitioner. Pri ovoj strategiji particioniranja prema zadanim postavkama djelujemo ovako:

• Ako je particija izričito navedena prilikom slanja poruke, koristimo je.
• Ako particija nije navedena, ali je naveden ključ — biramo particiju prema heširanju od ključa.
• Ako particija i ključ nisu navedeni — biramo particije naizmjenično (round-robin).

Osim toga, korištenje ključa i idempotentnog slanja s parametrima max.in.flight.requests.per.connection = 1 pruža vam strukturiranu obradu poruka na Consumeru. Također je važno napomenuti da, ako je na vašem klasteru postavljena kontrola pristupa, trebate imati prava za idempotentno pisanje u temu.

Ako vam nedostaje mogućnosti idempotentnog slanja po ključevoj ili logika na strani Producenta zahtijeva očuvanje dosljednosti podataka između različitih particija, tu su transakcije kao pomoć. Osim toga, putem lančane transakcije možete uvjetno sinkronizirati zapis u Kafki, na primjer, sa zapisom u bazi podataka. Da biste uključili transakcijsko slanje na Producentu, potrebno je da on ima idempotentnost i dodatno postaviti transactional.id. Ako je na vašem Kafka klasteru postavljena kontrola pristupa, tada su, kao i za idempotentno, potrebna prava za pisanje koja se mogu dodijeliti putem maske koristeći vrijednost pohranjenu u transactional.id.

Formalno kao identifikator transakcije možete koristiti bilo koji niz, kao što je ime aplikacije. No, ako pokrećete nekoliko instancija jedne aplikacije s istim transactional.id, prva pokrenuta instanca će biti zaustavljena s greškom, jer će Kafka smatrati da je to zombirani proces.

org.apache.kafka.common.errors.ProducerFencedException: Producent je pokušao izvršiti operaciju sa starom epohom. Ili postoji noviji producent s istim transactionalId, ili je transakcija producenta istekla od strane poslužitelja.

Da bismo riješili ovaj problem, dodajemo sufiks imenu aplikacije u obliku naziva hosta koji dobivamo iz varijabli okruženja.

Producent je postavljen, ali transakcije na Kafki upravljaju samo područjem vidljivosti poruke. Bez obzira na status transakcije, poruka odmah ulazi u temu, ali ima dodatne sistemske atribute.

Da takve poruke ne bi bile očitane od strane Consumer-a prerano, potrebno je postaviti parametar isolation.level na vrijednost read_committed. Takav Consumer moći će čitati netransakcijske poruke kao i dosad, a transakcijske samo nakon potvrde.
Ako ste postavili sve prethodno navedene postavke, postavili ste exactly once delivery. Čestitam!

Ali postoji još jedan detalj. Transactional.id, koji smo postavili ranije, zapravo je prefiks transakcije. Na upravitelju transakcija dodaje se redni broj. Dobiveni identifikator se izdaje na transactional.id.expiration.ms, koji se konfigurira na Kafka klasteru i ima zadanu vrijednost „7 dana”. Ako tijekom tog vremena aplikacija nije primila nikakve poruke, prilikom sljedeće transakcijske pošiljke dobit ćete InvalidPidMappingException. Nakon toga koordinator transakcija dodijelit će novi redni broj za sljedeću transakciju. Pri tome poruka može biti izgubljena ako se InvalidPidMappingException ne obrađuje ispravno.

Umjesto zaključaka

Kao što se može primijetiti, nije dovoljno samo slati poruke u Kafka-u. Potrebno je odabrati kombinaciju parametara i biti spreman na brze promjene. U ovom članku nastojao sam detaljno prikazati postavke exactly once delivery i opisati nekoliko problema s konfiguracijama client.id i transactional.id s kojima smo se susreli. U nastavku su u kratkom obliku navedene postavke Producer-a i Consumer-a.

Producer:

1. acks = all
2. retries > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 — za uređeno slanje)
5. transactional.id = ${application-name}-${hostname}

Consumer:

1. isolation.level = read_committed

Kako bismo umanjili greške u budućim aplikacijama, izradili smo vlastitu oblogu nad spring-konfiguracijom, gdje su već postavljene vrijednosti za neke od navedenih parametara.

Evo nekoliko materijala za samostalno učenje:

• KIP-98 — Exactly Once Delivery and Transactional Messaging
• Opis postavki

Izvor: habr.com