Как мне установить свойства производителя kafka для атомарного потребителя kafka

1. enable.auto.commit
   Когда потребитель перезагружается после сбоя, он начинает использовать раздел раздела со смещения, полученного из базы данных. Значение этого свойства будет использоваться в одном безаварийном прогоне потребителя, как и в любом другом сценарии.

   Автоматическая фиксация совершает фиксацию потребителя каждые 5 секунд (значение по умолчанию auto.commit.interval.ms) или на каждой poll() звонки.

   Ручная фиксация (enable.auto.commit=false) помогает избежать отказа от обработки сообщения. Например, если 5-секундный таймер автоматической фиксации истекает между чтением и обработкой сообщения, он может выполнить: (read, commit, process) в этой последовательности. И если потребитель выйдет из строя после фиксации без обработки сообщения (read,commit,crash), это сообщение никогда не будет обработано, потому что на следующем poll() потребитель получит следующее сообщение (поскольку фиксация была успешной). Мы можем предотвратить это, выполнив ручную фиксацию в следующем порядке: (read, process,commit).

   Однако в этом случае существует вероятность того, что потребитель может выйти из строя после обработки без фиксации (read,process,crash). Это приведет к повторной обработке того же сообщения в следующем poll().

   Чтобы избежать этой дублирующей обработки, мы сохраняем смещение во внешней базе данных и получаем его при перезапуске потребителя. Обратите внимание, что сохранение смещения в базе данных и его выборка при перезапуске потребителя также позволяет избежать обработки сообщения в случае последовательности (read,commit,crash), которая может произойти в случае автоматической фиксации.

   Таким образом, вкратце, ручная фиксация не служит никакой цели, когда мы сохраняем смещение во внешней базе данных. Следовательно, мы можем установить enable.auto.commit в любое значение true или false. Однако, если установлено значение false, мы не должны забывать о явной фиксации. В противном случае потребитель будет продолжать читать и обрабатывать одно и то же сообщение.

2. auto.offset.reset

   Его значение будет иметь значение, если в базе данных нет значения смещения для данного раздела темы. Это произойдет при первом запуске потребителя или при усечении базы данных. В этом случае нам нужно, чтобы потребитель начал использовать с первого сообщения, которое не было получено ни одним из потребителей в его группе потребителей. Для этого нам нужно установить свойство на earliest.

Если вы используете Kafka Stream, он поддерживает шаблон потока Exactly-once, вы можете сослаться на

Шаблон потока «Ровно один раз» - это просто возможность выполнить операцию чтения-обработки-записи ровно один раз. Означает, что вы потребляете одно сообщение за раз, получаете процесс, публикуете его в другой теме и фиксируете. Таким образом, фиксация будет обрабатываться Stream автоматически, по одному сообщению за раз.

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

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

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

enable.auto.commit: В случае использования внешнего управления смещением вы не используете сценарий фиксации, поэтому не имеет значения, включен или отключен enable.auto.commit. Вы будете продолжать получать сообщения, используя метод seek (раздел TopicPartition, длинное смещение) и сохраняя смещение извне. В случае перезапуска начните выборку последнего сохраненного смещения. Единственное влияние, если какая-то встроенная панель управления, такая как Confluent Control-Center, Grafana и т. Д., Используемая для поддержки темы Kafka, не будет отражаться в случае, если не будет сделана ручная фиксация и enable.auto.commit false. .

auto.offset.reset Да, это влияет только при первом запуске, но поскольку вы используете для получения сообщения из определенного раздела и смещения, никакого воздействия.

======================= Обновлено =======================

enable.auto.commit - если true (по умолчанию), периодически фиксировать смещение последнего сообщения, переданного приложению. Принятое смещение будет использоваться при перезапуске процесса, чтобы продолжить с того места, где он остановился.

auto.commit.interval.ms - частота в миллисекундах, с которой смещения потребителя фиксируются (записываются) в хранилище смещений.

Примечание. если enable.auto.commit false, auto.commit.interval.ms не используется.

В случае enable.auto.commit истинный метод фиксации будет вызывать при каждом опросе, и если auto.commit.interval.ms будет передан, смещение будет зафиксировано

1. интервал опроса> интервал фиксации: смещение фиксации во время интервала опроса
2. интервал опроса ‹интервал фиксации: метод фиксации будет вызывать при каждом опросе, но смещение будет зафиксировано при последовательном опросе () после того, как интервал фиксации пройден.