В данной статье я хочу рассказать об подходе к организации параллельной обработки сообщений из Kafka, когда Kafka Topic используется как транспорт бизнес-событий, например транзакций или ордеров, которые необходимо отправить во внешнюю систему. При это важно обеспечить стабильную скорость обработки трафика и надёжность (отсутствие потерь) в условиях, когда downstream-система не на все запросы отвечает стабильно и быстро.

Покажу почему стандартные подходы, такие как обработка batch'ами, в определённых сценариях перестают работать и приводят к деградации производительности или рисками потери сообщений.

Для иллюстрации подходов далее будет использоваться код из демонстрационного проекта на Kotlin с использованием spring boot, webFlux, spring reactor и reactor-kafka. Код проекта не является production-ready: в нём, например, отсутствует обработка rebalance, а также ряд других моментов, обязательных для промышленной системы. Используется reactor-kafka, однако описываемые решения не зависят от конкретного фреймворка работы с Kafka и могут быть реализованы с использованием других.

В статье намеренно опущены детали реализации бизнес-протокола и механизма идемпотентности. В реальной системе они реализованы с использованием внутреннего хранилища и машины состояний обработки ордеров, но это за рамками данной статьи.

Я активно использую на проекте Kafka Connect Framework и в частности Kafka JDBC Sink Connector для быстрого сохранения данных из Kafka Topic в БД PostgresSQL. Для большинства задач достаточно написать простую JSON-конфигурацию и все стабильно и быстро работает из коробки. Нет необходимости в написании собственного кода. Однако в нетиповых ситуациях расширяемость Kafka Connect тоже помогает - можно переопределить и написать один из компонентов.

В конфигурации JDBC Sink Connector Task существует настройка dialect.name, которая отвечает за выбор диалекта для работы с конкретной БД. Как правило, в 99% случаев используется один из уже реализованных для популярных БД диалектов, как например в моем случае PostgresSqlDatabaseDialect для PostgreSQL.

Может показаться, что вряд ли кому-то понадобится реализовывать свой диалект, если только не имеем дело с какой-то специфичной непопулярной БД. Однако на практике оказалось, что реализация своего диалекта даже для PostgreSQL может быть полезна для решения некоторых возникающих прикладных задач.

В данной статье я хочу показать идеи того, как реализация своего DatabaseDialect может помочь при имплементации нестандартных сценариев для вполне себе популярной БД Postgres, для которой существует PostgresSqlDatabaseDialect.

Для инжекции значений из различных источников в конфигурации тасок Kafka Connect используются различные реализации интерфейса org.apache.kafka.common.config.provider.ConfigProvider. Это специализированный интерфейс, реализации которого отвечают за то, чтобы подставить значения используя различные источники.

В данной статье хочу поделиться примером реального применения Drools для реализации требований бизнеса по гибкому конфигурированию сервиса. Здесь не будет полного обзора или пересказа всех фич Drools. Опишу лишь тот функционал, который оказался полезен и применим конкретно к моей задаче.

Хочу поделиться тем, как мы используем kafka для организации оперативного хранилища справочной информации.

В данной статье приведу реализацию своего kafka http sink connector. Он не претендует на универсальность, но возможно поможет разобраться как разработать свой connector.

Появилась потребность в том, чтобы определенные компоненты сервисов умели подтягивать обновленную конфигурации и работать на основе этой конфигурации, т.е. конфигурация приложения меняется уже после запуска сервиса. Я проведу небольшой обзор подходов, которые нашел применительно к такой задаче и то, на чем остановился.

На проекте используется spring boot 2.6.4 и kotlin 1.5.31. Также для конфигурации сервисов используется spring cloud config server, где в качестве backend используются Git + Vault.

Дано:

Сервис на spring boot(2.6.4) + kotlin(1.5.31) по выгрузке произвольного количества отчётов по крону. Каждый отчет имеет свои настройки. Для конфигурирования списка свойств отчётов используется собственно список. Для инжекта в приложение используется data class с аннотацией @ConfigurationProperties, где одно из свойств - список data class.