Giunsa ang Kafka nahimong kamatuoran

Hoy Habr!

Nagtrabaho ko sa grupo sa Tinkoff, nga nagpalambo sa kaugalingon nga sentro sa pagpahibalo. Kasagaran ako nag-develop sa Java gamit ang Spring boot ug nagsulbad sa lainlaing mga problema sa teknikal nga mitungha sa usa ka proyekto.

Kadaghanan sa among mga microservice nakigsulti sa usag usa nga asynchronous pinaagi sa usa ka broker sa mensahe. Kaniadto, gigamit namo ang IBM MQ isip usa ka broker, nga dili na makasagubang sa load, apan sa samang higayon adunay taas nga garantiya sa pagpadala.

Isip usa ka puli, gitanyagan kami sa Apache Kafka, nga adunay taas nga potensyal sa pag-scale, apan, sa kasubo, nanginahanglan usa ka hapit indibidwal nga pamaagi sa pag-configure alang sa lainlaing mga senaryo. Dugang pa, ang labing menos kausa nga mekanismo sa pagpadala nga nagtrabaho sa Kafka sa default wala magtugot sa pagpadayon sa gikinahanglan nga lebel sa pagkamakanunayon gikan sa kahon. Sunod, akong ipaambit ang among kasinatian sa pag-configure sa Kafka, labi na, isulti ko kanimo kung giunsa ang pag-configure ug pagkinabuhi nga eksakto kausa nga pagpadala.

Garantisado nga pagpadala ug uban pa

Ang mga setting nga gihisgutan sa ubos makatabang sa pagpugong sa daghang mga problema sa default setting sa koneksyon. Apan una gusto nako nga hatagan pagtagad ang usa ka parameter nga makapadali sa usa ka posible nga pag-debug.

Makatabang kini kliyente.id para sa Producer ug Consumer. Sa una nga pagtan-aw, mahimo nimong gamiton ang ngalan sa aplikasyon ingon nga kantidad, ug sa kadaghanan nga mga kaso kini molihok. Bisan kung ang sitwasyon kung ang usa ka aplikasyon naggamit daghang mga Konsyumer ug gihatagan nimo sila sa parehas nga kliyente.id, moresulta sa mosunod nga pasidaan:

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

Kung gusto nimo gamiton ang JMX sa usa ka aplikasyon sa Kafka, nan kini mahimong usa ka problema. Alang niini nga kaso, labing maayo nga mogamit usa ka kombinasyon sa ngalan sa aplikasyon ug, pananglitan, ang ngalan sa hilisgutan ingon ang bili sa client.id. Ang resulta sa among configuration makita sa command output kafka-consumer-groups gikan sa mga utilities gikan sa Confluent:

Karon atong tan-awon ang senaryo alang sa garantiya nga paghatud sa mensahe. Ang Kafka Producer adunay parameter acks, nga nagtugot kanimo sa pag-configure pagkahuman kung pila ang nag-ila nga kinahanglan nga tagdon sa lider sa cluster ang mensahe nga malampuson nga gisulat. Kini nga parameter mahimong makakuha sa mosunod nga mga kantidad:

• 0 - ang pag-ila dili isipon.
• Ang 1 mao ang default nga parameter, 1 ra nga kopya ang gikinahanglan aron ilhon.
• −1 — pag-ila gikan sa tanang gi-synchronize nga mga replika gikinahanglan (cluster setup min.insync.replicas).

Gikan sa nalista nga mga kantidad klaro nga ang mga acks nga katumbas sa −1 naghatag sa labing lig-on nga garantiya nga ang mensahe dili mawala.

Sama sa nahibal-an namong tanan, ang gipang-apod-apod nga mga sistema dili kasaligan. Aron mapanalipdan batok sa lumalabay nga mga sayup, ang Kafka Producer naghatag og kapilian sulayan usab, nga nagtugot kanimo sa pagtakda sa gidaghanon sa mga pagsulay pag-usab sa sulod delivery.timeout.ms. Tungod kay ang parameter sa pag-ulit adunay default nga bili sa Integer.MAX_VALUE (2147483647), ang gidaghanon sa mga pagsulay pag-usab mahimong ma-adjust pinaagi sa pag-ilis lamang sa delivery.timeout.ms.

Mobalhin kami padulong sa eksaktong kausa sa pagpadala

Ang nalista nga mga setting nagtugot sa among Producer sa paghatud sa mga mensahe nga adunay taas nga garantiya. Atong hisgutan karon kung unsaon pagsiguro nga usa ra ka kopya sa usa ka mensahe ang gisulat sa usa ka hilisgutan sa Kafka? Sa pinakasimple nga kaso, aron mahimo kini, kinahanglan nimo nga itakda ang parameter sa Producer enable.idempotence sa tinuod. Ang Idempotency naggarantiya nga usa ra ka mensahe ang gisulat sa usa ka piho nga partisyon sa usa ka hilisgutan. Ang nag-una nga kondisyon alang sa pag-enable sa idempotency mao ang mga kantidad acks = tanan, sulayi pag-usab > 0, max.in.flight.requests.per.connection ≤ 5. Kung kini nga mga parameter wala gitino sa developer, ang mga kantidad sa ibabaw awtomatiko nga itakda.

Kung gi-configure ang idempotency, gikinahanglan aron masiguro nga ang parehas nga mga mensahe mahuman sa parehas nga mga partisyon matag oras. Mahimo kini pinaagi sa pagbutang sa partitioner.class nga yawe ug parameter sa Producer. Magsugod ta sa yawe. Kinahanglan nga parehas kini sa matag pagsumite. Kini dali nga makab-ot pinaagi sa paggamit sa bisan unsang mga ID sa negosyo gikan sa orihinal nga post. Ang partitioner.class nga parametro adunay default value − DefaultPartitioner. Uban niini nga estratehiya sa partitioning, sa kasagaran kita molihok sama niini:

• Kung ang partisyon klaro nga gipiho sa pagpadala sa mensahe, nan among gigamit kini.
• Kung ang partisyon wala gitino, apan ang yawe gipiho, pilia ang partisyon pinaagi sa hash sa yawe.
• Kung ang partisyon ug yawe wala gitino, pilia ang mga partisyon nga tagsa-tagsa (round-robin).

Usab, gamit ang usa ka yawe ug idempotent nga pagpadala nga adunay usa ka parameter max.in.flight.requests.per.connection = 1 naghatag kanimo og streamline nga pagproseso sa mensahe sa Consumer. Angayan usab nga hinumdoman nga kung ang kontrol sa pag-access na-configure sa imong cluster, nan kinahanglan nimo ang mga katungod sa pagsulat sa usa ka hilisgutan.

Kung kalit nga kulang ka sa mga kapabilidad sa idempotent nga pagpadala pinaagi sa yawe o ang lohika sa Producer nga bahin nanginahanglan pagpadayon sa pagkamakanunayon sa datos tali sa lainlaing mga partisyon, nan ang mga transaksyon moabut aron maluwas. Dugang pa, gamit ang usa ka kadena nga transaksyon, mahimo nimong i-synchronize ang usa ka rekord sa Kafka, pananglitan, nga adunay usa ka rekord sa database. Aron mahimo ang transactional nga pagpadala sa Producer, kini kinahanglan nga idempotent ug dugang nga gitakda transactional.id. Kung ang imong Kafka cluster adunay access control nga na-configure, nan ang usa ka transactional record, sama sa usa ka idempotent record, magkinahanglan og mga permiso sa pagsulat, nga mahimong ihatag pinaagi sa maskara gamit ang kantidad nga gitipigan sa transactional.id.

Sa pormal nga paagi, ang bisan unsang string, sama sa ngalan sa aplikasyon, mahimong gamiton isip usa ka transaction identifier. Apan kung maglansad ka daghang mga higayon sa parehas nga aplikasyon nga adunay parehas nga transactional.id, unya ang una nga gilunsad nga pananglitan mahunong sa usa ka sayup, tungod kay ang Kafka mag-isip niini nga proseso sa zombie.

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.

Aron masulbad kini nga problema, magdugang kami ug suffix sa ngalan sa aplikasyon sa porma sa hostname, nga among makuha gikan sa mga variable sa palibot.

Ang prodyuser gi-configure, apan ang mga transaksyon sa Kafka nagkontrol lamang sa sakup sa mensahe. Bisan unsa pa ang kahimtang sa transaksyon, ang mensahe moadto dayon sa hilisgutan, apan adunay dugang nga mga hiyas sa sistema.

Aron mapugngan ang ingon nga mga mensahe nga mabasa sa Consumer sa una, kinahanglan nga itakda ang parameter pag-inusara.level sa read_committed nga bili. Ang maong Consumer makahimo sa pagbasa sa mga non-transaktional nga mga mensahe sama sa una, ug mga transactional nga mga mensahe human lamang sa usa ka commit.
Kung imong gitakda ang tanan nga mga setting nga gilista sa sayo pa, nan imong gi-configure ang eksakto kausa nga pagpadala. Mga pahalipay!

Apan adunay usa pa ka nuance. Ang Transactional.id, nga among gi-configure sa ibabaw, mao gyud ang prefix sa transaksyon. Sa manager sa transaksyon, usa ka sequence number ang idugang niini. Ang nadawat nga identifier gihatag sa transactional.id.expiration.ms, nga gi-configure sa usa ka cluster sa Kafka ug adunay default nga kantidad nga "7 ka adlaw". Kung niining panahona ang aplikasyon wala makadawat bisan unsang mga mensahe, unya kung imong sulayan ang sunod nga pagpadala sa transaksyon makadawat ka InvalidPidMappingException. Ang transaction coordinator unya mag-isyu og bag-ong sequence number para sa sunod nga transaksyon. Bisan pa, ang mensahe mahimong mawala kung ang InvalidPidMappingException dili pagdumala sa husto.

Inay total

Sama sa imong nakita, dili igo nga magpadala lang mga mensahe sa Kafka. Kinahanglan ka nga mopili usa ka kombinasyon sa mga parameter ug pag-andam sa paghimo sa dali nga pagbag-o. Niini nga artikulo, gisulayan nako nga ipakita sa detalye ang eksaktong kausa nga pag-setup sa pagpadala ug gihubit ang daghang mga problema sa mga configuration sa client.id ug transactional.id nga among nasugatan. Sa ubos mao ang usa ka summary sa mga setting sa Producer ug Consumer.

producer:

1. acks = tanan
2. pagsulay pag-usab > 0
3. enable.idempotence = tinuod
4. max.in.flight.requests.per.connection ≤ 5 (1 para sa hapsay nga pagpadala)
5. transactional.id = ${aplikasyon-ngalan}-${hostname}

Konsyumer:

1. isolation.level = read_committed

Aron mamenosan ang mga sayup sa umaabot nga mga aplikasyon, naghimo kami sa among kaugalingon nga wrapper sa pagsulud sa tingpamulak, diin ang mga kantidad alang sa pipila nga nalista nga mga parameter gitakda na.

Ania ang pipila ka mga materyales alang sa kaugalingon nga pagtuon:

• KIP-98 - Eksaktong Kausa nga Paghatud ug Transaksyonal nga Mensahe
• Deskripsyon sa mga setting

Source: www.habr.com