SpringKafka not honoring auto commit setting to false

user_6655151 0
Viewed 31

Probably I am missing a setting or two here but with KafkaListenerContainerFactory setup for ackMode as manual, I see messages are streaming through @KafkaListener method when I print them on receive. We are not acknowledging the message anywhere in the application. Appreciate if anyone can point out where the issue is. Thanks in advance.

Using spring kafka version -- 2.3.0.RELEASE

ListenerContainerFactory

    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<GenericData.Record, GenericData.Record>>
    kafkaListenerContainerFactory(
            ConsumerFactory<GenericData.Record, GenericData.Record> consumerFactory) {

        requireNonNull(consumerFactory, "consumerFactory must not be null");

        ConcurrentKafkaListenerContainerFactory<GenericData.Record, GenericData.Record> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        factory.setConcurrency(4);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

Receives the messages here

    public List<InventoryRecord> onMessage(@Payload List<ConsumerRecord<byte[],
            GenericArray<GenericRecord>>> consumerRecords, Acknowledgment acknowledgment) {

application.yml

  kafka:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      heartbeat-interval: 50000ms
      client-id: card-processor-consumer
      group-id: card-processor-consumer
      max-poll-records: 24
    properties:
      basic:
        auth:
          credentials:
            source: SASL_INHERIT
      security.protocol: SASL_SSL
      sasl.mechanism: SCRAM-SHA-512
      sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username=zzzzzzz password=xxxxxxxxx;
      request.timeout.ms: 1800000
      session.timeout.ms: 360000
      max.poll.interval.ms: 290000
    listener:
        concurrency: 4
        ack-mode: manual
spring-kafka kafka-consumer-api apache-kafka java
1 Answers

and my understanding is the next message will not arrive until the last message offset is committed.

Your understanding is incorrect.

Kafka maintains two pointers for each group/partition - the current position and the committed offset.

The current position is the next record that will be read on the next poll. The committed offset is the next record that will be read by this consumer when it is next restarted, or a rebalance occurs. i.e. the position is reset to the committed offset when the consumer leaves the group, or a partition is revoked.

There is no other relationship between these pointers; committing the offset has no bearing on fetching the next record from the current position.

user_12407630
Related