Кто управляет offset — Kafka или consumer?

В Apache Kafka offset — это числовой идентификатор позиции consumer внутри партиции топика. Управление этим смещением — это совместный процесс между брокером Kafka (хранилищем) и consumer-клиентом (логикой).

Роль Kafka (брокера)

Kafka отвечает за хранение и предоставление offset. Для этого используется внутренний, компактный топик __consumer_offsets. Когда consumer подтверждает (commits) offset, Kafka сохраняет эту пару «consumer group + топик + партиция → offset» в этом топике. При перезапуске или перебалансировке consumer-группы, новый consumer запрашивает у Kafka последний сохранённый offset, чтобы продолжить чтение с нужного места.

Роль Consumer

Consumer управляет логикой продвижения offset. После успешной обработки сообщения consumer решает, когда отправить commit в Kafka. Это можно делать автоматически (через enable.auto.commit=true) или вручную. Ручное управление (consumer.commitSync()) рекомендуется для точного контроля, чтобы избежать потери или повторной обработки сообщений при сбоях.

Пример кода (ручное управление offset)

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false"); // Отключаем авто-коммит
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

try {
    while (true) {
        ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord record : records) {
            // 1. Обрабатываем сообщение
            System.out.printf("offset = %d, key = %s, value = %s%n",
                              record.offset(), record.key(), record.value());
            // 2. После успешной обработки — коммитим offset
            //    Коммит текущей позиции (синхронно, для надёжности)
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

В этом примере consumer явно подтверждает offset после обработки каждого сообщения. Если обработка прервётся до вызова commitSync(), то при следующем запуске consumer прочитает то же сообщение снова (at-least-once семантика).

Практическое применение и настройки

• enable.auto.commit: Упрощает код, но может привести к потере данных (если consumer упал после коммита, но до обработки) или дублированию (если коммит происходит до фактической обработки).
• isolation.level: Настройка read_committed позволяет consumer читать только те сообщения, чьи транзакции были подтверждены (при использовании Kafka Transactions).
• Автоматическая перебалансировка: При изменении состава consumer-группы, Kafka координирует перераспределение партиций, а каждый consumer коммитит свои offset перед освобождением партиции.

Вывод: Управление offset — это разделённая ответственность, где Kafka выступает как надёжное хранилище состояния, а consumer — как логический контроллер, решающий, когда сообщение считается успешно обработанным. Ручное управление offset стоит применять в системах, где критична гарантированная доставка без потерь или дубликатов.