kevin-wu24 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3363291226
##########
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:
> is that the same as receiving a Errors.BROKER_NOT_AVAILABLE
Yes, it is the same. The specific case is when the local node cannot
establish a TCP connection with the leader endpoint. This causes the
destination node to be considered "disconnected" and the RPC is never sent over
the wire. If you trace through the `KafkaNetworkChannel` + network client code
this ends up "returning" a dummy RPC error response to the local raft client
with `Errors.BROKER_NOT_AVAILABLE`. This is what clears up the raft request
manager's `AWAITING_RESPONSE` state for the "connection."
In my opinion, there is less of a clear motivation to change the above
behavior compared to allowing the observer to behave more like a voter. What do
you think?
--
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]
