Details
Description
This is a followup ticket based on a question from junrao when reviewing the fetch request manager pull request:
It still seems weird that we only use the timer for refreshCommittedOffsetsIfNeeded, but not for other cases where we don't have valid fetch positions. For example, if all partitions are in AWAIT_VALIDATION state, it seems that PrototypeAsyncConsumer.poll() will just go in a busy loop, which is not efficient.
The goal here is to determine if we should also be propagating the Timer to the validate positions and reset positions operations.
Note: we should also investigate if the existing KafkaConsumer implementation should be fixed, too.