josefk31 commented on code in PR #22111:
URL: https://github.com/apache/kafka/pull/22111#discussion_r3358981037
##########
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:
Nit: do we with to add an assert for whether it becomes `Unattached`?
##########
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:
Nit: test would be clearer if we changed the variable name from `otherVoter`
to `bootstrapVoter`
##########
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:
We should add a test which explicitly checks that the observer can
transition to `unattached` if there is a timeout.
--
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]
