How does auto-commit actually work in Kafka and can we count on it?

In this article, I want to reveal in more detail how the auto-commit mechanism works for listeners in the kafka-clients library (consider version 2.6.0)

In the documentation, we can find the following formulation describing how auto-commit works:

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.

The java docs for KafkaConsumer, in turn, contain the following description:

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 (eg commitSync and commitAsync).

From these formulations, a misconception may arise that a non-blocking automatic offset commit occurs in the background and it is not entirely clear how it relates to the process of receiving messages by a specific consumer and, most importantly, what delivery guarantees do we have?

Let's take a closer look at the mechanism for receiving messages by the listener with the enable.auto.commit = true setting using the example of the implementation of the KafkaConsumer class from the org.apache.kafka library : kafka-clients: 2.6.0 

To do this, consider the example given in the 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());
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

How does auto-commit happen in this case? The answer should be found in the method itself for receiving new messages. 

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”, .