Ahoj, Habr!
Pracuji v týmu Tinkoff, který se zabývá vývojem vlastního centra notifikací. Většinou vyvíjím na Java pomocí Spring Boot a řeším různé technické problémy, které v projektu vznikají.
Většina našich mikroservisů vzájemně asynchronně komunikují prostřednictvím zprávového brokera. Dříve jsme jako broker používali IBM MQ, který přestal zvládat zátěž, ale měl vysoké záruky dodávky.
Jako náhradu nám nabídli Apache Kafka, která má vysoký potenciál škálování, ale bohužel vyžaduje téměř individuální přístup k konfiguraci pro různé scénáře. Kromě toho mechanismus at least once delivery, který v Kafce funguje jako výchozí, než umožňoval udržovat potřebnou úroveň konzistence z krabice. Dále se podělím o naše zkušenosti s konfigurací Kafky, konkrétně o tom, jak nastavit a žít s exactly once delivery.
Zaručená dodávka a nejen to
Parametry, o kterých se bude dále hovořit, pomohou předejít řadě problémů s výchozími nastaveními připojení. Ale nejprve bych chtěl věnovat pozornost jednomu parametru, který usnadní případný debugging.
Tomu pomůže client.id pro Producer a Consumer. Na první pohled lze jako hodnotu použít jméno aplikace, a ve většině případů to bude fungovat. I když situace, kdy aplikace používá více Consumerů a vy nastavíte stejný client.id, vede k následujícímu upozornění:
org.apache.kafka.common.utils.AppInfoParser — Chyba při registraci AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Pokud chcete používat JMX v aplikaci s Kafkou, může to být problém. Pro tuto situaci je nejlepší použít jako hodnotu client.id kombinaci jména aplikace a například názvu topiku. Výsledek naší konfigurace si můžete prohlédnout ve výstupu příkazu kafka-consumer-groups z nástrojů od Confluent:
Nyní se podívejme na scénář zaručeného doručení zprávy. Kafkový Producer má parametr acks, který umožňuje nastavit, po kolika potvrzeních musí vůdce clusteru považovat zprávu za úspěšně zapsanou. Tento parametr může mít následující hodnoty:
- 0 — potvrzení se nebudou počítat.
- 1 — výchozí parametr, vyžaduje potvrzení pouze od 1 repliky.
- −1 — je nutné potvrzení od všech synchronizovaných replik (nastavení clustera min.insync.replicas).
Z uvedených hodnot je vidět, že acks rovný −1 poskytuje nejvyšší záruky, že zpráva nebude ztracena.
Jak všichni víme, distribuované systémy jsou nespolehlivé. Abychom se chránili před dočasnými poruchami, poskytuje Kafka Producer parametr retries, který umožňuje nastavit počet pokusů o opětovné odeslání během delivery.timeout.ms. Protože má parametr retries ve výchozím nastavení hodnotu Integer.MAX_VALUE (2147483647), lze počet opětovných odeslání zprávy regulovat změnou pouze delivery.timeout.ms.
Přecházíme k exactly once delivery
Uvedená nastavení umožňují našemu Produceru doručovat zprávy s vysokou zárukou. Nyní si promluvme o tom, jak zajistit záznam pouze jedné kopie zprávy v Kafka-topiku? V nejjednodušším případě je třeba na Produceru nastavit parametr enable.idempotence na hodnotu true. Idempotence zaručuje záznam pouze jedné zprávy do konkrétní partition jednoho topiku. Předpokladem pro zapnutí idempotence jsou hodnoty acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Pokud tyto parametry nejsou nastaveny vývojářem, automaticky budou nastaveny uvedené výše.
Když je idempotence nastavena, je nutné zajistit, aby stejné zprávy přicházely vždy do stejných partitions. Toho lze dosáhnout nastavením klíče a parametru partitioner.class na Produceru. Pojďme začít s klíčem. Pro každé odeslání musí být stejný. Toho lze snadno dosáhnout použitím jakéhokoli obchodního identifikátoru z původní zprávy. Parametr partitioner.class má ve výchozím nastavení hodnotu — . Při této standardní strategii partitioningu postupujeme takto:
- Pokud je partition výslovně uvedena při odesílání zprávy, použijeme ji.
- Pokud partition není uvedena, ale je uveden klíč — vybereme partition podle hash klíče.
- Pokud partition a klíč nejsou uvedeny — vybereme partitiony podle pořadí (round-robin).
Kromě toho použití klíče a idempotentního odesílání s parametrem max.in.flight.requests.per.connection = 1 poskytuje uspořádané zpracování zpráv na Consumer. Je také důležité si pamatovat, že pokud je v vašem clusteru nastavena správa přístupu, budete potřebovat oprávnění k idempotentnímu zápisu do tématu.
Pokud vám náhodou chybí možnosti idempotentního odesílání podle klíče nebo logika na straně Producer vyžaduje zachování konzistence dat mezi různými partition, přicházejí na pomoc transakce. Pomocí řetězové transakce je také možné podmíněně synchronizovat zápis v Kafka, například se zápisem v DB. Pro povolení transakčního odesílání na Produceru je nutné, aby byl idempotentní, a navíc nastavit transactional.id. Pokud je ve vašem Kafka clusteru nastavena správa přístupu, pak pro transakční zápis, stejně jako pro idempotentní, budete potřebovat oprávnění k zápisu, které mohou být poskytnuta pomocí masky s hodnotou uloženou v transactional.id.
Formálně může být jako identifikátor transakce použita jakákoli posloupnost znaků, například název aplikace. Ale pokud spouštíte několik instancí jedné aplikace se stejným transactional.id, první spuštěná instance bude zastavena s chybou, protože Kafka ji bude považovat za zombie proces.
org.apache.kafka.common.errors.ProducerFencedException: Producer se pokusil provést operaci se starou epochou. Buď existuje novější producer se stejným transactionalId, nebo byla transakce producery vypršena brokerem.Abychom tento problém vyřešili, přidáme k názvu aplikace příponu v podobě názvu hostitele, který získáme z proměnných prostředí.
Producer je nastaven, ale transakce v Kafka řídí pouze oblast viditelnosti zpráv. Bez ohledu na stav transakce zpráva okamžitě vstupuje do tématu, ale má další systémové atributy.
Aby tyto zprávy nebyly Consumerem načteny předčasně, musí nastavit parametr isolation.level na hodnotu read_committed. Takový Consumer bude moct číst netransakční zprávy jako dřív, zatímco transakční — pouze po potvrzení.
Pokud jste nastavili všechny výše uvedené nastavení, nastavili jste exactly once delivery. Gratuluji!
Ale existuje ještě jedna nuanční. Transactional.id, který jsme nastavovali výše, je ve skutečnosti prefixem transakce. K němu je na správci transakcí přidán pořadový číslo. Získaný identifikátor je vydán na transactional.id.expiration.ms, který je konfigurován na Kafka clusteru a má výchozí hodnotu „7 dní“. Pokud aplikace během této doby nepřijme žádné zprávy, při pokusu o další transakční odeslání obdržíte InvalidPidMappingException. Po tomto kroku koordinator transakcí vygeneruje nové pořadové číslo pro další transakci. Přitom může být zpráva ztracena, pokud nebude InvalidPidMappingException správně zpracován.
Namísto závěrů
Jak už bylo zmíněno, nestačí jen posílat zprávy do Kafka. Je potřeba vybírat kombinaci parametrů a být připraven na rychlé úpravy. V tomto článku jsem se snažil podrobně ukázat konfiguraci exactly once delivery a popsal několik problémů s konfiguracemi client.id a transactional.id, se kterými jsme se setkali. Níže jsou stručně uvedeny nastavení Producer a Consumer.
Producer:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 — pro seřazené odesílání)
- transactional.id = ${application-name}-${hostname}
Consumer:
- isolation.level = read_committed
Abychom minimalizovali chyby v budoucích aplikacích, vytvořili jsme naši vlastní obálku nad spring-konfigurací, kde jsou již některé z uvedených parametrů nastaveny.
A zde je pár materiálů pro samostudium:
Zdroj: habr.com
