Како Кафка стана реалност

Еј Хабр!

Работам во тимот на Тинкоф, кој развива сопствен центар за известувања. Јас најчесто се развивам во Java користејќи Spring boot и решавам разни технички проблеми што се појавуваат во еден проект.

Повеќето од нашите микроуслуги комуницираат едни со други асинхроно преку брокер за пораки. Претходно, како брокер го користевме IBM MQ, кој повеќе не можеше да се носи со товарот, но истовремено имаше високи гаранции за испорака.

Како замена, ни беше понуден Apache Kafka, кој има висок потенцијал за скалирање, но, за жал, бара речиси индивидуален пристап кон конфигурацијата за различни сценарија. Дополнително, механизмот за испорака барем еднаш што стандардно работи во Кафка не дозволуваше одржување на потребното ниво на конзистентност надвор од кутијата. Следно, ќе го споделам нашето искуство во конфигурацијата на Кафка, особено, ќе ви кажам како да конфигурирате и живеете со точно еднаш испорака.

Гарантирана испорака и многу повеќе

Поставките дискутирани подолу ќе помогнат да се спречат голем број проблеми со стандардните поставки за поврзување. Но, прво би сакал да обрнам внимание на еден параметар што ќе го олесни можното дебагирање.

Ова ќе помогне клиент.ид за производител и потрошувач. На прв поглед, можете да го користите името на апликацијата како вредност, и во повеќето случаи тоа ќе функционира. Иако ситуацијата кога апликацијата користи неколку потрошувачи, а вие им го давате истиот client.id, резултира со следното предупредување:

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

Ако сакате да користите JMX во апликација со Кафка, тогаш ова може да биде проблем. За овој случај, најдобро е да се користи комбинација од името на апликацијата и, на пример, името на темата како вредност client.id. Резултатот од нашата конфигурација може да се види во излезот на командата кафка-потрошувачки-групи од комуналните услуги од Confluent:

Сега да го погледнеме сценариото за загарантирана испорака на пораки. Kafka Producer има параметар курови, што ви овозможува да конфигурирате по колку признавања на водачот на кластерот му требаат за да ја разгледа пораката успешно напишана. Овој параметар може да ги земе следните вредности:

• 0 — признава нема да се смета.
• 1 е стандардниот параметар, потребна е само 1 реплика за да се потврди.
• −1 — Потребна е потврда од сите синхронизирани реплики (поставување на кластерот min.insync.реплики).

Од наведените вредности е јасно дека актите еднакви на -1 даваат најсилна гаранција дека пораката нема да се изгуби.

Како што сите знаеме, дистрибуираните системи се несигурни. За заштита од минливи дефекти, Kafka Producer ја обезбедува опцијата се обидува повторно, што ви овозможува да го поставите бројот на обиди за повторно испраќање во рамките испорака.време.мс. Бидејќи параметарот повторувања има стандардна вредност од Integer.MAX_VALUE (2147483647), бројот на повторувања на пораките може да се прилагоди со менување само delivery.timeout.ms.

Се движиме кон точно еднократна испорака

Наведените поставки му овозможуваат на нашиот производител да испорачува пораки со висока гаранција. Ајде сега да разговараме за тоа како да се осигураме дека само една копија од пораката е напишана на тема на Кафка? Во наједноставен случај, за да го направите ова, треба да го поставите параметарот на Производител овозможи.идемпотенција до вистина. Идемпотенција гарантира дека само една порака е напишана на одредена партиција на една тема. Предуслов за овозможување на идемпотенција се вредностите acks = сите, обидете се повторно > 0, max.in.flight.requests.per.connection ≤ 5. Ако овие параметри не се наведени од развивачот, горенаведените вредности ќе се постават автоматски.

Кога е конфигурирана идемпотенција, неопходно е да се осигураме дека истите пораки завршуваат во истите партиции секој пат. Ова може да се направи со поставување на клучот partitioner.class и параметарот на Producer. Да почнеме со клучот. Мора да биде исто за секој поднесок. Ова лесно може да се постигне со користење на кој било од деловните идентификатори од оригиналната објава. Параметарот partitioner.class има стандардна вредност − Стандарден партиционер. Со оваа стратегија за партиционирање, стандардно постапуваме вака:

• Ако партицијата е експлицитно наведена при испраќање на пораката, тогаш ја користиме.
• Ако партицијата не е одредена, но клучот е наведен, изберете ја партицијата според хашот на клучот.
• Ако партицијата и клучот не се наведени, изберете ги партициите една по една (round-robin).

Исто така, користење на клуч и идемпотентно испраќање со параметар max.in.flight.requests.per.connection = 1 ви дава рационализирана обработка на пораките на потрошувачот. Исто така, вреди да се запамети дека ако контролата на пристап е конфигурирана на вашиот кластер, тогаш ќе ви требаат права за идемотентно пишување на тема.

Ако одеднаш ви недостигаат можности за идемотентно испраќање со клуч или логиката на страната на производителот бара одржување на конзистентноста на податоците помеѓу различните партиции, тогаш трансакциите ќе дојдат на помош. Дополнително, користејќи верижна трансакција, можете условно да синхронизирате запис во Кафка, на пример, со запис во базата на податоци. За да се овозможи трансакциско испраќање до Производителот, тој мора да биде немоќен и дополнително поставен трансакциска.ид. Ако вашиот кластер Кафка има конфигурирана контрола на пристап, тогаш трансакцискиот запис, како идемпотентниот запис, ќе има потреба од дозволи за пишување, кои може да се доделат со маска користејќи ја вредноста зачувана во transaksional.id.

Формално, секоја низа, како што е името на апликацијата, може да се користи како идентификатор на трансакцијата. Но, ако стартувате неколку примероци од иста апликација со истиот трансакциски.ид, тогаш првиот стартуван пример ќе биде запрен со грешка, бидејќи Кафка ќе го смета за процес на зомби.

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.

За да го решиме овој проблем, додаваме суфикс на името на апликацијата во форма на името на домаќинот, што го добиваме од променливите на околината.

Производителот е конфигуриран, но трансакциите на Кафка го контролираат само опсегот на пораката. Без оглед на статусот на трансакцијата, пораката веднаш оди на темата, но има дополнителни системски атрибути.

За да спречи таквите пораки да бидат прочитани од страна на потрошувачот пред време, тој треба да го постави параметарот изолација.ниво до read_committed вредност. Таквиот потрошувач ќе може да чита нетрансакциски пораки како досега, а трансакциските пораки само по извршено извршување.
Ако сте ги поставиле сите поставки наведени претходно, тогаш сте конфигурирале точно еднаш испорака. Секоја чест!

Но, има уште една нијанса. Transaction.id, кој го конфигуриравме погоре, е всушност префиксот на трансакцијата. На менаџерот на трансакции, на него се додава секвенционен број. Примениот идентификатор се издава на трансакциска.ид.истек.ms, кој е конфигуриран на кластерот Кафка и има стандардна вредност од „7 дена“. Доколку во ова време апликацијата не примила никакви пораки, тогаш кога ќе се обидете со следното трансакциско испраќање ќе добиете InvalidPidMappingException. Координаторот на трансакцијата потоа ќе издаде нов секвенционен број за следната трансакција. Сепак, пораката може да се изгуби ако не се постапува правилно со InvalidPidMappingException.

Наместо тотали

Како што можете да видите, не е доволно само да му испраќате пораки на Кафка. Треба да изберете комбинација од параметри и да бидете подготвени да направите брзи промени. Во оваа статија, се обидов детално да го прикажам поставувањето точно еднаш за испорака и опишав неколку проблеми со конфигурациите на client.id и трансакциска.id на кои наидовме. Подолу е резиме на поставките за производител и потрошувач.

Продуцент:

1. акс = сите
2. се повторува > 0
3. овозможи.идемпотенција = точно
4. max.in.flight. requests.per.connection ≤ 5 (1 за уредно испраќање)
5. transactional.id = ${application-name}-${hostname}

Потрошувач:

1. изолација.ниво = read_committed

За да ги минимизираме грешките во идните апликации, направивме сопствен обвивка преку конфигурацијата на пружината, каде што вредностите за некои од наведените параметри се веќе поставени.

Еве неколку материјали за самостојно учење:

• KIP-98 - Точно еднаш испорака и размена на пораки
• Опис на поставките

Извор: www.habr.com