Bawo ni Kafka di otito

Hey Habr!

Mo ṣiṣẹ lori ẹgbẹ Tinkoff, eyiti o ndagba ile-iṣẹ ifitonileti tirẹ. Mo dagbasoke pupọ julọ ni Java ni lilo bata orisun omi ati yanju ọpọlọpọ awọn iṣoro imọ-ẹrọ ti o dide ni iṣẹ akanṣe kan.

Pupọ julọ awọn iṣẹ microservices wa ni ibasọrọ pẹlu ara wọn ni asynchronously nipasẹ alagbata ifiranṣẹ kan. Ni iṣaaju, a lo IBM MQ bi alagbata, eyiti ko le farada ẹru naa mọ, ṣugbọn ni akoko kanna ni awọn iṣeduro ifijiṣẹ giga.

Gẹgẹbi rirọpo, a funni ni Apache Kafka, eyiti o ni agbara igbelowọn giga, ṣugbọn, laanu, nilo ọna ti o fẹrẹẹkọọkan si iṣeto fun awọn oju iṣẹlẹ oriṣiriṣi. Ni afikun, o kere ju ẹẹkan ẹrọ ifijiṣẹ ti o ṣiṣẹ ni Kafka nipasẹ aiyipada ko gba laaye lati ṣetọju ipele ti aitasera ti a beere lati inu apoti. Nigbamii ti, Emi yoo pin iriri wa ni iṣeto Kafka, ni pataki, Emi yoo sọ fun ọ bi o ṣe le tunto ati gbe pẹlu ifijiṣẹ ni ẹẹkan.

Ifijiṣẹ iṣeduro ati diẹ sii

Awọn eto ti a sọrọ ni isalẹ yoo ṣe iranlọwọ lati yago fun nọmba awọn iṣoro pẹlu awọn eto asopọ aiyipada. Sugbon akọkọ Emi yoo fẹ lati san ifojusi si ọkan paramita ti yoo dẹrọ a ṣee ṣe yokokoro.

Eyi yoo ṣe iranlọwọ klient.id fun Olupese ati Olumulo. Ni wiwo akọkọ, o le lo orukọ ohun elo bi iye, ati ni ọpọlọpọ igba eyi yoo ṣiṣẹ. Botilẹjẹpe ipo naa nigbati ohun elo kan nlo ọpọlọpọ Awọn onibara ati pe o fun wọn ni client.id kanna, awọn abajade ni ikilọ atẹle:

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

Ti o ba fẹ lo JMX ninu ohun elo kan pẹlu Kafka, lẹhinna eyi le jẹ iṣoro kan. Fun ọran yii, o dara julọ lati lo apapọ orukọ ohun elo ati, fun apẹẹrẹ, orukọ koko bi iye client.id. Abajade ti iṣeto wa ni a le rii ninu iṣelọpọ aṣẹ kafka-onibara-awọn ẹgbẹ lati awọn ohun elo lati Confluent:

Bayi jẹ ki a wo oju iṣẹlẹ fun ifijiṣẹ ifiranṣẹ ti o ni iṣeduro. Kafka Producer ni o ni a paramita acks, eyiti o fun ọ laaye lati tunto lẹhin melo ni o jẹwọ oludari iṣupọ nilo lati gbero ifiranṣẹ ti a kọ ni aṣeyọri. Paramita yii le gba awọn iye wọnyi:

• 0 — jẹwọ kii yoo ṣe akiyesi.
• 1 jẹ paramita aiyipada, ẹda 1 nikan ni o nilo lati jẹwọ.
• -1 - jẹwọ lati gbogbo awọn ẹda ti a muṣiṣẹpọ ni a nilo (Ṣeto iṣupọ min.insync.replicas).

Lati awọn iye ti a ṣe akojọ o han gbangba pe awọn acks dogba si -1 funni ni iṣeduro ti o lagbara julọ pe ifiranṣẹ naa kii yoo padanu.

Gẹgẹbi gbogbo wa ṣe mọ, awọn ọna ṣiṣe ti a pin kaakiri ko ni igbẹkẹle. Lati daabobo lodi si awọn aṣiṣe igba diẹ, Olupilẹṣẹ Kafka pese aṣayan atunwo, eyi ti o faye gba o lati ṣeto awọn nọmba ti resend igbiyanju laarin ifijiṣẹ.akoko.ms. Niwọn igba ti paramita atunbere naa ni iye aiyipada ti Integer.MAX_VALUE (2147483647), nọmba awọn igbiyanju ifiranṣẹ le ṣe atunṣe nipasẹ yiyipada ifijiṣẹ nikan.timeout.ms.

A ti wa ni gbigbe si ọna gangan ni kete ti ifijiṣẹ

Eto ti a ṣe akojọ gba Olupese wa laaye lati fi awọn ifiranṣẹ ranṣẹ pẹlu iṣeduro giga. Jẹ ki a sọrọ ni bayi nipa bii o ṣe le rii daju pe ẹda kan ti ifiranṣẹ kan ni a kọ si koko Kafka kan? Ni ọran ti o rọrun julọ, lati ṣe eyi, o nilo lati ṣeto paramita lori Olupese mu ṣiṣẹ.idempotence si otitọ. Idempotency ṣe iṣeduro pe ifiranṣẹ kan ṣoṣo ni a kọ si ipin kan pato ti koko kan. Ipilẹ-iṣaaju fun muu arapotency ṣiṣẹ ni awọn iye acks = gbogbo rẹ, tun gbiyanju> 0, max.in.flight.requests.per.connection ≤ 5. Ti ko ba ṣe alaye awọn aye wọnyi nipasẹ olupilẹṣẹ, awọn iye ti o wa loke yoo ṣeto laifọwọyi.

Nigbati arapotency ti tunto, o jẹ dandan lati rii daju pe awọn ifiranṣẹ kanna pari ni awọn ipin kanna ni gbogbo igba. Eyi le ṣee ṣe nipa tito bọtini partitioner.class ati paramita si Olupese. Jẹ ká bẹrẹ pẹlu awọn bọtini. O gbọdọ jẹ kanna fun ifakalẹ kọọkan. Eyi le ni irọrun ni irọrun nipasẹ lilo eyikeyi awọn ID iṣowo lati ifiweranṣẹ atilẹba. Paramita partitioner.class ni iye aiyipada - DefaultPartitioner. Pẹlu ilana ipinpin yii, nipasẹ aiyipada a ṣe bii eyi:

• Ti ipin naa ba wa ni pato nigbati o nfi ifiranṣẹ ranṣẹ, lẹhinna a lo.
• Ti ipin naa ko ba ni pato, ṣugbọn bọtini ti wa ni pato, yan ipin nipasẹ hash ti bọtini.
• Ti ipin ati bọtini ko ba pato, yan awọn ipin ọkan nipasẹ ọkan (yika-robin).

Paapaa, lilo bọtini ati ifiranšẹ idempotent pẹlu paramita kan max.in.flight.requests.per.connection = 1 yoo fun o streamlined ifiranṣẹ processing lori awọn onibara. O tun tọ lati ranti pe ti iṣakoso iwọle ba tunto lori iṣupọ rẹ, lẹhinna iwọ yoo nilo awọn ẹtọ lati kọ ni ikọsilẹ si koko-ọrọ kan.

Ti o ba jẹ lojiji o ko ni awọn agbara ti ifiranšẹ idempotent nipasẹ bọtini tabi ọgbọn ti o wa ni ẹgbẹ Olupese nilo mimu aitasera data laarin awọn ipin oriṣiriṣi, lẹhinna awọn iṣowo yoo wa si igbala. Ni afikun, lilo idunadura pq kan, o le muuṣiṣẹpọ ni majemu ni ipo Kafka, fun apẹẹrẹ, pẹlu igbasilẹ kan ninu ibi ipamọ data. Lati mu fifiranṣẹ iṣowo ṣiṣẹ si Olupilẹṣẹ, o gbọdọ jẹ agbara ati ṣeto ni afikun transactional.id. Ti iṣupọ Kafka rẹ ba ni tunto iṣakoso iwọle, lẹhinna igbasilẹ iṣowo kan, bii igbasilẹ idempotent, yoo nilo awọn igbanilaaye kikọ, eyiti o le funni nipasẹ iboju-boju nipa lilo iye ti o fipamọ sinu transactional.id.

Ni deede, eyikeyi okun, gẹgẹbi orukọ ohun elo, le ṣee lo bi idamo idunadura kan. Ṣugbọn ti o ba ṣe ifilọlẹ ọpọlọpọ awọn iṣẹlẹ ti ohun elo kanna pẹlu idunadura kanna.id, lẹhinna apẹẹrẹ ifilọlẹ akọkọ yoo duro pẹlu aṣiṣe kan, nitori Kafka yoo ro pe o jẹ ilana Ebora.

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.

Lati yanju iṣoro yii, a ṣafikun suffix kan si orukọ ohun elo ni irisi orukọ agbalejo, eyiti a gba lati awọn oniyipada ayika.

Olupilẹṣẹ ti tunto, ṣugbọn awọn iṣowo lori Kafka nikan ṣakoso iwọn ti ifiranṣẹ naa. Laibikita ipo iṣowo, ifiranṣẹ naa lẹsẹkẹsẹ lọ si koko-ọrọ, ṣugbọn o ni awọn eroja eto afikun.

Lati yago fun iru awọn ifiranṣẹ lati ka nipasẹ Olumulo ṣaaju akoko, o nilo lati ṣeto paramita naa ipinya.ipele to read_committed iye. Iru Olumulo yoo ni anfani lati ka awọn ifiranṣẹ ti kii ṣe iṣowo bi iṣaaju, ati awọn ifiranṣẹ idunadura nikan lẹhin adehun kan.
Ti o ba ti ṣeto gbogbo awọn eto ti a ṣe akojọ tẹlẹ, lẹhinna o ti tunto ni deede ni kete ti ifijiṣẹ. Oriire!

Ṣugbọn nuance kan wa. Transactional.id, eyiti a tunto loke, jẹ gangan asọtẹlẹ idunadura. Lori oluṣakoso idunadura, nọmba ọkọọkan kan ni afikun si. Idanimọ ti o gba ti wa ni ti oniṣowo si transactional.id.expiration.ms, eyiti o tunto lori iṣupọ Kafka ati pe o ni iye aiyipada ti “awọn ọjọ 7”. Ti lakoko yii ohun elo naa ko gba awọn ifiranṣẹ eyikeyi, lẹhinna nigba ti o ba gbiyanju ifiranšẹ iṣowo atẹle iwọ yoo gba InvalidPidMappingException. Alakoso iṣowo yoo lẹhinna fun nọmba ọkọọkan tuntun fun idunadura atẹle. Bibẹẹkọ, ifiranṣẹ naa le padanu ti InvalidPidMappingException ko ba mu ni deede.

Dipo ti lapapọ

Bii o ti le rii, ko to lati firanṣẹ awọn ifiranṣẹ nirọrun si Kafka. O nilo lati yan akojọpọ awọn paramita ati mura lati ṣe awọn ayipada iyara. Ninu nkan yii, Mo gbiyanju lati ṣafihan ni awọn alaye ni pato iṣeto ifijiṣẹ ẹẹkan ati ṣapejuwe awọn iṣoro pupọ pẹlu client.id ati awọn atunto transactional.id ti a pade. Ni isalẹ ni akopọ ti Olupese ati awọn eto Olumulo.

Oludasile:

1. acks = gbogbo
2. igbiyanju> 0
3. agbara.idempotence = otitọ
4. max.in.flight.requests.per.connection ≤ 5 (1 fun fifiranṣẹ lẹsẹsẹ)
5. transactional.id = ${application-name}-${hostname}

Olumulo:

1. isolation.level = read_committed

Lati dinku awọn aṣiṣe ni awọn ohun elo iwaju, a ṣe apẹrẹ tiwa lori iṣeto orisun omi, nibiti awọn iye fun diẹ ninu awọn aye ti a ṣe akojọ ti ṣeto tẹlẹ.

Eyi ni awọn ohun elo meji fun ikẹkọ ara ẹni:

• KIP-98 - Ni kete ti Ifijiṣẹ ati Ifijiṣẹ Iṣowo
• Apejuwe ti awọn eto

orisun: www.habr.com