Автоматическое сохранение оффсетов в Kafka

На примере пакета Golang Sarama

Эти строки конфигурации касаются автоматического сохранения оффсетов в Kafka. Давайте подробно рассмотрим, что они означают.

1. config.Consumer.Offsets.AutoCommit.Enable = true

Эта строка включает автоматическое сохранение оффсетов в Kafka. Когда AutoCommit.Enable установлено в true, Kafka будет автоматически сохранять оффсеты после того, как Consumer обработает сообщения. Это позволяет Kafka отслеживать, с какого места должен продолжиться Consumer, если он будет перезапущен или если возникнет сбой.

Что происходит при Enable = true:

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

Пример использования:

• Если Consumer успешно обработал сообщения с оффсетом 100, а затем приложение перезапустилось, Kafka помнит, что этот оффсет был обработан, и при следующем запуске Consumer будет читать с оффсета 101.

Возможные значения:

• true — включение автоматического коммита оффсетов.
• false — отключение автоматического сохранения оффсетов. В этом случае вы должны вручную подтверждать (коммитить) оффсеты после обработки сообщений.

---

2. config.Consumer.Offsets.AutoCommit.Interval = 10 * 1000

Эта строка задает интервал времени (в миллисекундах), через который Kafka будет автоматически сохранять оффсеты, если AutoCommit.Enable установлено в true.

В этом примере 10 * 1000 = 10,000 миллисекунд (10 секунд).

Что это значит?

Каждые 10 секунд Kafka будет автоматически сохранять текущие оффсеты, т.е. зафиксирует текущие позиции, с которых был прочитан топик. Это гарантирует, что даже если Consumer будет обрабатывать сообщения в течение длительного времени, его позиция будет отслеживаться и обновляться периодически.

Пример:

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

Возможные значения:

• Это значение в миллисекундах. Значение по умолчанию — 5 секунд.
• Вы можете настроить это значение в зависимости от ваших требований. Например, если вы хотите, чтобы коммиты происходили чаще, установите меньший интервал (например, 1 секунду), если необходимо уменьшить нагрузку на Kafka — увеличьте интервал.

---

Пример использования с пояснением

В этом примере:

• AutoCommit.Enable = true означает, что Kafka будет автоматически коммитить оффсеты, после того как Consumer обработает сообщения.
• AutoCommit.Interval = 10 * 1000 устанавливает интервал в 10 секунд. Это означает, что каждые 10 секунд Kafka будет автоматически сохранять текущие оффсеты, что позволяет продолжить чтение с последнего обработанного сообщения после перезапуска Consumer.

Когда использовать AutoCommit?

• Когда важна простота: Автоматическое сохранение оффсетов удобно для случаев, когда вы хотите минимизировать необходимость в ручном управлении оффсетами.
• Для простых и быстрых приложений: Если ваше приложение не имеет сложной логики обработки ошибок, и вы хотите, чтобы Kafka автоматически отслеживала, какие сообщения были обработаны.
• Если вам не важно точно зафиксировать, какие сообщения были обработаны: Например, если потеря одного сообщения не критична.

Когда не использовать AutoCommit?

• Если нужна высокая точность в сохранении состояния: Если вам нужно точно контролировать, какие сообщения были обработаны, используйте ручной коммит. Это позволяет вам сохранять оффсеты только после успешной обработки всех сообщений, что уменьшает вероятность потери данных.
• Если хотите избежать двойной обработки: В случае, если вы не хотите, чтобы сообщения обрабатывались повторно после сбоя, рекомендуется вручную подтверждать (коммитить) оффсеты только после успешной обработки сообщения. Это гарантирует, что если приложение упадет до того, как будет сохранен оффсет, сообщения не будут считаться обработанными.

Ручной коммит оффсетов

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

Таким образом, вам нужно будет контролировать, когда и как подтверждать оффсеты в зависимости от логики обработки сообщений.

По умолчанию в библиотеке Sarama для Go настройки коммита оффсетов настроены следующим образом:

1. config.Consumer.Offsets.AutoCommit.Enable

• По умолчанию, эта настройка включена (true).
• Это означает, что автоматическое сохранение оффсетов включено, и Kafka будет автоматически сохранять оффсеты после обработки сообщений.

2. config.Consumer.Offsets.AutoCommit.Interval

• По умолчанию, этот интервал установлен на 5 секунд (5000 миллисекунд).
• То есть, каждые 5 секунд Kafka будет автоматически сохранять текущие оффсеты, если включен автоматический коммит.

Итог:

По умолчанию, Sarama автоматически коммитит оффсеты каждые 5 секунд после обработки сообщений, если AutoCommit.Enable установлено в true.