Set the enable-auto-commit property to false:
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
Set the ack-mode to MANUAL_IMMEDIATE:
factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
Then, in your consumer/listener code, you can commit the offset manually, like this:
@KafkaListener(topics = "testKafka")
public void receive(ConsumerRecord<?, ?> consumerRecord,
Acknowledgment acknowledgment) {
System.out.println("Received message: ");
System.out.println(consumerRecord.value().toString());
acknowledgment.acknowledge();
}
Update: I created a small POC for this. Check it out here, might help you.