kevin-wu24 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3414472433
##########
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());
Review Comment:
Fixed.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2011,7 +2011,7 @@ public void
testFollowerAsObserverDoesNotBecomeProspectiveAfterFetchTimeout(bool
context.time.sleep(context.fetchTimeoutMs);
context.pollUntilRequest();
- assertTrue(context.client.quorum().isFollower());
+ assertFalse(context.client.quorum().isProspective());
Review Comment:
Added.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3390,6 +3423,9 @@ private long pollFollowerAsObserver(FollowerState state,
long currentTimeMs) {
state.resetUpdateVoterSetPeriod(currentTimeMs);
}
return sendResult.timeToWaitMs();
+ } else if (state.hasFetchTimeoutExpired(currentTimeMs)) {
+ transitionToUnattached(state.epoch(),
OptionalInt.of(state.leaderId()));
Review Comment:
I added this to the existing unit test.
--
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]
