Čau Habr!
Pracuji v týmu Tinkoff, který vyvíjí vlastní notifikační centrum. Většinou vyvíjím v Javě pomocí Spring boot a řeším různé technické problémy, které v projektu vyvstanou.
Většina našich mikroslužeb spolu komunikuje asynchronně prostřednictvím zprostředkovatele zpráv. Dříve jsme jako brokera využívali IBM MQ, který již nezvládal zátěž, ale zároveň měl vysoké garance dodání.
Jako náhrada nám byl nabídnut Apache Kafka, který má vysoký potenciál škálování, ale bohužel vyžaduje téměř individuální přístup ke konfiguraci pro různé scénáře. Kromě toho alespoň jednou mechanismus doručení, který ve výchozím nastavení funguje v Kafce, neumožňoval udržovat požadovanou úroveň konzistence hned po vybalení. Dále se podělím o naše zkušenosti s konfigurací Kafka, zejména vám řeknu, jak nakonfigurovat a žít s přesně jednou dodávkou.
Garantované doručení a další
Níže popsaná nastavení pomohou předejít řadě problémů s výchozím nastavením připojení. Nejprve bych se ale chtěl věnovat jednomu parametru, který usnadní případné ladění.
To pomůže client.id pro výrobce a spotřebitele. Na první pohled můžete jako hodnotu použít název aplikace a ve většině případů to bude fungovat. Přestože situace, kdy aplikace používá několik spotřebitelů a vy jim dáte stejné client.id, vede k následujícímu varování:
org.apache.kafka.common.utils.AppInfoParser — Error registering AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Pokud chcete používat JMX v aplikaci s Kafkou, tak to může být problém. Pro tento případ je nejlepší použít jako hodnotu client.id kombinaci názvu aplikace a například názvu tématu. Výsledek naší konfigurace je vidět na výstupu příkazu kafka-spotřebitelské-skupiny od utilit od Confluence:
Nyní se podívejme na scénář pro zaručené doručení zpráv. Kafka Producer má parametr acks, což vám umožňuje nakonfigurovat, po kolika potvrzeních musí vedoucí klastru považovat zprávu za úspěšně zapsanou. Tento parametr může nabývat následujících hodnot:
- 0 – potvrzení nebude zohledněno.
- 1 je výchozí parametr, k potvrzení je potřeba pouze 1 replika.
- −1 — je vyžadováno potvrzení ze všech synchronizovaných replik (nastavení clusteru min.insync.repliky).
Z uvedených hodnot je zřejmé, že acks rovné −1 dává nejsilnější záruku, že zpráva nebude ztracena.
Jak všichni víme, distribuované systémy jsou nespolehlivé. Pro ochranu před přechodnými poruchami nabízí Kafka Producer možnost opakuje, která umožňuje nastavit počet pokusů o opětovné odeslání v rámci delivery.timeout.ms. Protože parametr opakování má výchozí hodnotu Integer.MAX_VALUE (2147483647), lze počet opakování zprávy upravit změnou pouze delivery.timeout.ms.
Směřujeme k přesně XNUMX doručení
Uvedená nastavení umožňují našemu Producentovi doručovat zprávy s vysokou zárukou. Pojďme si nyní říci, jak zajistit, aby ke kafkovskému tématu byla napsána pouze jedna kopie zprávy? V nejjednodušším případě je k tomu potřeba nastavit parametr na Producer umožnit.idempotence pravda. Idempotency zaručuje, že se do konkrétního oddílu jednoho tématu zapíše pouze jedna zpráva. Předpokladem pro umožnění idempotence jsou hodnoty acks = vše, opakovat > 0, max.požadavek.za letu na připojení ≤ 5. Pokud tyto parametry nejsou vývojářem specifikovány, výše uvedené hodnoty se nastaví automaticky.
Když je idempotence nakonfigurována, je nutné zajistit, aby stejné zprávy skončily pokaždé ve stejných oddílech. To lze provést nastavením klíče a parametru partitioner.class na Producer. Začněme klíčem. Musí být stejný pro každé podání. Toho lze snadno dosáhnout použitím jakéhokoli obchodního ID z původního příspěvku. Parametr partitioner.class má výchozí hodnotu − . S touto strategií rozdělení se ve výchozím nastavení chováme takto:
- Pokud je oddíl výslovně uveden při odesílání zprávy, použijeme jej.
- Pokud oddíl není zadán, ale je zadán klíč, vyberte oddíl podle hash klíče.
- Pokud oddíl a klíč nejsou specifikovány, vyberte oddíly jeden po druhém (cyklicky).
Také pomocí klíče a idempotentního odesílání s parametrem max.požadavek.za letu na připojení = 1 vám poskytuje zjednodušené zpracování zpráv na spotřebiteli. Je také třeba si uvědomit, že pokud je na vašem clusteru nakonfigurováno řízení přístupu, budete potřebovat práva k idempotentnímu zápisu do tématu.
Pokud vám náhle chybí možnosti idempotentního odesílání pomocí klíče nebo logika na straně Producer vyžaduje zachování konzistence dat mezi různými oddíly, přijdou na pomoc transakce. Navíc pomocí řetězové transakce můžete podmíněně synchronizovat záznam například v Kafce se záznamem v databázi. Aby bylo možné transakční odesílání Producerovi, musí být idempotentní a dodatečně nastaveno transakční.id. Pokud má váš cluster Kafka nakonfigurované řízení přístupu, pak transakční záznam, jako idempotentní záznam, bude potřebovat oprávnění k zápisu, která lze udělit maskou pomocí hodnoty uložené v transakčním.id.
Formálně lze jako identifikátor transakce použít jakýkoli řetězec, například název aplikace. Pokud ale spustíte několik instancí stejné aplikace se stejným transakčním.id, pak bude první spuštěná instance zastavena s chybou, protože Kafka to bude považovat za zombie proces.
org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.Abychom tento problém vyřešili, přidáme ke jménu aplikace příponu ve tvaru hostname, který získáme z proměnných prostředí.
Producent je nakonfigurován, ale transakce na Kafce řídí pouze rozsah zprávy. Bez ohledu na stav transakce zpráva okamžitě přejde k tématu, ale má další systémové atributy.
Aby spotřebitel takové zprávy nečetl předem, musí nastavit parametr izolace.úroveň na hodnotu read_committed. Takový spotřebitel bude moci číst netransakční zprávy jako dříve a transakční zprávy pouze po potvrzení.
Pokud jste nastavili všechna výše uvedená nastavení, pak jste nakonfigurovali přesně po dodání. Gratulujeme!
Ale je tu ještě jedna nuance. Transakční.id, které jsme nakonfigurovali výše, je ve skutečnosti předpona transakce. Ve správci transakcí se k němu přidá pořadové číslo. Přijatý identifikátor je vydán na transakční.id.vypršení platnosti.ms, který je nakonfigurován na clusteru Kafka a má výchozí hodnotu „7 dní“. Pokud během této doby aplikace neobdrží žádné zprávy, pak při pokusu o další transakční odeslání obdržíte InvalidPidMappingException. Koordinátor transakce pak vydá nové pořadové číslo pro další transakci. Zpráva však může být ztracena, pokud není správně zpracována výjimka InvalidPidMappingException.
Namísto součtů
Jak vidíte, Kafkovi nestačí jednoduše posílat zprávy. Je třeba zvolit kombinaci parametrů a připravit se na rychlé změny. V tomto článku jsem se pokusil podrobně ukázat nastavení doručení přesně jednou a popsal jsem několik problémů s konfiguracemi client.id a transakční.id, na které jsme narazili. Níže je uveden souhrn nastavení výrobce a spotřebitele.
Výrobce:
- acks = vše
- opakování > 0
- umožnit.idempotence = pravda
- max.požadavek.za letu na připojení ≤ 5 (1 pro řádné odeslání)
- transakční.id = ${application-name}-${hostname}
Spotřebitel:
- isolation.level = read_committed
Abychom minimalizovali chyby v budoucích aplikacích, vytvořili jsme vlastní wrapper nad konfigurací pružiny, kde jsou hodnoty pro některé z uvedených parametrů již nastaveny.
Zde je několik materiálů pro samostudium:
Zdroj: www.habr.com