KAFKA-20514: kraft observers should use previous fetch response to decide where to send the next fetch

[kevin-wu24]

Background

Currently, there is a timing issue where a KRaft observer can be stuck fetching from the leader if the next poll occurs after the previous fetch's backoff has completed, and the previous request did not time out. An example of this is if the leader's advertised endpoints are not routable. The bootstrap server endpoints could contain routable endpoints for the leader, but the observer would be stuck fetching from the unroutable endpoints.

Previously, there was an issue where observers could be stuck fetching from the bootstrap servers even if it discovers leader endpoints from the bootstrap fetch. This is because the fetch timeout is not reset on the observer.

What changed

Observer fetching logic should ensure that within the same epoch, all the bootstrap server endpoints and the leader have a chance to serve fetch requests. This logic should be independent of request manager's state.

• If the fetch timeout has not expired, the observer fetches from the leader
• If the fetch timeout is expired, the observer fetches from the bootstrap servers
• If the observer receives a fetch response from bootstrap servers with leader endpoints, the observer resets the fetch timeout

A voter has similar functionality where a fetch timeout expiration and a failed pre-vote election results in a reset of the fetch timer to the same leader in the same epoch.

Testing

Added unit test to KafkaRaftClientFetchTest to show fetches oscillate between the leader and bootstrap endpoints based on the fetch timer.