In AckMode MANUAL_IMMEDIATE, is acknowledgment.acknowledge() optional ?

[coderbunty]

@KafkaListener(
      groupId = "demo-consumer-group",
      topics = {"demo_topic"},
      containerFactory = "demoListenerFactory")
  public void consumeDemoEvent(
      @Payload ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
    try {
      String key = consumerRecord.key();
      String value = consumerRecord.value();
      processDemoEvent(value);
    } catch (CustomException e) {
      log.error("Error while processing record: ", e);
    } finally {
      acknowledgment.acknowledge();
    }
  }

@Bean("demoListenerFactory")
  public ConcurrentKafkaListenerContainerFactory<String, String>
      demoKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(demoConsumerFactory());
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setAutoStartup(true);
    return factory;
  }

@Bean
  public ConsumerFactory<String, String> demoConsumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
  }

Consider the below 2 scenarios,

1. processDemoEvent(value) runs without any issues, and we don't execute acknowledgment.acknowledge() in finally block --> even then the offset gets committed even though I am using manual ack configuration
2. processDemoEvent(value) throws an exception which we don't catch & then in the finally block execute acknowledgment.acknowledge() --> even though we acknowledged, the offset is not committed

While I see acknowledgment.nack() working fine, from the above scenarios it looks like we can skip using acknowledgment.acknowledge() as it not contributing anything.

Is this a bug or am I missing something here ?

[garyrussell]

It is not "optional", calling acknowledge() commits the offset; if there are no more records and you restart the consumer, the record will not be redelivered.

Conversely, if you exit without acknowledging, the offset won't be committed and, if there are no records after this one, and you restart the consumer, the record will be redelivered.

Kafka maintains two pointers, the current position and the committed offset. Each poll returns record(s) from the current position. When a consumer starts, the current position is set to the committed offset (if there is one).

It might "appear" to be optional because, if there are more records (successfully processed) after the one you failed to acknowledge, then their offsets will be committed (and by definition, their offsets will be after the one you failed to commit).

[garyrussell]

In the uncaught exception case, the default error handler re-seeks the record, resetting the position, effectively undoing the commit locally, but not on the broker.

[coderbunty]

@garyrussell Thanks for the explanation, it makes sense. Tried from my side as well, restarting the consumer makes it clear that the offset wasn't committed without acknowledgement.acknowledge()