Hello, Habr!
Delam v timu Tinkoff, ki se ukvarja z razvojem lastnega centra za obvestila. Večinoma razvijam v Javi z uporabo Spring boota in reševam različne tehnične težave, ki se pojavijo v projektu.
Večina naših mikroservisov asinkrono komunicira drug z drugim preko sporočilnega posrednika. Prej smo uporabljali IBM MQ, ki ni več zmožen obvladovati bremena, a je hkrati zagotavljal visoke stopnje dostave.
Kot zamenjavo so nam predlagali Apache Kafka, ki ponuja veliko potenciala za skaliranje, a zahteva praktično individualen pristop k konfiguraciji za različne scenarije. Poleg tega mehanizem 'at least once delivery', ki deluje v Kafki privzeto, ne omogoča vzdrževanja potrebne ravni konsistence kar tako. V nadaljevanju bom delil naše izkušnje s konfiguracijo Kafke, še posebej kako nastaviti in živeti z 'exactly once delivery'.
Zagotovljena dostava in še več
Parametri, o katerih bomo govorili naprej, bodo pomagali preprečiti vrsto težav s privzetimi nastavitvami povezav. A najprej bi rad osvetlil en parameter, ki bo olajšal morebitno odpravljanje napak.
Pri tem bo v pomoč client.id za Producer in Consumer. Na prvi pogled je možno uporabiti ime aplikacije kot vrednost, kar bo v večini primerov delovalo. Vendar situacija, ko v aplikaciji uporabljate več Consumer-jev in jim dodelite enako client.id, vodi do naslednjega opozorila:
org.apache.kafka.common.utils.AppInfoParser — Napaka pri registraciji AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Če želite uporabiti JMX v aplikaciji s Kafko, je to lahko težava. Za to situacijo je najbolje uporabiti kombinacijo imena aplikacije in, na primer, imena topic-a kot vrednost za client.id. Rezultat naše konfiguracije lahko vidite v izhodu ukaza kafka-consumer-groups iz orodij podjetja Confluent:
Zdaj razmislimo o scenariju zagotavljene dostave sporočila. Kafka Producer ima parameter acks, ki omogoča nastavitev, po katerem številu potrditev mora vodja gruče šteti sporočilo za uspešno zapisano. Ta parameter lahko sprejme naslednje vrednosti:
- 0 — potrditev ne bo štela.
- 1 — privzeta nastavitev, zahtevana je potrditev le s strani 1 replike.
- −1 — potrebna so potrdila od vseh sinhroniziranih replik (nastavitev grozda min.insync.replicas).
Iz navedenih vrednosti je razvidno, da acks enako −1 zagotavlja najmočnejše jamstvo, da sporočilo ne bo izgubljeno.
Kot vemo, so distribuirani sistemi nezanesljivi. Da bi se zaščitili pred začasnimi napakami, Kafka Producer ponuja parameter retries, ki omogoča določitev števila poskusov ponovnega pošiljanja v okviru delivery.timeout.ms. Ker ima parameter retries privzeto vrednost Integer.MAX_VALUE (2147483647), lahko število ponovnih pošiljanj sporočila reguliramo tako, da spreminjamo le delivery.timeout.ms.
Prehajamo na exactly once delivery
Nastavitve omogočajo našemu Producerju dostavo sporočil z visokim jamstvom. Zdaj pa se pogovorimo o tem, kako zagotoviti zapis samo ene kopije sporočila v Kafka-topik? V najpreprostejšem primeru je za to potrebno, da na Producerju nastavite parameter enable.idempotence na vrednost true. Idempotentnost zagotavlja zapis samo enega sporočila v določeno particijo enega topika. Predhodni pogoj za vključitev idempotentnosti so vrednosti acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Če teh parametrov ne nastavi razvijalec, se avtomatsko nastavi zgoraj navedene vrednosti.
Ko je idempotentnost nastavljena, je treba zagotoviti, da enaka sporočila vedno pridejo v iste particije. To lahko dosežemo z nastavitvijo ključa in parametra partitioner.class na Producerju. Začnimo s ključem. Pri vsakem pošiljanju mora biti enak. To lahko enostavno dosežemo z uporabo katerega koli poslovnega identifikatorja iz izvirnega sporočila. Parameter partitioner.class ima privzeto vrednost — . Pri tej strategiji particioniranja po privzeti nastavitvi delujemo tako:
- Če je particija izrecno navedena pri pošiljanju sporočila, jo uporabimo.
- Če particija ni navedena, ampak je naveden ključ — izberemo particijo po hashu ključa.
- Če particija in ključ nista navedena — izberemo particije po vrsti (round-robin).
Poleg tega uporaba ključa in idempotentnega pošiljanja s parametrom max.in.flight.requests.per.connection = 1 vam daje uređen pristup porukama na Consumeru. Takođe, važno je napomenuti da, ako je na vašem klasteru podešena kontrola pristupa, biće vam potrebna prava za idempotentno pisanje u temu.
Ako vam nedostaju mogućnosti idempotentnog slanja po ključnoj reči ili logika na strani Producenta zahteva očuvanje doslednosti podataka između različitih particija, transakcije će vam pomoći. Pored toga, uz pomoću lchain transaction može se uslovno sinhronizovati zapis u Kafka-u, na primer, sa zapisom u bazi podataka. Da biste uključili transakcijsko slanje na Producentu, potrebno je da on bude idempotentan i dodatno da definiše transactional.id. Ako je na vašem Kafka klasteru postavljena kontrola pristupa, za transakcijsko pisanje, kao i za idempotentno pisanje, biće vam potrebna prava za pisanje koja se mogu dodeliti prema maski koristeći vrednost koja se čuva u transactional.id.
Formalno, kao identifikator transakcije može se koristiti bilo koji string, na primer ime aplikacije. Ali ako pokrećete više instanci jedne aplikacije sa istim transactional.id, prva pokrenuta instanca će biti zaustavljena sa greškom, jer će Kafka smatrati da je to zombi proces.
org.apache.kafka.common.errors.ProducerFencedException: Producent je pokušao operaciju sa starom epohom. Ili postoji noviji producent sa istim transactionalId, ili je transakcija proizvođača istekla od strane brokera.Da bismo rešili ovaj problem, dodajemo sufiks u ime aplikacije koji se sastoji od imena hosta, dobijenog iz promenljivih okruženja.
Producent je podešen, ali transakcije u Kafki upravljaju samo obuhvatom vidljivosti poruke. Bez obzira na status transakcije, poruka odmah ulazi u temu, ali sa dodatnim sistemskim atributima.
Da takve poruke ne bi bile očitane od strane Consumer-a pre vremena, potrebno je postaviti parametar isolation.level na vrednost read_committed. Takav Consumer moći će da čita netransakcionisane poruke kao i ranije, a transakcionisane – samo nakon potvrde.
Ako ste postavili sve ranije navedene postavke, onda ste konfigurisali exactly once delivery. Čestitam!
Ali postoji još jedan aspekt. Transactional.id, koji smo ranije konfigurisali, zapravo je prefiks transakcije. Na menadžeru transakcija se dodaje redni broj. Dobijeni identifikator se dodeljuje na transactional.id.expiration.ms, ki se konfigurira na Kafka grozdu in ima privzeto vrednost „7 dni“. Če aplikacija v tem času ne prejme nobenih sporočil, boste ob naslednjem poskusu transakcijskega pošiljanja prejeli InvalidPidMappingException. Po tem bo koordinator transakcij dodelil novo zaporedno številko za naslednjo transakcijo. Pri tem lahko pride do izgube sporočila, če InvalidPidMappingException ne bo pravilno obvladan.
Namesto povzetkov
Kot lahko opazite, ni dovolj samo pošiljati sporočila v Kafka. Potrebno je izbrati kombinacijo parametrov in biti pripravljen na hitre spremembe. V tem članku sem poskušal podrobno prikazati nastavitev 'exactly once delivery' in opisal nekaj težav konfiguracij client.id in transactional.id, s katerimi smo se srečali. Spodaj so v kratki obliki navedene nastavitve Producer in Consumer.
Producer:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 — za urejeno pošiljanje)
- transactional.id = ${application-name}-${hostname}
Consumer:
- isolation.level = read_committed
Da bi zmanjšali napake v prihodnjih aplikacijah, smo naredili svojo oviro nad spring konfiguracijo, kjer so že nastavljene vrednosti za nekatere od naštetih parametrov.
Tukaj je nekaj gradiv za samostojno študij:
Izvor: habr.com
