Ensure timeout is checked after each fetch position update

[k61n]

For Consumer.position() I had a case when self._update_fetch_positions always resulted in True and position was always None. This hangs the loop for ~305 seconds, and it appears to me that the line

elif timer.expired:

is never reached.

The root cause is probably how timeout_ms value is propagated and checked in the following methods, however I believe my change is still valid as the timeout check should work disregarding of self._update_fetch_positions value.

[bpedersen2]

if position is None and time.expired ?

[dpkp]

I had a case when self._update_fetch_positions always resulted in True and position was always None

Do you have any more details on this case? The internal contract is/should be that _update_fetch_positions returns True iff all assigned partitions have positions.

[k61n]

Yes,_update_fetch_positions returns True.
At the same time self._subscription.assignment[partition].position returns None every time andelif timer.expired is never reached.

[k61n]

Here is a minimal example.
You'd need a fresh kafka instance

FROM apache/kafka:4.0.0

COPY server.properties /kafka/config/kraft/

EXPOSE 9092

with server.properties

node.id=0
process.roles=controller,broker
controller.listener.names=CONTROLLER
inter.broker.listener.name=PLAINTEXT
controller.quorum.voters=0@localhost:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://localhost:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
auto.create.topics.enable=true

build and run it:

docker build -t kafka .
docker run --rm --name kafka -p 9092:9092 -it kafka

And python code

import time

from kafka import KafkaConsumer, KafkaProducer, TopicPartition

prod = KafkaProducer(bootstrap_servers='localhost:9092')

cons = KafkaConsumer(
            bootstrap_servers='localhost:9092',
            auto_offset_reset='earliest',
            session_timeout_ms=1000,
        )

partitions = cons.partitions_for_topic('test')
cons.assign([TopicPartition('test', p) for p in partitions])

end = cons.end_offsets(list(cons.assignment()))
for partition in cons.assignment():
    print(1)
    while (pos := cons.position(partition, timeout_ms=1000)) \
            is not None and pos < end[partition]:
        print(2)

prod.send(
    topic='test',
    value=b'0000',
    key=b'1111',
    timestamp_ms=int(time.time() * 1000))
prod.flush()

First run finishes normally, on 2nd run it hangs on cons.position(partition, timeout_ms=1000).