Hvordan Kafka ble virkelighet

Hei, Habr!

Eg jobbar i Tinkoff-teamet som utviklar vårt eige varslingssenter. For det meste utviklar eg i Java med Spring Boot og løser ulike tekniske problem som oppstår i prosjektet.

De fleste av våre mikrotjenester kommuniserer asynkront med kvarandre via ein meldingsbroker. Tidlegare brukte vi IBM MQ som broker, men den klarte ikkje å takle lasta lenger, sjølv om den hadde høge leveringsgarantiar.

Som erstatning fekk vi anbefalt Apache Kafka, som har høg skaleringspotensial, men som dessverre krev nærare ein individuell tilnærming til konfigurasjon for ulike scenario. I tillegg tillot mekanismen for minst ein gong levering, som fungerer i Kafka som standard, ikkje å oppretthalde det nødvendige nivået av konsistens ut av boksen. Seinare vil eg dele vår erfaring med konfigurering av Kafka, spesielt fortelje korleis ein kan sette opp og leve med nøyaktig ein gong levering.

Garantert levering og meir

Parameters som vil bli omtala vidare, vil hjelpe til med å unngå ei rekkje problem med standard tilkoplingsinnstillingane. Men først er det eit punkt eg gjerne vil fokusere på, som vil lette mogleg debugging.

Dette vil hjelpast av client.id for Producer og Consumer. Ved første augekast kan ein bruke applikasjonsnamnet som verdi, og i dei fleste tilfeller vil det fungere. Men situasjonen der applikasjonen har fleire Consumerar og de får same client.id, fører til følgjande advarsel:

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

Dersom du vil bruke JMX i ein applikasjon med Kafka, kan dette bli eit problem. For dette tilfellet er det best å bruke ein kombinasjon av applikasjonsnamn og til dømes topiknamnet som verdi for client.id. Resultatet av vår konfigurasjon kan sjåast i utdataene frå kommandoen kafka-consumer-groups frå verktøya frå Confluent:

No skal vi sjå på scenarioet for garantert levering av meldingar. Kafka Producer har parameteren acks, som lar deg konfigurere etter kor mange acknowledge leiarclusteret treng å rekne med at meldinga er skriven vel. Denne parameteren kan ta følgjande verdiar:

• 0 — acknowledge vil ikkje bli rekna.
• 1 — standardparameter, krev acknowledge berre frå 1 replika.
• −1 — bekreftelser fra alle synkroniserte replikater er nødvendig (klusterinnstilling) min.insync.replicas).

Av de oppførte verdiene er det klart at acks lik −1 gir de sterkeste garantiene for at meldingen ikke går tapt.

Som vi alle vet, er distribuerte systemer upålitelige. For å beskytte seg mot midlertidige feil, gir Kafka Producer parameteren retries, som lar deg spesifisere antall forsøk på å sende på nytt i løpet av delivery.timeout.ms. Siden parameteren retries har standardverdi Integer.MAX_VALUE (2147483647), kan antall gjenutsendte meldinger justeres ved kun å endre delivery.timeout.ms.

Vi beveger oss mot exactly once delivery

De nevnte innstillingene lar vår Producer levere meldinger med høy sikkerhet. La oss nå snakke om hvordan man kan garantere at bare én kopi av meldingen skrives til Kafka-topikken? I det enkleste tilfellet må Produceren sette parameteren enable.idempotence til true. Idempotens garanterer at bare én melding skrives til en bestemt partisjon av en topikk. Forutsetningen for å aktivere idempotens er verdiene acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Hvis disse alternativene ikke er spesifisert av utvikleren, vil de automatisk bli satt til de nevnte verdiene.

Når idempotens er konfigurert, må man sikre at identiske meldinger alltid havner i de samme partisjonene. Dette kan gjøres ved å konfigurere nøkkelen og parameteren partitioner.class på Producer. La oss begynne med nøkkelen. Den må være lik for hver sending. Dette er lett å oppnå ved å bruke en forretningsidentifikator fra den originale meldingen. Parameteren partitioner.class har standardverdi — DefaultPartitioner. Med denne standard partisjoneringsstrategien fungerer vi slik:

• Hvis partisjonen er spesifisert ved sending av meldingen, bruker vi den.
• Hvis partisjonen ikke er spesifisert, men nøkkelen er angitt — velger vi partisjonen basert på hash av nøkkelen.
• Hvis verken partisjonen eller nøkkelen er spesifisert — velger vi partisjonene på rundgang (round-robin).

I tillegg, bruk av nøkkel og idempotent sending med parameteren max.in.flight.requests.per.connection = 1 gir deg en organisert behandling av meldinger på Consumer. Det er også viktig å huske at hvis clusteret ditt har tilgangskontroll aktivert, vil du trenge rettigheter for idempotent skriving til topic.

Hvis du mangler muligheten til idempotent sending etter nøkkel, eller logikken på produsentsiden krever at du opprettholder konsistens i data mellom forskjellige partisjoner, kan transaksjoner hjelpe. I tillegg kan du med kjedetransaksjoner betinget synkronisere skriving til Kafka, for eksempel med skriving til DB. For å aktivere transaksjonell sending på produsenten må den være idempotent, og du må også angi transactional.id. Hvis Kafka-clusteret ditt har tilgangskontroll aktivert, trenger du som med idempotent skriving, skrivetilgang som kan gis via en maske ved å bruke verdien lagret i transactional.id.

Formelt kan du bruke hvilken som helst streng som transaksjonsidentifikator, for eksempel navnet på applikasjonen. Men hvis du kjører flere instanser av den samme applikasjonen med samme transactional.id, vil den første kjørende instansen bli stoppet med en feil, da Kafka vil anse den som en zombieprosess.

org.apache.kafka.common.errors.ProducerFencedException: Produsenten forsøkte en operasjon med en gammel epoke. Enten er det en nyere produsent med samme transactionalId, eller produsentens transaksjon har utløpt på broker.

For å løse dette problemet, legger vi til et suffiks til applikasjonsnavnet i form av vertsnavn, som vi får fra miljøvariablene.

Produsenten er konfigurert, men transaksjoner i Kafka styrer kun synligheten av meldingen. Uansett status på transaksjonen, blir meldingen umiddelbart plassert i topic, men har ekstra systemattributter.

For at slike meldinger ikke skal bli lest av Consumer for tidlig, må den stille inn parameteren isolation.level til verdien read_committed. En slik Consumer kan lese ikke-transaksjonelle meldinger som før, men transaksjonelle meldinger kun etter commit.
Hvis du har satt opp alle de førnevnte innstillingene, så har du konfigurert exactly once delivery. Gratulerer!

Men det er en annen nyanse. Transactional.id, som vi konfigurerte tidligere, er faktisk et prefiks for transaksjonen. På transaksjonslederen legges det til et sekvensnummer. Den resulterende identifikatoren utstedes ved transactional.id.expiration.ms, som er konfigurert på Kafka-klyngen og har en standardverdi på «7 dager». Hvis applikasjonen ikke har mottatt noen meldinger i løpet av denne tiden, vil du under det neste transaksjonsforsøket få InvalidPidMappingException. Etter dette vil transaksjonskoordinatoren gi et nytt sekvensnummer for neste transaksjon. Meldingen kan bli tapt hvis ikke InvalidPidMappingException blir riktig håndtert.

I stedet for oppsummeringer

Som man kan se, er det ikke nok bare å sende meldinger til Kafka. Man må velge en kombinasjon av parametere og være klar til å gjøre raske endringer. I denne artikkelen har jeg prøvd å vise i detalj oppsettet for exactly once delivery, og jeg har beskrevet noen konfigurasjonsproblemer med client.id og transactional.id som vi har møtt. Nedenfor er innstillingene for Produsent og Konsument oppsummert kort.

Produsent:

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

Konsument:

1. isolation.level = read_committed

For å minimere feil i fremtidige applikasjoner har vi laget et lag rundt spring-konfigurasjonen, hvor verdiene for noen av de nevnte parameterne allerede er satt.

Her er et par ressurser for selvstudium:

• KIP-98 — Exactly Once Delivery and Transactional Messaging
• Beskrivelse av innstillingene

Kilde: habr.com