Прывітанне, Хабр!
Я працую ў камандзе Tinkoff, якая займаецца распрацоўкай уласнага цэнтру натыфікацый. Па большай частцы я распрацоўваю на Java з выкарыстаннем Spring boot і вырашаю розныя тэхнічныя праблемы, якія ўзнікаюць у праекце.
Большасць нашых мікрасэрвісаў асінхронна ўзаемадзейнічаюць адзін з адным праз брокер паведамленняў. Раней у якасці брокера мы выкарыстоўвалі IBM MQ, які перастаў спраўляцца з нагрузкай, але пры гэтым валодаў высокімі гарантыямі дастаўкі.
У якасці замены нам прапанавалі Apache Kafka, якая валодае высокім патэнцыялам маштабавання, але, нажаль, патрабуе практычна індывідуальнага падыходу да канфігуравання для розных сцэнараў. Акрамя таго, механізм at least once delivery, які працуе ў Kafka па змаўчанні, не дазваляў падтрымліваць неабходны ўзровень кансістэнтнасці са скрынкі. Далей я падзялюся нашым досведам канфігурацыі Kafka, у прыватнасці распавяду, як наладзіць і жыць з exactly once delivery.
Гарантаваная дастаўка і не толькі
Параметры, аб якіх пайдзе прамову далей, дапамогуць прадухіліць шэраг праблем з наладамі падлучэння па змаўчанні. Але спачатку жадаецца надаць увагу аднаму параметру, які палегчыць магчымы дэбаг.
У гэтым дапаможа client.id для Producer і Consumer. На першы погляд, у якасці значэння можна выкарыстоўваць імя дадатку, і ў большасці выпадкаў гэта будзе працаваць. Хоць сітуацыя, калі ў дадатку выкарыстоўваецца некалькі Consumer'ов і вы задаяце ім аднолькавы 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 у дадатку з Kafka, то гэта можа быць праблемай. Для гэтага выпадку лепш за ўсё выкарыстоўваць у якасці значэння client.id камбінацыю з імя дадатку і, напрыклад, імя топіка. Вынік нашай канфігурацыі можна паглядзець у вывадзе каманды kafka-consumer-groups з утыліт ад Confluent:
Цяпер разбяром сцэнар гарантаванай дастаўкі паведамлення. У Kafka Producer ёсць параметр acks, Які дазваляе наладжваць, пасля колькіх acknowledge лідэру кластара неабходна лічыць паведамленне паспяхова запісаным. Гэты параметр можа прымаць наступныя значэнні:
- 0 - acknowledge не будуць лічыцца.
- 1 - параметр па змаўчанні, неабходны acknowledge толькі ад 1 рэплікі.
- −1 — неабходны acknowledge ад усіх сінхранізаваных рэплік (настройка кластара min.insync.replicas).
З пералічаных значэнняў відаць, што acks роўны −1 дае найболей моцныя гарантыі, што паведамленне не згубіцца.
Як мы ўсе ведаем, размеркаваныя сістэмы ненадзейныя. Каб абараніцца ад часовых няспраўнасцяў, Kafka Producer падае параметр паўторныя спробы, які дазваляе задаваць колькасць спроб паўторнай адпраўкі на працягу delivery.timeout.ms. Паколькі параметр retries мае значэнне па змаўчанні Integer.MAX_VALUE (2147483647), колькасць паўторных адпраўак паведамлення можна рэгуляваць, змяняючы толькі delivery.timeout.ms.
Рухаемся да exactly once delivery
Пералічаныя налады дазваляюць нашаму Producer'у дастаўляць паведамленні з высокай гарантыяй. Давайце зараз пагаворым аб тым, як гарантаваць запіс толькі адной копіі паведамлення ў Kafka-топік? У самым простым выпадку для гэтага на Producer трэба ўсталяваць параметр enable.idempotence у значэнне true. Ідэмпатэнтнасць гарантуе запіс толькі аднаго паведамлення ў пэўную партыцыю аднаго топіка. Папярэдняй умовай для ўключэння ідэмпатэнтнасці з'яўляюцца значэнні acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Калі гэтыя параметры не зададзеныя распрацоўшчыкам, то аўтаматычна будуць выстаўлены паказаныя вышэй значэнні.
Калі ідэмпатэнтнасць настроена, неабходна дабіцца таго, каб аднолькавыя паведамленні траплялі кожны раз у адны і тыя ж партыцыі. Гэта можна зрабіць, наладжваючы ключ і параметр partitioner.class на Producer. Давайце пачнем з ключа. Для кожнай адпраўкі ён павінен быць аднолькавым. Гэтага лёгка дамагчыся, выкарыстоўваючы які-небудзь бізнэс-ідэнтыфікатар з арыгінальнага паведамлення. Параметр partitioner.class мае значэнне па змаўчанні . Пры гэтай стратэгіі партыцыянавання па змаўчанні дзейнічаем так:
- Калі партыцыя відавочна паказана пры адпраўцы паведамлення, то выкарыстоўваем яе.
- Калі партыцыя не пазначана, але пазначаны ключ - выбіраем партыцыю па хэшу ад ключа.
- Калі партыцыя і ключ не пазначаны - выбіраем партіціі па чарзе (round-robin).
Акрамя таго, выкарыстанне ключа і ідэмпатэнтнай адпраўкі з параметрам max.in.flight.requests.per.connection = 1 дае вам спарадкаваную апрацоўку паведамленняў на Consumer. Асобна варта памятаць, што, калі на вашым кластары наладжана кіраванне доступам, то вам спатрэбяцца правы на ідэмпатэнтны запіс у топік.
Калі раптам вам бракуе магчымасцяў ідэмпатэнтнай адпраўкі па ключы ці логіка на боку Producer патрабуе захаванні кансістэнтнасці дадзеных паміж рознымі партыямі, то на дапамогу прыйдуць транзакцыі. Акрамя таго, з дапамогай ланцужной транзакцыі можна ўмоўна сінхранізаваць запіс у Kafka, напрыклад, з запісам у БД. Для ўключэння транзакцыйнай адпраўкі на Producer неабходна, каб ён валодаў ідэмпатэнтнасцю, і дадаткова задаць transactional.id. Калі на вашым Kafka-кластары наладжана кіраванне доступам, то для транзакцыйнага запісу, як і для ідэмпатэнтнага, спатрэбяцца правы на запіс, якія могуць быць прадстаўлены па масцы з выкарыстаннем значэння, які захоўваецца ў transactional.id.
Фармальна ў якасці ідэнтыфікатара транзакцыі можна выкарыстоўваць любы радок, напрыклад імя прыкладання. Але калі вы запускаеце некалькі інстансаў аднаго прыкладання з аднолькавым transactional.id, то першы запушчаны інстанс будзе спынены з памылкай, бо Kafka будзе лічыць яго зомбі-працэсам.
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.Каб вырашыць гэтую праблему, мы дадаем да імя прыкладання суфікс у выглядзе імя хаста, які атрымліваем з зменных асяроддзі.
Producer настроены, але транзакцыі на Kafka кіруюць толькі вобласцю бачнасці паведамлення. Незалежна ад статуту транзакцыі, паведамленне адразу пападае ў топік, але валодае дадатковымі сістэмнымі атрыбутамі.
Каб такія паведамленні не счытваліся Consumer'ам раней чакай, яму неабходна ўсталяваць параметр isolation.level у значэнне read_committed. Такі Consumer зможа чытаць нетранзакцыйныя паведамленні як і раней, а транзакцыйныя - толькі пасля коміта.
Калі вы ўсталявалі ўсе пералічаныя раней наладкі, тыя вы наладзілі exactly once delivery. Віншую!
Але ёсць яшчэ адзін нюанс. Transactional.id, які мы настройвалі вышэй, на самой справе з'яўляецца прэфіксам транзакцыі. На менеджэры транзакцый да яго дапісваецца парадкавы нумар. Атрыманы ідэнтыфікатар выдаецца на transactional.id.expiration.ms, які канфігуруецца на Kafka кластары і валодае значэннем па змаўчанні "7 дзён". Калі за гэты час прыкладанне не атрымлівала ніякіх паведамленняў, то пры спробе наступнай транзакцыйнай адпраўкі вы атрымаеце InvalidPidMappingException. Пасля гэтага каардынатар транзакцый выдасць новы парадкавы нумар для наступнай транзакцыі. Пры гэтым паведамленне можа быць страчана, калі InvalidPidMappingException не будзе правільна апрацаваны.
замест вынікаў
Як можна заўважыць, нядосыць проста адпраўляць паведамленні ў Kafka. Трэба выбіраць камбінацыю параметраў і быць гатовым да занясення хуткіх змен. У гэтым артыкуле я пастараўся ў дэталях паказаць наладу exactly once delivery і апісаў некалькі праблем канфігурацый client.id і transactional.id, з якімі мы сутыкнуліся. Ніжэй у кароткай форме прыведзены наладкі Producer і Consumer.
Вытворца:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 - для спарадкаванай адпраўкі)
- transactional.id = ${application-name}-${hostname}
Спажывец:
- isolation.level = read_committed
Каб мінімізаваць памылкі ў будучых дадатках, мы зрабілі сваю абгортку над spring-канфігурацыяй, дзе ўжо зададзены значэння для некаторых з пералічаных параметраў.
А вось пары матэрыялаў для самастойнага вывучэння:
Крыніца: habr.com