Kā Kafka kļuva par realitāti

Sveiki, Habr!

Es strādāju Tinkoff komandā, kas izstrādā savu paziņojumu centru. Galvenokārt es programmēju Java, izmantojot Spring Boot, un risinu dažādas tehniskas problēmas, kas rodas projektā.

Lielākā daļa mūsu mikropakalpojumu savā starpā sazinās asinkroni caur ziņojumu brokeri. Agrāk mēs izmantojām IBM MQ kā brokeri, kurš vairs nespēja tikt galā ar slodzi, bet tam bija augsta piegādes garantija.

Kā aizvietojumu mums piedāvāja Apache Kafka, kas piedāvā augstas mērogojamības iespējas, taču, diemžēl, prasa praktiski individuālu pieeju konfigurācijai dažādiem scenārijiem. Turklāt mehānisms 'at least once delivery', kas darbojas Kafka pēc noklusējuma, neļāva nodrošināt nepieciešamo konsekvenci tieši no kastītes. Tālāk es dalīšos mūsu pieredzē par Kafka konfigurēšanu, īpaši par to, kā iestatīt un sadzīvot ar 'exactly once delivery'.

Garantēta piegāde un ne tikai

Parametri, par kuriem runāsim tālāk, palīdzēs novērst virkni problēmu ar noklusējuma savienojuma iestatījumiem. Bet vispirms vēlos pievērst uzmanību vienam parametram, kas atvieglos iespējamo atkļūdošanu.

Šajā palīdzēs client.id gan producentam, gan patērētājam. Sākotnēji kā vērtību var izmantot lietojumprogrammas nosaukumu, un lielākajā daļā gadījumu tas darbosies. Tomēr situācija, kad lietojumprogrammā tiek izmantoti vairāki patērētāji un jūs ievērojat viņiem vienādu client.id, noved pie sekojoša brīdinājuma:

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

Ja vēlaties izmantot JMX lietojumprogrammā ar Kafka, tas var radīt problēmas. Šajā gadījumā vislabāk ir izmantot kā vērtību client.id kombināciju no lietojumprogrammas nosaukuma un, piemēram, topika nosaukuma. Mūsu konfigurācijas rezultāts ir redzams komandas kafka-consumer-groups iz Confluent utilītēm:

Tagad aplūkosim garantētās ziņojumu piegādes scenāriju. Kafka producentam ir parametrs acks, kas ļauj iestatīt, pēc cik daudzām apstiprinājuma atbildēm klastera līderim ir nepieciešams uzskatīt ziņojumu par veiksmīgi ierakstītu. Šis parametrs var pieņemt sekojošas vērtības:

• 0 — apstiprinājumi netiks uzskaitīti.
• 1 — noklusējuma parametrs, nepieciešams apstiprinājums tikai no 1 replikas.
• −1 — nepieciešama apstiprināšana no visām sinhronizētajām replika (klastera iestatīšana min.insync.replicas).

No uzskaitītajiem vērtībām redzams, ka acks, kas ir −1, dod visstrongākās garantijas, ka ziņojums netiks zaudēts.

Kā mēs visi zinām, izplatītās sistēmas ir neuzticamas. Lai pasargātos no īslaicīgām kļūdām, Kafka Producer nodrošina parametru retries, kas ļauj noteikt atkārtotās nosūtīšanas mēģinājumu skaitu laikā delivery.timeout.ms. Tā kā parametrs retries pēc noklusējuma ir Integer.MAX_VALUE (2147483647), ziņojuma atkārtoto nosūtīšanu var regulēt, mainot tikai delivery.timeout.ms.

Pārejam pie exactly once delivery

Uzskaitītie iestatījumi ļauj mūsu Producer piegādāt ziņojumus ar augstu garantiju. Tagad runāsim par to, kā garantēt, ka tikai viena ziņojuma kopija tiek ierakstīta Kafka tēmas brīdī? Visvienkāršākajā gadījumā šim nolūkam Producer jāiestata parametrs enable.idempotence uz true. Idempotence garantē, ka konkrētā partition tiek ierakstīts tikai viens ziņojums. Priekšnoteikums idempotences ieslēgšanai ir vērtības acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Ja šie parametri nav noteikti izstrādātāja, tad automātiski tiks iestatītas iepriekš minētās vērtības.

Kad idempotence ir iestatīta, jāpanāk, lai identiski ziņojumi vienmēr nonāktu tajās pašās partition. To var izdarīt, iestatot atslēgu un parametru partitioner.class uz Producer. Sāksim ar atslēgu. Katram nosūtīšanas gadījumam tai jābūt vienādai. To ir viegli panākt, izmantojot kādu biznesa identifikatoru no oriģinālā ziņojuma. Parametra partitioner.class noklusējuma vērtība ir DefaultPartitioner. Šajā stratēģijā noklusējuma partionēšana tiek veikta šādi:

• Ja partition tiek skaidri norādīta ziņojuma nosūtīšanas laikā, tad to izmantojam.
• Ja partition nav norādīta, bet ir norādīta atslēga — izvēlamies partition, pamatojoties uz atslēgas hash.
• Ja tā partition un atslēga nav norādīti — izvēlamies partition pa kārtām (round-robin).

Turklāt atslēgas un idempotentās nosūtīšanas izmantošana ar parametru max.in.flight.requests.per.connection = 1 sniedz jums strukturētu ziņojumu apstrādi uz Consumer. Atsevišķi jāatceras, ka, ja jūsu klasterī ir iestatīta piekļuves kontrole, tad jums būs nepieciešamas tiesības uz idempotenta ieraksta izveidi tēmu.

Ja jums nejauši pietrūkst iespēju, ko sniedz idempotenta sūtīšana pēc atslēgas, vai loģika Producer pusē prasa saglabāt datu konsekvenci starp dažādām partīcijām, tad var palīdzēt transakcijas. Turklāt, izmantojot secīgas transakcijas, var nosacīti sinhronizēt ierakstu Kafka, piemēram, ar ierakstu datu bāzē. Lai aktivizētu transakciju nosūtīšanu uz Producer, ir jānodrošina, ka tam ir idempotence, un papildus jānorāda transactional.id. Ja jūsu Kafka klasterī ir iestatīta piekļuves kontrole, tad transakciju ierakstam, tāpat kā idempotentajam, būs nepieciešamas rakstīšanas tiesības, kuras var tikt piešķirtas saskaņā ar masku, izmantojot vērtību, kas tiek saglabāta transactional.id.

Formāli kā transakcijas identifikatoru var izmantot jebkuru virkni, piemēram, lietojumprogrammas nosaukumu. Bet, ja jūs palaižat vairākus viena lietojumprogrammas instancē ar vienādiem transactional.id, tad pirmais palaistais instances tiks apturēts ar kļūdu, jo Kafka to uzskatīs par zombiju procesu.

org.apache.kafka.common.errors.ProducerFencedException: Producers mēģināja veikt darbību ar vecu epoku. Vai nu ir jaunāks producer ar to pašu transactionalId, vai arī producenta transakcija ir beigusies brokera izpildē.

Lai atrisinātu šo problēmu, mēs pievienojam lietojumprogrammas nosaukumam sufiksu, kas ir hostname, ko iegūst no vides mainīgajiem.

Producer ir iestatīts, bet transakcijas uz Kafka pārvalda tikai ziņojuma redzamību. Neatkarīgi no transakcijas statusa ziņojums tūlītējā laikā nonāk tēma, bet tam ir papildu sistēmas atribūti.

Lai tādi ziņojumi netiktu lasīti no Consumer agrāk, tam ir jānoregulē parametrs isolation.level uz vērtību read_committed. Šāds Consumer varēs lasīt netransakcijas ziņojumus kā agrāk, bet transakcijas tikai pēc apstiprināšanas.
Ja esat iestatījis visas iepriekš minētās konfigurācijas, tad esat iestatījis exactly once delivery. Apsveicam!

Bet ir vēl viens nians, par kuru jādomā. Transactional.id, ko mēs iestatījām iepriekš, patiesībā ir transakcijas prefikss. Uz transakciju menedžera tam tiek pievienots secības numurs. Iegūtā identifikators tiek izsniegts uz transactional.id.expiration.ms, kas tiek konfigurēts uz Kafka klastera un tam ir noklusējuma vērtība „7 dienas”. Ja šajā laikā lietotne nav saņēmusi nevienu ziņu, tad, mēģinot veikt nākamo darījuma sūtījumu, jūs saņemsiet InvalidPidMappingException. Pēc tam darījumu koordinators piešķirs jaunu secības numuru nākamajam darījumam. Šajā gadījumā ziņa var tikt zaudēta, ja InvalidPidMappingException netiks pareizi apstrādāta.

Tā vietā, lai apkoptu rezultātus

Kā var pamanīt, nepietiek vienkārši nosūtīt ziņas uz Kafka. Ir jāizvēlas parametru kombinācija un jāsagatavojas ātrām izmaiņām. Šajā rakstā es centos detalizēti parādīt precisely once delivery iestatījumus un aprakstīt dažas problēmas ar client.id un transactional.id konfigurāciju, ar kurām mēs saskārāmies. Zemāk īsā formā ir sniegtas Producer un Consumer iestatījumi.

Producer:

1. acks = all
2. retries > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 — secīgai sūtīšanai)
5. transactional.id = ${application-name}-${hostname}

Consumer:

1. isolation.level = read_committed

Lai minimizētu kļūdas nākotnes lietotnēs, mēs izveidojām mūsu apvalku virs spring konfigurācijas, kurā jau ir noteiktas vērtības daudziem no uzskaitītajiem parametriem.

Un šeit ir pāris materiālu patstāvīgai izpētei:

• KIP-98 — Precisely Once Delivery un Darījumu Ziņošana
• Iestatījumu apraksts

Avots: habr.com