Hello, Habr!
Jeg arbejder i Tinkoff-teamet, der udvikler vores egen notifikationscentral. Størstedelen af tiden arbejder jeg med Java ved hjælp af Spring Boot og løser forskellige tekniske udfordringer, der opstår i projektet.
De fleste af vores mikrotjenester interagerer asynkront med hinanden gennem en beskedbroker. Tidligere anvendte vi IBM MQ som broker, men den kunne ikke længere håndtere belastningen, selvom den havde høje leveringsgarantier.
Som erstatning blev Apache Kafka foreslået, som har et højt skaleringspotentiale, men desværre kræver en næsten individuel tilgang til konfiguration til forskellige scenarier. Desuden gjorde mekanismen for 'at least once delivery', som fungerer i Kafka som standard, det ikke muligt at opretholde det nødvendige konsistensniveau lige ud af boksen. Her vil jeg dele vores erfaring med konfiguration af Kafka, især hvordan man opsætter og lever med 'exactly once delivery'.
Garanteret levering og mere til
De parametre, der vil blive diskuteret her, vil hjælpe med at forhindre en række problemer med de standardforbindelsesindstillinger. Men først vil jeg gerne fokusere på én parameter, der vil lette mulig debugging.
Det vil hjælpe med client.id for Producer og Consumer. Ved første øjekast kan applikationens navn bruges som værdi, og i de fleste tilfælde vil det fungere. Men situationen, hvor flere Consumers bruger det samme client.id i applikationen, vil resultere i følgende advarsel:
org.apache.kafka.common.utils.AppInfoParser — Fejl ved registrering af AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Hvis du vil bruge JMX i en applikation med Kafka, kan det være et problem. I dette tilfælde er det bedst at bruge en kombination af applikationens navn og for eksempel emnets navn som værdien for client.id. Resultatet af vores konfiguration kan ses i output fra kommandoen kafka-consumer-groups fra Confluents værktøjer:
Nu lad os gennemgå scenariet for garanteret levering af beskeder. En Kafka Producer har parameteren acks, som giver mulighed for at indstille, efter hvor mange acknowledged, lederen af klyngen skal betragte beskeden som vellykket skrevet. Denne parameter kan antage følgende værdier:
- 0 — acknowledgments vil ikke blive talt med.
- 1 — standard parameter, acknowledgment is only required from 1 replica.
- −1 — acknowledgment is required from all synchronized replicas (cluster configuration min.insync.replicas).
From the listed values, it is clear that an acks value of −1 provides the strongest guarantees that the message will not be lost.
As we all know, distributed systems are unreliable. To protect against temporary failures, the Kafka Producer provides the parameter retries, which allows you to set the number of retry attempts within delivery.timeout.ms. Since the retries parameter has a default value of Integer.MAX_VALUE (2147483647), the number of message retries can be regulated by changing only the delivery.timeout.ms.
Now let's move on to exactly once delivery.
The listed settings allow our Producer to deliver messages with high assurance. Now let's talk about how to guarantee the writing of only one copy of a message to a Kafka topic. In the simplest case, the Producer needs to set the parameter enable.idempotence to true. Idempotence guarantees that only one message will be written to a specific partition of one topic. The prerequisites for enabling idempotence are the values acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. If these parameters are not set by the developer, the specified values will be applied automatically.
When idempotence is configured, it is necessary to ensure that identical messages always end up in the same partitions. This can be done by configuring the key and the partitioner.class parameter on the Producer. Let's start with the key. It must be the same for each send operation. This is easily achieved by using some business identifier from the original message. The partitioner.class parameter has a default value of . With this default partitioning strategy, we proceed as follows:
- If the partition is explicitly specified when sending a message, we use it.
- If the partition is not specified but a key is provided — we choose the partition based on the hash of the key.
- If neither the partition nor the key is specified — we choose partitions in a round-robin manner.
Additionally, using the key and idempotent sending with the parameter max.in.flight.requests.per.connection = 1 giver dig systematisk behandling af meddelelser på Consumer. Det er vigtigt at huske, at hvis der er indstillet adgangskontrol på dit cluster, vil du have brug for rettigheder til idempotent skrivning til emnet.
Hvis du pludselig mangler muligheder for idempotent afsendelse efter nøgle, eller logikken på Producer-siden kræver, at dataenes konsistens bevares mellem forskellige partitioner, vil transaktioner hjælpe. Derudover kan kædetransaktioner betinget synkronisere skrivning i Kafka, for eksempel med skrivning til databasen. For at aktivere transaktionel afsendelse på Produceren skal den være idempotent og yderligere angive transactional.id. Hvis der er indstillet adgangskontrol på dit Kafka-cluster, kræves der også skriveadgang for transaktionelle skrivninger, ligesom for idempotente, som kan gives baseret på en maske ved hjælp af værdien, der er gemt i transactional.id.
Formelt kan enhver streng bruges som transaktions-id, for eksempel applikationens navn. Men hvis du kører flere instanser af den samme applikation med det samme transactional.id, vil den første kørte instans blive stoppet med en fejl, da Kafka vil betragte den som en zombie-proces.
org.apache.kafka.common.errors.ProducerFencedException: Producer forsøgte en operation med en gammel epoch. Enten er der en nyere producer med det samme transactionalId, eller også er producerens transaktion udløbet af broker'en.For at løse dette problem tilføjer vi et suffiks til applikationens navn i form af værtens navn, som vi får fra miljøvariabler.
Produceren er indstillet, men transaktioner i Kafka styrer kun meddelelsens synlighed. Uanset transaktionens status kommer meddelelsen straks ind i emnet, men har ekstra systemattributter.
For at sikre, at disse meddelelser ikke læses af Consumer for tidligt, skal det indstille parameteren isolation.level til værdien read_committed. En sådan Consumer kan læse ikke-transaktionelle meddelelser som før, mens transaktionelle kun kan læses efter commit-ing.
Hvis du har indstillet alle de førnævnte indstillinger, så har du konfigureret exactly once delivery. Tillykke!
Men der er endnu en detalje. Transactional.id, som vi indstillede ovenfor, er faktisk et præfiks for transaktionen. Til transaktionsmanageren tilføjes et serienummer. Den resulterende identifikator udstedes på transactional.id.expiration.ms, som er konfigureret på et Kafka-kluster og har en standardværdi på "7 dage". Hvis applikationen ikke har modtaget nogen meddelelser i denne periode, vil du ved næste transaktionelle forsendelse opleve InvalidPidMappingException. Efter dette vil transaktionskoordinatoren udstede et nyt sekventielt nummer til den næste transaktion. Meddelelser kan dog gå tabt, hvis InvalidPidMappingException ikke håndteres korrekt.
I stedet for resultater
Som man kan se, er det ikke nok blot at sende meddelelser til Kafka. Man skal vælge kombinationen af parametre og være parat til at foretage hurtige ændringer. I denne artikel har jeg forsøgt at beskrive opsætningen af exactly once delivery detaljeret og beskrive nogle problemer med konfigurationer af client.id og transactional.id, som vi stødte på. Nedenfor er opsætningerne for Producer og Consumer præsenteret kort.
Producer:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 — for ordnet forsendelse)
- transactional.id = ${application-name}-${hostname}
Consumer:
- isolation.level = read_committed
For at minimere fejl i fremtidige applikationer har vi lavet vores egen indpakning over spring-konfigurationen, hvor værdierne for nogle af de nævnte parametre allerede er angivet.
Her er et par materialer til selvstudium:
Kilde: habr.com
