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

Здравейте, Хабр!

Работя в екипа на Tinkoff, който се занимава с разработването на собствен център за нотификации. По-голямата част от времето пиша на Java с помощта на Spring boot и решавам различни технически проблеми, възникващи в проекта.

Повечето от нашите микросервизи взаимодействат помежду си асинхронно чрез брокер на съобщения. Преди това използвахме IBM MQ като брокер, който спря да се справя с натоварването, но имаше високи гаранции за доставка.

Като алтернатива ни предложиха Apache Kafka, която предлага висок потенциал за мащабиране, но, за съжаление, изисква практически индивидуален подход при конфигурирането за различни сценарии. Освен това, механизмът за доставка ‘at least once’, който работи по подразбиране в Kafka, не позволява поддържането на необходимото ниво на консистентност извън кутията. По-долу ще споделя нашия опит с конфигурацията на Kafka, като особено ще разкажа как да настроите и да живеете с ‘exactly once’ доставка.

Гарантирана доставка и не само

Параметрите, за които ще стане въпрос по-долу, ще помогнат да се предотвратят редица проблеми с настройките по подразбиране. Но първо искам да обърна внимание на един параметър, който ще улесни евентуалното дебъгване.

В това ще помогне client.id и за Producer, и за Consumer. На пръв поглед, като стойност може да се използва името на приложението, и в повечето случаи това ще проработи. Въпреки това, ситуацията, в която приложението използва няколко Consumer-а и вие задавате един и същ client.id, води до следното предупреждение:

org.apache.kafka.common.utils.AppInfoParser — Грешка при регистрация на 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, който позволява да се настрои след колко потвърждения на лидера на клъстера, съобщението трябва да се счита за успешно записано. Този параметър може да приема следните стойности:

• 0 — потвържденията няма да се считат.
• 1 — параметър по подразбиране, изисква се потвърждение само от 1 реплика.
• −1 — необходима проверка от всех синхронизированных реплик (настройка кластера min.insync.replicas).

Из перечисленных значений видно, что acks равный −1 дает наилучшие гарантии того, что сообщение не потеряется.

Как мы все знаем, распределенные системы ненадежны. Для защиты от временных неисправностей Kafka Producer предоставляет параметр retries, который позволяет задавать количество попыток повторной отправки в течение delivery.timeout.ms. Параметр retries по умолчанию равен Integer.MAX_VALUE (2147483647), так что количество повторных отправок сообщения можно регулировать, изменяя только delivery.timeout.ms.

Двигаемся к exactly once delivery

Вышеупомянутые настройки позволяют нашему Producer’у доставлять сообщения с высокой гарантией. Теперь давайте обсудим, как гарантировать запись только одной копии сообщения в Kafka-топик? В самом простом случае для этого необходимо установить параметр enable.idempotence в значение true. Идемпотентность гарантирует запись только одного сообщения в конкретную партицию одного топика. Предварительными условиями для включения идемпотентности являются значения acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Если эти параметры не указаны разработчиком, то автоматически будут установлены упомянутые выше значения.

Когда идемпотентность настроена, необходимо добиться того, чтобы одинаковые сообщения всегда попадали в одни и те же партиции. Это можно сделать, настроив ключ и параметр partitioner.class на Producer. Начнем с ключа. Для каждой отправки он должен быть одинаковым. Этого можно легко добиться, используя какой-либо бизнес-идентификатор из оригинального сообщения. Параметр partitioner.class по умолчанию — DefaultPartitioner. При этой стратегии партиционирования по умолчанию действуем следующим образом:

• Если партиция явно указана при отправке сообщения, то используем ее.
• Если партиция не указана, но указан ключ — выбираем партицию по хэшу от ключа.
• Если партиция и ключ не указаны — выбираем партиции по очереди (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 е опитал операция с стара епоха. Има или по-нов producer със същия transactionalId, или транзакцията на producer е била изтекла от брокера.

За да решите този проблем, добавяме към името на приложението суфикс под формата на името на хоста, който получаваме от променливите на средата.

Producer е настроен, но транзакциите в Kafka управляват само обхвата на видимост на съобщението. Независимо от статуса на транзакцията, съобщението моментално попада в топика, но има допълнителни системни атрибути.

За да не бъдат такива съобщения прочетени от Consumer-a преди време, той трябва да зададе параметъра 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.

Producer:

1. acks = all
2. retries > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 — за подредено изпращане)
5. transactional.id = ${application-name}-${hostname}

Consumer:

1. isolation.level = read_committed

За да минимизираме грешките в бъдещите приложения, създадохме собствена обвивка над spring конфигурацията, където вече са зададени стойности за някои от изброените параметри.

А ето и няколко материала за самостоятелно изучаване:

• KIP-98 — Exactly Once Delivery and Transactional Messaging
• Описание на настройките

Източник: habr.com