kevin-wu24 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3482185733
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,122 @@ void testUpdatedHighWatermarkCompleted() throws Exception
{
assertEquals(localLogEndOffset, partitionResponse.highWatermark());
}
}
+
+ @Test
+ void testObserverFetchesBetweenLeaderAndBootstrapServers() throws
Exception {
+ final var epoch = 2;
+ final var local = KafkaRaftClientTest.replicaKey(
+ KafkaRaftClientTest.randomReplicaId(),
+ true
+ );
+ final var leader = KafkaRaftClientTest.replicaKey(local.id() + 1,
true);
+ final var bootstrapVoter = KafkaRaftClientTest.replicaKey(local.id() +
2, true);
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ bootstrapVoter.id(), VoterSetTest.voterNode(bootstrapVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
+ // configure the bootstrap servers to only include the bootstrap
voter
+ // to reliably check the destination of the observer's fetch
requests
+ // alternates between the leader and the bootstrap voter
+ .withBootstrapServers(
+
Optional.of(List.of(RaftClientTestContext.mockAddress(bootstrapVoter.id())))
+ )
+
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
+ .build();
+
+ // The observer initially fetches from the bootstrap servers,
+ // where it will discover the leader's endpoints.
+ final var bootstrapFetch = pollAndCheckObserverFetchRequest(
+ context,
+ true,
+ bootstrapVoter.id()
+ );
+ context.deliverResponse(
+ bootstrapFetch.correlationId(),
+ bootstrapFetch.destination(),
+ context.fetchResponse(
+ epoch,
+ leader.id(),
+ MemoryRecords.EMPTY,
+ 0L,
+ Errors.NOT_LEADER_OR_FOLLOWER
+ )
+ );
+
+ // Subsequent fetch from the observer is sent to the leader
+ // Return a BROKER_NOT_AVAILABLE error, and then advance time past the
fetch timeout,
+ // which should cause the observer to fetch from the bootstrap servers
on the next fetch.
+ final var leaderFetch = pollAndCheckObserverFetchRequest(
+ context,
+ false,
+ leader.id()
+ );
+ context.deliverResponse(
+ leaderFetch.correlationId(),
+ leaderFetch.destination(),
+ RaftUtil.errorResponse(
+ ApiKeys.FETCH,
+ Errors.BROKER_NOT_AVAILABLE
+ )
+ );
+
+ // The fetch timeout is much greater than the request manager's
configured backoff, so the
+ // current unreachable connection will no longer be backing off when
the next fetch is sent.
+ // Expire the fetch timeout and check that the next fetch is sent to
the bootstrap server again.
+ context.time.sleep(context.fetchTimeoutMs + 1);
Review Comment:
My test is written incorrectly. I agree with your above comments that the
response from the leader fetch is not delivered until after the fetch timeout
expires.
I need to handle the `BROKER_NOT_AVAILABLE` response first via `poll()` to
accurately simulate this scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
