Batch Listener(AckMode.MANUAL_IMMEDIATE) acks message even if acknowledgment.acknowledge() is not invoked

[HimanshuLakra]

Hi
I have a query regarding working of MANUAL_IMMEDIATE mode with BatchAcknowledgingMessageListener.
In the onMessage function if I do not invoke acknowledgment.acknowledge() the messages are still getting acked, is this expected?

Spring Kafka: 3.3.0
enable.auto.commit = false
AckMode: AckMode.MANUAL_IMMEDIATE

Below is my onMessage function of
public class KafkaMessageBatchManualModeListener implements BatchAcknowledgingMessageListener<String, Object>

`@Override
public void onMessage(List<ConsumerRecord<String, Object>> data, Acknowledgment acknowledgment) {

    if(CollectionUtils.isEmpty(data)) {
        return;
    }

    Map<Integer, CompletableFuture<EventStatus>> futures = new LinkedHashMap<>();
    int recordIterationIndex = 0;
    String batchId = UUID.randomUUID().toString();

    try{
        for (ConsumerRecord<String, Object> record : data) {
            CompletableFuture<EventStatus> future = CompletableFuture.supplyAsync(
                    () -> {
                        return kafkaMessageProcessor.processMessage(createKafkaEventRecord(record, batchId));
                    }, kafkaFanoutExecutor
            );

            logger.info("Message Received with BatchId: {}, batchIndex: {}, Offset: {}, Partition: {}", batchId, recordIterationIndex, record.offset(), record.partition());
            futures.put(recordIterationIndex, future);
            recordIterationIndex += 1;
        }

        // Wait for completion of all futures
        waitForFuturesProcessing(futures);

        Integer ackIndex = getAcknowledgmentIndexForBatch(futures);

        // Acknowledge the messages
        if(ackIndex != null){
            logger.info("Acknowledging messages upto index:{} for batch id: {}", ackIndex, batchId);
            acknowledgment.acknowledge(ackIndex);
            logger.info("Acknowledged messages upto index:{} for batch id: {}", ackIndex, batchId);
        }else{
            logger.warn("No message to acknowledge for batch id: {} ", batchId);
        }
    }catch(Exception exception){
        logger.error("Error while listening message in KafkaMessageBatchManualModeListener", exception);
    }`

My expectation is if I do not invoke acknowledgment.acknowledge(index) function (ref: else block or catch block case)
then my messages should get redelivered in next poll, but they get acked.
To stop that ack I have to explicitly use nack

`if(ackIndex != null){

            logger.info("Acknowledging messages upto index:{} for batch id: {}", ackIndex, batchId);
            
            acknowledgment.acknowledge(ackIndex);
            
            logger.info("Acknowledged messages upto index:{} for batch id: {}", ackIndex, batchId);
            
        }else{
        
            logger.warn("No message to acknowledge for batch id: {} ", batchId);
            
            acknowledgment.nack(0, Duration.ZERO);
            
        }

`

[HimanshuLakra]

Hi @artembilan @sobychacko, Could anyone please help with above query? Thanks

[sobychacko]

When using MANUAL_IMMEDIATE, the container will immediately acknowledge and commit, unlike MANUAL, where the ack only happens at the end of the entire batch received from the last poll.

You're seeing expected behavior here. When you don't explicitly call acknowledge, the last committed offset on the partition is still the one you acknowledged successfully last time. However, upon the next poll in the consumer loop (assuming the application does not restart), you will receive the next record (+1 from the previous poll) regardless of where your committed offset is.

Imagine this: You have nine records in the partition. In your first batch, let's say you receive three records (offsets 0, 1, and 2). However, you only acknowledge the first one, meaning the committed offset is at 1. However, the next poll in the consumer loop will deliver the next three records (offsets 3, 4, and 5). After processing this, your committed offset will still be at offset 1. Now, let's say you restart your application. You will receive records from offset 1 onwards since that is your last committed offset. As long as you are in the same consumer loop, you will continue receiving new batch records regardless of ack and commit.

If you want to explicitly redeliver records from a prior poll due to some issues, you either have to use nack() as you did above, which will re-seek the consumer to that particular offset, or you can manually seek on the consumer, but nack() automatically does that.

To recap, you see the expected behavior and by design.

[sobychacko]

@HimanshuLakra, I clarified my answer above as there was some vagueness in my original answer. Please take a look.