consumer: auto.offset.store commits incorrect offsets

[iblislin]

Sponsored-by: China Medical University Hospital

--
Hi,

I guess this issue is introduced by #4208, but I do not understand the details in that issue.

I inspected the partition offset via on_commit by the Python library.
W/o this patch, there are two case tried and I found the offsets committed are different:

case 1.

The consumer is configured with

c = Consumer({
    'bootstrap.servers': os.environ.get('BOOTSTRAP_SERVERS'),
    'group.id': 'magic',
    'auto.offset.reset': 'earliest',
    'enable.auto.offset.store': True,
    'enable.auto.commit': True,
    'on_commit': on_commit,
})

where the on_commit function is:

from confluent_kafka import  TopicPartition
                                          
def on_commit(err, partitions: list[TopicPartition]):
    for i in sorted(partitions, key=lambda x: x.partition):
        print(f'{i.partition}\t-> {i.offset}')

case 2

c = Consumer({
    'bootstrap.servers': os.environ.get('BOOTSTRAP_SERVERS'),
    'group.id': 'magic2',
    'auto.offset.reset': 'earliest',
    'enable.auto.offset.store': False,
    'enable.auto.commit': True,
    'on_commit': on_commit,
})

and then I invoked the c.store_offset(message=msg) manually, after processed a message.

results

I run case 1 and case 2 twice. The first run is for reading the topic to the end and waiting for the offsets committed.
The expected result in the second run is that the consumers should not get any messages, since there aren't any producers running at this point.

The case 2 works correctly in the second run.
The case 1 starts consuming the log from some unexpected point of offset in the second run.

This patch fixed case 1. But I'm not sure of the side effect of it.

[cla-assistant]

All committers have signed the CLA.