Где хранится Offset в Kafka?

Хранение Offset в Apache Kafka

В Apache Kafka offset (смещение) — это уникальный идентификатор позиции сообщения внутри партиции топика. Это ключевой механизм для отслеживания прогресса потребителей (consumers). Место хранения offset зависит от используемого подхода и версии Kafka.

Основные места хранения offset

1. Apache Kafka (внутренний топик __consumer_offsets)

• Основной и современный способ для потребителей из Consumer Group.
• Специальный, компактный (compacted) внутренний топик Kafka, создаваемый автоматически.
• Каждая запись содержит:

     * Ключ: `Group ID`, `Topic`, `Partition`
     * Значение: `Offset`, `metadata`, `timestamp`

• Потребитель (Kafka Consumer API) периодически автоматически коммитит (сохраняет) текущий offset в этот топик с помощью механизма auto-commit или вручную через commitSync()/commitAsync().
• Пример кода (auto-commit):

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

2. Внешние системы хранения (например, база данных)

• Используется для более точного контроля, особенно в паттерне exactly-once semantics.
• Потребитель явно сохраняет offset во внешнем хранилище (например, PostgreSQL, Redis, Apache ZooKeeper — устаревший способ) вместе с результатом обработки сообщения в рамках единой транзакции.
• Это гарантирует, что offset обновится только если сообщение успешно обработано и сохранено.
• Пример паттерна:

// Пример логики на Go (псевдокод)
func processMessage(tx *sql.Tx, msg kafka.Message) error {
    // 1. Обработать сообщение
    result := businessLogic(msg.Value)
    // 2. Сохранить результат в БД
    err := saveResult(tx, result)
    if err != nil {
        return err
    }
    // 3. Сохранить offset в отдельную таблицу в ТОЙ ЖЕ транзакции
    err = saveOffset(tx, msg.Topic, msg.Partition, msg.Offset)
    return err
}

3. Локально у потребителя (при отключенном auto-commit и без коммита в группу)

• Потребитель просто отслеживает позицию в памяти, но это ненадежно. При перезапуске потребитель начнет чтение с позиции, определенной настройкой auto.offset.reset (earliest/latest/none).

Детали работы с топиком __consumer_offsets

• Формат: Компактный топик. Для каждой группы сохраняется только последнее значение offset для комбинации topic-partition.
• Репликация: По умолчанию реплицируется, как и обычные топики (фактор репликации обычно высокий для отказоустойчивости).
• Партиционирование: Партиционируется по Group ID. Это позволяет распределить нагрузку по запись/чтение offset между брокерами.
• Чтение offset при ребалансе: Когда потребитель присоединяется к группе или происходит rebalance, он читает "свой" последний коммитнутый offset из этого топика, чтобы продолжить чтение с нужного места.

Важные настройки и сценарии

• auto.offset.reset: Определяет поведение потребителя, если offset для topic-partition не найден (например, при первом запуске группы). Значения: earliest (с начала), latest (с новых сообщений), none (выбросить исключение).
• Изоляция потребителей: Потребитель никогда не читает чужие offset из __consumer_offsets. Он читает только offset для своей group.id.
• Проблемы и решения:
  • Duplicate processing: Может возникнуть при auto-commit, если потребитель успел обработать сообщение, но упал до коммита offset. При рестарте он прочитает те же сообщения снова.
  • Lost messages: Может возникнуть, если потребитель закоммитил offset, но не успел обработать сообщение (например, при crash после коммита, но до сохранения результата бизнес-логики).

Резюме

Offset — это указатель на позицию в партиции. Для потребителей из группы он по умолчанию надежно хранится в реплицированном внутреннем топике Kafka __consumer_offsets. Для реализации семантики "точно один раз" или при отсутствии группы часто используется внешнее транзакционное хранилище. Выбор стратегии хранения напрямую влияет на гарантии доставки сообщений (at-most-once, at-least-once, exactly-once) и отказоустойчивость приложения. Современные Kafka-клиенты, как в Java, так и в Go (например, sarama), предоставляют API для гибкого управления этим процессом.