Kuidas Kafka sai reaalsuseks

Tere, Habr!

Töötan Tinkoffi meeskonnas, mis tegeleb enda teavituskeskuse arendamisega. Enamasti arendan Java-s, kasutades Spring Boot'i, ja lahendan erinevaid tehnilisi probleeme, mis projektis ilmnevad.

Enamik meie mikroteenustest suhtleb omavahel asünkroonselt sõnumite vahendaja kaudu. Varem kasutasime brokerina IBM MQ-d, mis ei suutnud koormuse tõttu enam hakkama saada, kuid pakkus siiski kõrgeid kohaletoimetamise garantiisid.

Asenduseks pakuti meile Apache Kafka, mis omab suurt skaleeritavuse potentsiaali, kuid nõuab paraku praktiliselt individuaalset lähenemist erinevate stsenaariumide seadistamisel. Lisaks ei võimaldanud vaikimisi töötav at least once delivery mehhanism Kafkas säilitada vajalikku järjepidevuse taset. Edasi jagan meie kogemust Kafka seadistamisel, milles räägin, kuidas seadistada ja elada exactly once delivery'ga.

Kohustatud kohaletoimetamine ja mitte ainult

Seadistused, millest räägime, aitavad vältida rida probleeme vaikeseadistustega. Kuid enne tahaksin pöörata tähelepanu ühele seadistusele, mis lihtsustab võimalikku veaotsingut.

Sellel aitab client.id Produceri ja Consumeri jaoks. Esmapilgul saab väärtusena kasutada rakenduse nime ja enamikul juhtudel see ka toimib. Siiski, olukord, kus rakenduses kasutatakse mitu Consumer'i ja seadistate neile ühe ja sama client.id, toob kaasa järgmise hoiatuse:

org.apache.kafka.common.utils.AppInfoParser — Veateade AppInfo mbean'i registreerimisel javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

Kui soovite kasutada JMX-i rakenduses Kafkaga, võib see olla probleem. Sellise juhtumi jaoks on kõige parem kasutada client.id väärtusena rakenduse nime ja näiteks teema nime kombinatsiooni. Meie seadistuse tulemust saab vaadata käsu kafka-consumer-groups Confluenti utiliidilt:

Nüüd vaatame üle sõnumi garantii kohaletoimetamise stsenaariumi. Kafkal on Produceril parameeter acks, mille abil saab seadistada, pärast kui palju tunnustusi peab klastriliider sõnumit edukaks kirjutamiseks lugema. Sellel parameetril võivad olla järgmised väärtused:

• 0 — tunnustusi ei arvestata.
• 1 — vaikimisi parameeter, nõuab tunnustust ainult 1 repliigilt.
• −1 — vajalik tunnustus kõikidelt sünkroniseeritud repliikidelt (klastriseade min.insync.replicas).

Loetletud väärtustest on näha, et acks, mille väärtus on −1, tagab kõige tugevama kinnituse, et sõnum ei kao.

Kuna me kõik teame, on jaotatud süsteemid ebausaldusväärsed. Ajutiste tõrgete kaitsmiseks pakub Kafka Producer parameetrit retries, mis võimaldab määrata saatmise katsete arvu jooksul delivery.timeout.ms. Kuna parameetri retries vaikimisi väärtus on Integer.MAX_VALUE (2147483647), saab sõnumi korduskatsete arvu reguleerida, muutes ainult delivery.timeout.ms.

Liigume edasi exactly once delivery

Loetletud seaded võimaldavad meie Producer'itel sõnumeid kõrge garantii järgi edastada. Räägime nüüd, kuidas tagada, et ainult üks koopia sõnumist salvestatakse Kafka teemas? Kõige lihtsamal juhul peab Producer seadma parameetri enable.idempotence tõeks. Idempotentsus tagab, et konstateeritakse ainult üks sõnum kindlasse partitsiooni ühes teemas. Idempotentsuse aktiveerimise eelduseks on väärtused acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Kui need parameetrid pole arendaja poolt määratud, määratakse automaatselt eespool loetletud väärtused.

Kui idempotentsus on seadistatud, tuleb tagada, et samad sõnumid jõuavad iga kord samadesse partitsioonidesse. Seda saab teha tootja võtme ja parameetri partitioner.class seadistamisega. Alustame võtmega. Iga saatmise korral peab see olema sama. Seda on lihtne saavutada, kasutades mõnda äridentifikaatorit algsest sõnumist. Parameetri partitioner.class vaikimisi väärtus on DefaultPartitioner. Selle vaikimisi partitsioneerimise strateegia kohaselt toimime järgmiselt:

• Kui partitsioon on selgesõnaliselt määratud sõnumi saatmise ajal, siis kasutame seda.
• Kui partitsioon ei ole määratud, kuid võti on määratud — valime partitsiooni võtme hash'i põhjal.
• Kui partitsioon ja võti ei ole määratud — valime partitsioonide järgnevuse alusel (round-robin).

Lisaks sellele põhjustab võti ja idempotentne saatmine parameetriga max.in.flight.requests.per.connection = 1 annab teie jaoks korrastatud sõnumite töötlemise Consumeris. Oluline on meeles pidada, et kui teie klastris on seadistatud juurdepääsu haldamine, vajate õigusi idempotentseks kirjutamiseks teemasse.

Kui teil on puudus idempotentsest saatmisest võtme järgi või Produceri poolel nõuab loogika erinevate partitsioonide vahel andmete järjepidevuse säilitamist, tulevad appi tehingud. Lisaks saab ketittehingute abil tinglikult sünkroonida kirjutamist Kafka's, näiteks andmebaasi kirjaga. Tehingulise saatmise lubamiseks peab Producer omama idempotentsust ja lisaks määrama transactional.id. Kui teie Kafka klastris on seadistatud juurdepääsu haldamine, vajate tehinguliste kirjutiste jaoks, nagu ka idempotentsete jaoks, kirjutamisõigusi, mis võivad olla antud maskeeri kaudu, kasutades väärtust, mis on salvestatud transactional.id-sse.

Formaalselt võib tehingu identifikaatorina kasutada mistahes stringi, näiteks rakenduse nime. Kuid kui käitate mitut sarnase nimega rakenduse instantsi sama transactional.id-ga, peatub esmakäivitunud instants tõrge tõttu, kuna Kafka peab seda zombiprotsessiks.

org.apache.kafka.common.errors.ProducerFencedException: Producer üritas toimingut vanas ajas. Või on olemas uuem tootja sama transactionalId-ga või tootja tehingu on brokeri poolt aegunud.

Selle probleemi lahendamiseks lisame rakenduse nimele sufiksi, milleks on hostinimi, mille saame keskkonnamuutujatest.

Producer on seadistatud, kuid tehingud Kafka's haldavad vaid sõnumi nähtavuse ulatus. Olenemata tehingu staatust on sõnum vahetult teema juurde viidud, kuid sellel on täiendavad süsteemi atribuudid.

Et selliseid sõnumeid ei loetaks Consumer'i poolt liiga vara, tuleb tal seada parameeter isolation.level väärtusele read_committed. Selline Consumer suudab lugeda mitte-tehingulisi sõnumeid nagu varem, kuid tehingulisi ainult pärast kinnitamist.
Kui olete seadistanud kõik eelnevalt loetletud seaded, olete seadistanud exactly once delivery. Palju õnne!

Kuid on veel üks nüanss. Transactional.id, mida me ülal seadistasime, on tegelikult tehingu eelosa. Tehingu halduri juurde lisatakse sellele järjestusnumber. Saadud identifikaator väljastatakse transactional.id.expiration.ms, mis on konfigureeritud Kafka klastris ja mille vaikimisi väärtus on „7 päeva“. Kui selle aja jooksul rakendus ei saa mingeid sõnumeid, siis järgmise tehingu saatmise katse korral saate InvalidPidMappingException. Pärast seda annab tehingute koordineerija järgmise tehingu jaoks uue järjestusnumbri. Samuti võib sõnum kaduda, kui InvalidPidMappingException ei käsitleta õigesti.

Kokkuvõtete asemel

Nagu näha, ei piisa lihtsalt sõnumite saatmisest Kafka-sse. Tuleb valida parameetrite kombinatsioon ja olla valmis kiireteks muutusteks. Selles artiklis püüdsin detailides näidata exactly once delivery seadistust ja kirjeldasin mitmeid client.id ja transactional.id konfiguratsiooni probleeme, millega me kokku puutusime. Allpool on lühidalt toodud tootja ja tarbija seadistused.

Tootja:

1. acks = kõik
2. retries > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 — järjekorras saatmiseks)
5. transactional.id = ${application-name}-${hostname}

Tarbija:

1. isolation.level = read_committed

Tulevaste rakenduste vigade minimeerimiseks lõime oma mähise spring-konfiguratsiooni ümber, kus on juba seatud väärtused mõnele ülalnimetatud parameetrile.

Ja siin on paar materjali iseseisvaks õppimiseks:

• KIP-98 — Exactly Once Delivery and Transactional Messaging
• Seadistuste kirjeldus

Allikas: habr.com