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