I'm trying to commit offset manually in Kafka consumer by using java functions(Consumer) in Spring cloud stream, but the offset is getting committed after java function executed successfully.
enable.auto.commit: false
ack-mode: MANUAL_IMMEDIATE
@Bean
public Consumer<KStream<String, String>> process() {
return stream -> stream.foreach((key, value) -> {
logger.info("Key: " + key + " Value: " + key);
}
}
Can some one help to provide the correct configuration for manual commit offset in Spring cloud kafka consumer function