josefk31 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3358958295
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java:
##########
@@ -765,4 +766,75 @@ 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 otherVoter = KafkaRaftClientTest.replicaKey(local.id() + 2,
true);
+
+ final var voters = VoterSet.fromMap(
+ Map.of(
+ leader.id(), VoterSetTest.voterNode(leader),
+ otherVoter.id(), VoterSetTest.voterNode(otherVoter)
+ )
+ );
+
+ final var context = new RaftClientTestContext.Builder(
+ local.id(),
+ local.directoryId().get()
+ )
+ .withStaticVoters(voters)
+
.withBootstrapServers(Optional.of(List.of(RaftClientTestContext.mockAddress(otherVoter.id()))))
+
.withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1166_PROTOCOL)
+ .build();
+
+ for (int i = 0; i < 10; ++i) {
+ // The observer initially fetches from the bootstrap servers,
where it will discover the leader's endpoints.
+ context.pollUntilRequest();
+ final var bootstrapFetch = context.assertSentFetchRequest();
+ assertEquals(-2, bootstrapFetch.destination().id());
+
assertEquals(RaftClientTestContext.mockAddress(otherVoter.id()).getHostName(),
bootstrapFetch.destination().host());
+
assertEquals(RaftClientTestContext.mockAddress(otherVoter.id()).getPort(),
bootstrapFetch.destination().port());
+
+ 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
+ context.pollUntilRequest();
+ final var leaderFetch = context.assertSentFetchRequest();
+ assertEquals(leader.id(), leaderFetch.destination().id());
+
assertEquals(RaftClientTestContext.mockAddress(leader.id()).getHostName(),
leaderFetch.destination().host());
+
assertEquals(RaftClientTestContext.mockAddress(leader.id()).getPort(),
leaderFetch.destination().port());
+
+ // 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.
+
+ // 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.
+ context.deliverResponse(
Review Comment:
IIUC in the actual failure mode which causes this bug, we never get a
response from the leader at all. It's completely disconnected. is that the same
as receiving a `Errors.BROKER_NOT_AVAILABLE`? In this test case we actually
receive a response while in IRL we do not.
--
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]
