Как на самом деле работает auto-commit в Kafka и можем ли мы на него расчитывать?

В этой статье я хочу чуть подробнее раскрыть как же устроен механизм auto-commit у слушателей в библиотеке kafka-clients (рассмотрим версию 2.6.0)

В документации мы можем найти следующую формулировку описывающую работу auto-commit:

Auto-commit basically works as a cron with a period set through the auto.commit.interval.ms configuration property. If the consumer crashes, then after a restart or a rebalance, the position of all partitions owned by the crashed consumer will be reset to the last committed offset.

В java docs к KafkaConsumer в свою очередь присутствует следующее описание:

The consumer can either automatically commit offsets periodically; or it can choose to control this committed position manually by calling one of the commit APIs (e.g. commitSync and commitAsync).

Из этих формулировок может сложиться заблуждение о том, что неблокирующий автоматический коммит смещения (offset) происходит в фоновом режиме и не совсем понятно как он связан с процессом получения сообщений конкретным слушателем (consumer) и самое главное какие гарантии доставки мы при этом имеем?

Давайте подробнее разберем механизм получения сообщений слушателем с настройкой enable.auto.commit = true на примере реализации класса KafkaConsumer из библиотеки org.apache.kafka:kafka-clients:2.6.0 

Для этого рассмотрим пример приведённый в java doc KafkaConsumer:

Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "test"); props.setProperty("enable.auto.commit", "true"); props.setProperty("auto.commit.interval.ms", "1000"); props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) {   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));   for (ConsumerRecord<String, String> record : records)     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); }

Как же происходит auto-commit в данном случае? Разгадку стоит искать в самом методе для получения новых сообщений 

consumer.poll(Duration.ofMillis(100));

Но сперва сделаем шаг назад. При инициализации класса KafkaConsumer параметры относящиеся к auto-commit такие как enable.auto.commit и auto.commit.interval.ms предаются в экземпляр класса ConsumerCoordinator одно из назначений которого как раз и состоит в том, чтобы реализовать механизм auto-commit.

Обратим внимание на метод maybeAutoCommitOffsetsAsync из этого класса

public void maybeAutoCommitOffsetsAsync(long now) {   if (autoCommitEnabled) {     nextAutoCommitTimer.update(now);     if (nextAutoCommitTimer.isExpired()) {       nextAutoCommitTimer.reset(autoCommitIntervalMs);       doAutoCommitOffsetsAsync();     }   } }

При установленном свойстве enable.auto.commit = true и в случае если установленный в параметре auto.commit.interval.ms интервал времени с момента последнего коммита истек именно тут происходит коммит последнего смещения, а в случае неудачи поток выполнения не прерывается, а происходит просто логирование ошибки (метод doAutoCommitOffsetsAsync)

private void doAutoCommitOffsetsAsync() {   Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();   log.debug("Sending asynchronous auto-commit of offsets {}", allConsumedOffsets);    commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> {     if (exception != null) {       if (exception instanceof RetriableCommitFailedException) {         log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", offsets,                   exception);         nextAutoCommitTimer.updateAndReset(rebalanceConfig.retryBackoffMs);       } else {         log.warn("Asynchronous auto-commit of offsets {} failed: {}", offsets, exception.getMessage());       }     } else {       log.debug("Completed asynchronous auto-commit of offsets {}", offsets);     }   }); }

Теперь вернемся назад в метод poll класса KafkaConsumer. Во время выполнения метода перед получением новых данных из топиков происходит вызов метода updateAssignmentMetadataIfNeeded, который в свою очередь обращается к методу poll класса ConsumerCoordinator, в самом конце, которого вызывается уже известный нам maybeAutoCommitOffsetsAsync

Таким образом собирая нашу матрешку получаем следующую последовательность действий при получении новых сообщений в методе poll класса KafkaConsumer:

1. Слушатель сохраняет предыдущий offset

2. Далее присходит получение новых сообщений из топиков с которыми связан данный слушатель.

Учитывая что слушатель KafkaConsumer не является потокобезопасным и выполняется одним потоком можем заключить, что операции выполняются строго последовательно. 

Стоит отметить что п.1 выполняется если enable.auto.commit = true и с момента прошлого коммита прошло больше времени чем указано в параметре auto.commit.interval.ms. Т.е. если метод poll() у нашего слушателя вызывается каждые 3 секунды, а значение auto.commit.interval.ms=6000, то смещение будет сохраняться каждый второй вызов. 

А что же с гарантиями доставки? При такой реализации мы имеем вполне честный “at least once delivery”, что было бы затруднительно при сохранении смещений отдельным потоком.