Auto commit in kafka with spring cloud stream

Question

I have an application where I would like to perform (n)ack in the Kafka messages manually. According to spring cloud documentation, it should be done with autoCommitOffset spring cloud documentation

However, in my application, even defining such property the header KafkaHeaders.ACKNOWLEDGMENT is still coming as null.

Here is what my configuration looks like

spring.cloud.stream.kafka.binder.brokers=${KAFKA_BROKER_LIST}
spring.cloud.stream.default.contentType=application/json
spring.cloud.stream.bindings.mytopic.destination=MyInputTopic
spring.cloud.stream.bindings.mytopic.group=myConsumerGroup
spring.cloud.stream.kafka.bindings.mytopic.consumer.autoCommitOffset=false

And my consumer:

@StreamListener("myTopic")
public void consume(@NotNull @Valid Message<MyTopic> message) {
    MyTopic payload = message.getPayload();
    Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); // always null
}

I am using java 13 with spring boot 2.2.5.RELEASE and spring cloud Hoxton.SR1

Any help is appreciated.

Answer 1

I found why my consumer was not working as expected:

In my configuration, I have something like spring.cloud.stream.bindings. mytopic.destination=MyInputTopic, however, the stream binding was done like this:

@StreamListener("Mytopic")

Apparently, the configurations prefixed with spring.cloud.stream.bindings are not case sensitive (as all the configurations worked as expected), but the ones prefixed with spring.cloud.stream.kafka.bindings are case sensitive leading to my issue.

Answer 2

I just copied your properties and it works fine for me...

GenericMessage [payload=foo, headers={kafka_offset=0, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@55d4844d, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=MyInputTopic, kafka_receivedTimestamp=1589488691039, kafka_acknowledgment=Acknowledgment for ConsumerRecord(topic = MyInputTopic, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1589488691039, serialized key size = -1, serialized value size = 3, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@572887c3), contentType=application/json, kafka_groupId=myConsumerGroup}]

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @StreamListener(Sink.INPUT)
    public void listen(Message<String> in) {
        System.out.println(in);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
        return args -> {
            template.send("MyInputTopic", "foo".getBytes());
        };
    }

}

spring.cloud.stream.default.contentType=application/json
spring.cloud.stream.bindings.input.destination=MyInputTopic
spring.cloud.stream.bindings.input.group=myConsumerGroup
spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset=false