В большинстве систем транзакции Kafka не нужны. Чаще всего проще и надёжнее использовать тех же самых идемпотентных консьюмеров. В статье как раз разобраны сценарии, где транзакции действительно оправданы (если очень уж нужна атомарность), и почему во многих случаях от них больше вреда, чем пользы.
При использовании StatefulSet в Kubernetes (или других решений со стабильными идентификаторами) каждый pod имеет постоянное имя, которое сохраняется между перезапусками (например, my-service-0, my-service-1 и т.д.). В таком сценарии можно использовать transactional.id в виде префикса, основанного на имени pod’а (например, ${POD_NAME}), что позволяет избежать как накопления незавершённых транзакций, так и проблем с fencing.
В статье я сфокусировался на базовой настройке Spring Boot и не углублялся в детали deployment'а — всё-таки добавлю это уточнение в статью.
Имеете право не согласиться. Выбор зависит от контекста. С позиции SOLID вы полностью правы. Когда я писал "костыльно", я имел в виду "решение, которое создаёт больше сложности, чем решает проблему" для нашего учебного проекта. А "чище" относилось к тому, что мы пишем меньше кода и создаём меньше сущностей (что оптимально для небольших проектов). Для крупных проектов 2 бина предпочтительнее из-за лучшей тестируемости и разделения ответственности между командами. Обновлю статью с этими уточнениями. Спасибо за фидбек!
Действительно, с августа 2025 Bitnami переместили все образы в bitnami legacy, где они больше не получают обновления. Статья будет обновлена. Спасибо за фидбек!
Если вы добавляете консьюмеров без групп на топик, то партиции не распределяются. Каждый консьюмер читает все сообщения. Если же с группами, то в рамках одной группы будет распределение партиций.
Теперь давайте рассмотрим ваш кейс с 3 партициями:
Если 1 консьюмер, то он читает все партиции.
Если 2 консьюмера, то один читает две партиции, другой — одну.
Если 3 консьюмера, то каждый читает по одной.
Ещё предлагаю для большего понимания рассмотреть случай, когда консьюмеров в группе больше, чем партиций. В этом случае "лишние" консьюмеры будут простаивать.
Если хотите, чтобы разными сервисами обрабатывались одни и те же сообщения, надо все инстансы в рамках одного сервиса засунуть в одну группу. Так надо сделать для каждого интересующего вас сервиса.
Вторая часть вопроса несколько более сложная. Попытаюсь адекватно объяснить:
Оффсеты напрямую не сопоставляются с консьюмерами. Видите ли, какая ситуация, в топике, ответственном за оффсеты, хранятся сообщения следующего вида:
Ключ — <group_id, topic, partition>
Значение — <commited_offset, metadata>
Это нам даёт возможности для перебалансировки, так как нет жётской связи между оффсетом и конкретным инстансом.
В вашем кейсе с упавшим единственным консьюмером будет примерно следующее:
Kafka переведёт группу в состояние перебалансировки.
Контроллер кластера смаппит сообщение из offser-топика на текущий активный консьюмер.
Как следствие, активный консьюмер начнёт читать с последнего закоммиченного оффсета. То есть никаких потерь не будет, так как мёртвый брокер не мог закоммитить оффсеты для новых сообщений. На всякий случай отмечу, что коммит в Kafka — это некий процесс сохранения оффсета консьюмером.
И в этом случае настройка latest применена не будет, так как кафка смаппит оффсет на консьюмера, а эта настройка срабатывает, как вы, надеюсь, помните, когда Kafka не находит оффсет для консьюмера.
Все будет хорошо. Консьюмеры в одной группе не знают о консьюмерах в другой. Чтение будет происходить независимо.
На конкретное сообщение, как мне известно, настроить нельзя. Можно только настроить на топик. О том, как конкретно настроить, используя Spring Boot, я рассказал в своей свежей практической статье. Рекомендую к прочтению :)
Мы создаём новую группу не для того, чтобы разгрузить первую. Мы это делаем для того, чтобы эта самая новая группа имела доступ к тем же самым сообщениям, но выполняла другую логику. То есть для независимого чтения одних и тех же данных. Для разгрузки мы увеличиваем количество консьюмеров в рамках одной группы. Тогда уже между ними распределяются партиции и каждый читает свою часть данных.
Согласен с вами. Действительно, слегка запнулся в начале. Поправил статью. Теперь в начале задаю правильный контекст (называю Kafka журналом). Но дальше провожу аналогии с сообщениями, чтобы безболезненно ввести читателя в курс дела.
В большинстве систем транзакции Kafka не нужны. Чаще всего проще и надёжнее использовать тех же самых идемпотентных консьюмеров. В статье как раз разобраны сценарии, где транзакции действительно оправданы (если очень уж нужна атомарность), и почему во многих случаях от них больше вреда, чем пользы.
Спасибо за комментарий!
На самом деле да, это правда.
При использовании
StatefulSetв Kubernetes (или других решений со стабильными идентификаторами) каждый pod имеет постоянное имя, которое сохраняется между перезапусками (например,my-service-0,my-service-1и т.д.). В таком сценарии можно использоватьtransactional.idв виде префикса, основанного на имени pod’а (например,${POD_NAME}), что позволяет избежать как накопления незавершённых транзакций, так и проблем с fencing.В статье я сфокусировался на базовой настройке Spring Boot и не углублялся в детали deployment'а — всё-таки добавлю это уточнение в статью.
Вам ещё раз спасибо!
Да, лучше прописать это явно
Всегда пожалуйста!
Эту тему как раз буду разбирать в следующей статье.
Рад, что вам нравится!
Приветствую! Рад, что вам нравится!
Avro планировал в скором времени рассмотреть. Кастомную имплементацию думал опустить, но раз уж интересно, то обязательно посмотрим.
Имеете право не согласиться. Выбор зависит от контекста. С позиции SOLID вы полностью правы. Когда я писал "костыльно", я имел в виду "решение, которое создаёт больше сложности, чем решает проблему" для нашего учебного проекта. А "чище" относилось к тому, что мы пишем меньше кода и создаём меньше сущностей (что оптимально для небольших проектов). Для крупных проектов 2 бина предпочтительнее из-за лучшей тестируемости и разделения ответственности между командами. Обновлю статью с этими уточнениями. Спасибо за фидбек!
Обновил статью, исправив некоторые недочёты.
Лучше поздно, чем никогда :)
Старался :)
Обновил статью.
Используем
confluentinc/cp-kafkaвместо Bitnami (который переехал в legacy без обновлений).Действительно, с августа 2025 Bitnami переместили все образы в bitnami legacy, где они больше не получают обновления. Статья будет обновлена. Спасибо за фидбек!
Рад помочь!
Ответил на ваши вопросы во второй части :)
Ну поехали :)
Если вы добавляете консьюмеров без групп на топик, то партиции не распределяются. Каждый консьюмер читает все сообщения. Если же с группами, то в рамках одной группы будет распределение партиций.
Теперь давайте рассмотрим ваш кейс с 3 партициями:
Если 1 консьюмер, то он читает все партиции.
Если 2 консьюмера, то один читает две партиции, другой — одну.
Если 3 консьюмера, то каждый читает по одной.
Ещё предлагаю для большего понимания рассмотреть случай, когда консьюмеров в группе больше, чем партиций. В этом случае "лишние" консьюмеры будут простаивать.
Если хотите, чтобы разными сервисами обрабатывались одни и те же сообщения, надо все инстансы в рамках одного сервиса засунуть в одну группу. Так надо сделать для каждого интересующего вас сервиса.
Вторая часть вопроса несколько более сложная. Попытаюсь адекватно объяснить:
Оффсеты напрямую не сопоставляются с консьюмерами. Видите ли, какая ситуация, в топике, ответственном за оффсеты, хранятся сообщения следующего вида:
Ключ —
<group_id, topic, partition>Значение —
<commited_offset, metadata>Это нам даёт возможности для перебалансировки, так как нет жётской связи между оффсетом и конкретным инстансом.
В вашем кейсе с упавшим единственным консьюмером будет примерно следующее:
Kafka переведёт группу в состояние перебалансировки.
Контроллер кластера смаппит сообщение из offser-топика на текущий активный консьюмер.
Как следствие, активный консьюмер начнёт читать с последнего закоммиченного оффсета. То есть никаких потерь не будет, так как мёртвый брокер не мог закоммитить оффсеты для новых сообщений. На всякий случай отмечу, что коммит в Kafka — это некий процесс сохранения оффсета консьюмером.
И в этом случае настройка
latestприменена не будет, так как кафка смаппит оффсет на консьюмера, а эта настройка срабатывает, как вы, надеюсь, помните, когда Kafka не находит оффсет для консьюмера.Надеюсь, нигде не накосячил в объяснениях :)
Обновил статью.
Всё-таки решил дать гарантии доставки здесь, ибо в теоретической статье была бы слишком большая теоретическая вставка :)
Это понятно. Надо же чем-то заплатить за удобство :)
А вообще я говорил конкретно про отказоустойчивость в кластере.
Рад слышать, спасибо!
Я планирую обновить достаточно сильно эту статью. Выкладывал вечером, подуставший :)
Добавлю новый сервис, исправлю некоторые архитектурные косяки. И вставлю осмысленный ключ в примере.
Вам спасибо, что прочитали!
Все будет хорошо. Консьюмеры в одной группе не знают о консьюмерах в другой. Чтение будет происходить независимо.
На конкретное сообщение, как мне известно, настроить нельзя. Можно только настроить на топик. О том, как конкретно настроить, используя Spring Boot, я рассказал в своей свежей практической статье. Рекомендую к прочтению :)
Никак не поможет :)
Мы создаём новую группу не для того, чтобы разгрузить первую. Мы это делаем для того, чтобы эта самая новая группа имела доступ к тем же самым сообщениям, но выполняла другую логику. То есть для независимого чтения одних и тех же данных. Для разгрузки мы увеличиваем количество консьюмеров в рамках одной группы. Тогда уже между ними распределяются партиции и каждый читает свою часть данных.
Согласен с вами. Действительно, слегка запнулся в начале. Поправил статью. Теперь в начале задаю правильный контекст (называю Kafka журналом). Но дальше провожу аналогии с сообщениями, чтобы безболезненно ввести читателя в курс дела.
Спасибо!