Что такое auto-commit в Kafka?

Что такое auto-commit в Kafka

Auto-commit — это механизм в Apache Kafka, который автоматически сохраняет смещение (offset) обработанного сообщения, позволяя потребителю отслеживать, какие сообщения он уже обработал. Это позволяет избежать повторной обработки сообщений при перезапуске потребителя.

Как работает auto-commit

Kafka потребитель отслеживает позицию в разделе (partition) через offset — номер последнего обработанного сообщения. Когда auto-commit включен, Kafka автоматически сохраняет этот offset в специальном топике __consumer_offsets.

Потребитель читает сообщение
        ↓
Обрабатывает сообщение
        ↓
Автоматически коммитит offset (сохраняет позицию)
        ↓
Примерно через enable.auto.commit.interval.ms (по умолчанию 5 сек)

Пример с auto-commit включенным

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.util.Properties;
import java.util.Collections;

public class KafkaAutoCommitExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        
        // Включаем auto-commit (по умолчанию true)
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        
        // Интервал между коммитами (по умолчанию 5000 мс)
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Partition: " + record.partition() + 
                    ", Offset: " + record.offset() + 
                    ", Value: " + record.value());
                // Сообщение автоматически будет закоммичено
                // через enable.auto.commit.interval.ms
            }
        }
    }
}

Проблемы с auto-commit

Проблема 1: Потеря сообщений

Если потребитель упадёт сразу после коммита offset, но до обработки сообщения, сообщение будет потеряно:

Оффсет 0: прочитано и закоммичено
Оффсет 1: прочитано, коммит ещё не произошёл
Потребитель падает → Офсет 1 потеряется

Проблема 2: Дублирование сообщений

Если потребитель падёт между обработкой сообщения и коммитом, при перезапуске он снова обработает то же сообщение.

Manual commit — лучшая практика

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import java.util.Properties;
import java.util.Collections;

public class KafkaManualCommitExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        
        // Отключаем auto-commit
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                try {
                    // Обрабатываем сообщение
                    System.out.println("Processing: " + record.value());
                    processMessage(record.value());
                    
                    // Только после успешной обработки коммитим
                    consumer.commitSync();
                } catch (Exception e) {
                    System.err.println("Error processing message: " + e.getMessage());
                    // Коммит не происходит, сообщение будет обработано снова
                }
            }
        }
    }
    
    private static void processMessage(String message) throws Exception {
        // Обработка сообщения
    }
}

Асинхронный коммит

public class KafkaAsyncCommitExample {
    public static void main(String[] args) {
        // ... конфиг с enable.auto.commit = false ...
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Processing: " + record.value());
                processMessage(record.value());
            }
            
            // Асинхронный коммит с callback
            consumer.commitAsync((offsets, exception) -> {
                if (exception != null) {
                    System.err.println("Commit failed: " + exception.getMessage());
                } else {
                    System.out.println("Committed offsets: " + offsets);
                }
            });
        }
    }
}

Рекомендации

Когда использовать auto-commit vs manual

• Auto-commit: простые сценарии, где допустимо дублирование или потеря сообщений
• Manual commit: критичные данные, финансовые операции, где нужна гарантия обработки

Выбор между auto-commit и manual commit — ключевой момент при разработке потребителей Kafka, влияющий на надёжность системы.