Unexpected behavior consuming messages after rebalance from async+manual consumer

[AminArria]

This is a PoC for an unexpected behavior we are seeing in spring-kafka 3.0.2

Consumer is configured with asyncAcks and AckMode.MANUAL and as such we expect that when a record from the batch of a poll is not acked the offset is committed until the most recent in the sequence and no further batches are polled. The first part of this assumption is true, we see the offset unchanged, but the second part is not true

Our assumption comes from this part in the docs "The consumer will be paused (no new records delivered) until all the offsets for the previous poll have been committed.", but is this assumption correct?

The (unexpected) behavior

We are seeing that when a poll batch has been processed, but not been fully acknowledge and there is a rebalance event for some reason the instance re-consumes everything and more (beyond the 500 batch size)

This is our setup for the PoCtests:

• Single topic with a single partition and 1200 messages in it,
• App does not ack records with offsets divisible by 10, offset strategy is set to earliest, and we start with 0 instances

Test 1

1. We bring 1st consumer instance up and it consumes 500 messages (1 to 500), and then nothing, consumer seems paused (as expected), and the committed offset is 10
2. Next we bring a 2nd instance up, this causes a rebalance and the topic partition gets assigned to the 2nd instance, thus it polls and consumes 500 messages (from 10 to 510), but committed offset remains at 10
3. Finally the 1st instance goes down causing a rebalance and this causes 2nd instance to process all records (10 to 1200)

Test 2

1. We bring 1st consumer instance up and it consumes 500 messages (1 to 500), and then nothing, consumer seems paused (as expected), and the committed offset is 10
2. Next we bring a 2nd instance up, this causes a rebalance and the topic partition gets assigned to the 1st instance and it processes all records (10 to 1200)

Code

You can find the full code here https://github.com/AminArria/consumer-poc

What I imagine are the important parts

ConsumerConfig.java

    @Bean(LISTENER_CONTAINER_FACTORY)
    public ConcurrentKafkaListenerContainerFactory<byte[], byte[]> ListenerContainerFactory(ConsumerFactory<byte[], byte[]> consumerFactory) {
        final ConcurrentKafkaListenerContainerFactory<byte[], byte[]> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(1);
        factory.getContainerProperties().setAsyncAcks(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.TRACE);
        return factory;
    }

    @Bean
    public ConsumerFactory<byte[], byte[]> consumerFactory() {
        final Map<String, Object> configs = kafkaProperties.buildConsumerProperties();
        configs.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        configs.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        configs.put(org.apache.kafka.clients.consumer.ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, List.of(RoundRobinAssignor.class));
        configs.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        configs.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, List.of("localhost:19092"));
        return new DefaultKafkaConsumerFactory<>(configs);
    }

KafkaListenerService.java

    @KafkaListener(id = "listener-destination",
                   topics = "destination",
                   autoStartup = "true",
                   containerFactory = LISTENER_CONTAINER_FACTORY)
    public void onDestination(final ConsumerRecord<byte[], byte[]> record,
                              final Acknowledgment acknowledgment) {
        accept(record, acknowledgment);
    }

    private void accept(final ConsumerRecord<byte[], byte[]> record,
                        final Acknowledgment acknowledgment) {
        if (record.offset() > 0 && record.offset() % 10 == 0) {
            // no ack
        } else {
            System.out.println("ACK " + record.offset());
            acknowledgment.acknowledge();
        }
    }

[garyrussell]

The reported behavior is exactly as I would have expected; why would you expect something different?

[garyrussell]

As the documentation says, when there are gaps in the commit log, the consumer is paused and won't receive any new records until the gap is filled. I can't explain why you are receiving more than the 500 records after a rebalance, perhaps enabling DEBUG logging will provide some clues.

I'll have to take a look at your example when I get time, hopefully later today.

[garyrussell]

Sorry, I missed the nuances of your description; your repro indeed shows that the container does not pause after the 490 (+10) are delivered after the rebalance.

[garyrussell]

#2581

[garyrussell]

I figured it out; thanks for your patience.

[AminArria]

Amazing, thanks for your help!