Како је Кафка постао стварност

Хеј Хабр!

Радим у Тинкофф тиму, који развија сопствени центар за обавештавање. Углавном развијам у Јави користећи Спринг боот и решавам разне техничке проблеме који се јављају у пројекту.

Већина наших микросервиса асинхроно комуницира једни са другима преко посредника за поруке. Раније смо користили ИБМ МК као брокера, који више није могао да се носи са оптерећењем, али је у исто време имао високе гаранције испоруке.

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

Гарантована испорука и још много тога

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

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

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

Ако желите да користите ЈМКС у апликацији са Кафком, онда би ово могао бити проблем. У овом случају, најбоље је користити комбинацију назива апликације и, на пример, назива теме као вредности цлиент.ид. Резултат наше конфигурације се може видети у излазу команде кафка-потрошачке групе из услужних програма из Цонфлуента:

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

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

Из наведених вредности јасно је да ацкс једнак -1 даје најјачу гаранцију да порука неће бити изгубљена.

Као што сви знамо, дистрибуирани системи су непоуздани. Да би се заштитио од пролазних грешака, Кафка Продуцер пружа опцију поново покушава, што вам омогућава да подесите број покушаја поновног слања унутар испоруке.тимеоут.мс. Пошто параметар поновних покушаја има подразумевану вредност Интегер.МАКС_ВАЛУЕ (2147483647), број поновних покушаја поруке може да се подеси променом само испоруке.тимеоут.мс.

Идемо ка тачно једној испоруци

Наведена подешавања омогућавају нашем Произвођачу да испоручује поруке са високом гаранцијом. Хајде сада да разговарамо о томе како осигурати да само једна копија поруке буде написана на Кафкину тему? У најједноставнијем случају, да бисте то урадили, потребно је да подесите параметар на Продуцер омогућити.идемпотенција на истинито. Идемпотенција гарантује да се само једна порука уписује на одређену партицију једне теме. Предуслов за омогућавање идемпотенције су вредности ацкс = све, покушај поново > 0, мак.ин.флигхт.рекуестс.пер.цоннецтион ≤ 5. Ако програмер не одреди ове параметре, горенаведене вредности ће бити аутоматски подешене.

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

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

Такође, коришћење кључа и идемпотентно слање са параметром мак.ин.флигхт.рекуестс.пер.цоннецтион = 1 даје вам поједностављену обраду порука на Потрошачу. Такође је вредно запамтити да ако је контрола приступа конфигурисана на вашем кластеру, онда ће вам требати права да идемпотентно пишете у тему.

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

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

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.

Да бисмо решили овај проблем, имену апликације додајемо суфикс у облику имена хоста, који добијамо из променљивих окружења.

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

Да би се спречило да потрошач унапред прочита такве поруке, потребно је да подеси параметар изолација.ниво на реад_цоммиттед вредност. Такав потрошач ће моћи да чита нетрансакционе поруке као и раније, а трансакцијске поруке тек након урезивања.
Ако сте подесили сва претходно наведена подешавања, онда сте конфигурисали тачно једном испоруку. Честитам!

Али постоји још једна нијанса. Трансацтионал.ид, који смо конфигурисали изнад, је заправо префикс трансакције. У менаџеру трансакција му се додаје редни број. Примљени идентификатор се издаје на трансакција.ид.екпиратион.мс, који је конфигурисан на Кафка кластеру и има подразумевану вредност „7 дана“. Ако за то време апликација није примила ниједну поруку, примићете када покушате следеће трансакцијско слање ИнвалидПидМаппингЕкцептион. Координатор трансакције ће тада издати нови редни број за следећу трансакцију. Међутим, порука може бити изгубљена ако се изузетком ИнвалидПидМаппингЕкцептион не рукује исправно.

Уместо укупних

Као што видите, није довољно само послати поруке Кафки. Морате да изаберете комбинацију параметара и да будете спремни за брзе промене. У овом чланку сам покушао да детаљно прикажем подешавање тачно једном испоруке и описао неколико проблема са конфигурацијама цлиент.ид и трансацтионал.ид на које смо наишли. Испод је сажетак подешавања произвођача и потрошача.

Продуцент:

1. ацкс = све
2. покушаји > 0
3. омогућити.идемпотенција = истина
4. мак.ин.флигхт.рекуестс.пер.цоннецтион ≤ 5 (1 за уредно слање)
5. трансакција.ид = ${име-апликације}-${име хоста}

Потрошач:

1. исолатион.левел = реад_цоммиттед

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

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

• КИП-98 - Испорука тачно једном и размена порука о трансакцијама
• Опис подешавања

Извор: ввв.хабр.цом