One of the _retrieve_offsets calls is missing the timeout_ms variable - this leads to an infinite while loop

[hntirgeam]

Hi.

When trying to get max offset of the topic like so:

consumer.assign(partitions)
consumer.seek_to_end(*partitions)
max_offset = consumer.position(partitions[0])

consumer.position(partitions[0]) call get gets stuck in endless while loop due to the missing timeout_ms variable passed to the _retrieve_offsets method call.

I can't tell for sure if this was done on purpose or just omitted - this variable is passed to other calls to this method.

I think timeout_ms should be passed to _reset_offset so that it can then be passed to _retrieve_offsets.

kafka-python/kafka/consumer/fetcher.py

Line 237 in 0864817

offsets = self._retrieve_offsets({partition: timestamp})