kevin-wu24 commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3462729740
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java:
##########
@@ -1696,6 +1698,172 @@ void testUpdateVoter() throws Exception {
assertTrue(context.client.quorum().isVoter(follower));
}
+ @Test
+ void testUpdateVoterWithUnreachableListener() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower = replicaKey(local.id() + 1, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));
+
+ RaftClientTestContext context = new
RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower,
context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch,
OptionalInt.of(local.id()));
+
+ // Attempt to update voter with new listeners
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower,
+ Feature.KRAFT_VERSION.supportedVersionRange(),
+ voters.listeners(follower.id())
+ )
+ );
+
+ // Leader sends API_VERSIONS to verify reachability
+ context.pollUntilRequest();
+ context.assertSentApiVersionsRequest();
+
+ // Simulate unreachable voter by not responding and letting it timeout
+ context.time.sleep(context.requestTimeoutMs());
+ context.pollUntilResponse();
+
+ // UpdateVoter should fail since voter is unreachable
+ context.assertSentUpdateVoterResponse(
+ Errors.REQUEST_TIMED_OUT,
+ OptionalInt.of(local.id()),
+ epoch
+ );
+
+ // Voter set should not be updated
+ assertEquals(voters, context.listener.lastCommittedVoterSet().get());
+ }
+
+ @Test
+ void testUpdateVoterWithBrokerNotAvailable() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower = replicaKey(local.id() + 1, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));
+ InetSocketAddress defaultAddress =
InetSocketAddress.createUnresolved("localhost", 9990 + follower.id());
+
+ RaftClientTestContext context = new
RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower,
context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch,
OptionalInt.of(local.id()));
+
+ // Attempt to update voter with new listeners
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower,
+ Feature.KRAFT_VERSION.supportedVersionRange(),
+ voters.listeners(follower.id())
Review Comment:
This applies to the other tests too, but it would be nice if we can have the
`UPDATE_VOTER` request's endpoint for the follower be different than what is
created for it in the test via `VoterSetTestUtil#voterNode`.
Without looking at the implementation, we don't know whether assertions on
the destination endpoint of the `API_VERSIONS` request are passing because the
endpoint was supplied by the `UPDATE_VOTER` request body (actually correct), or
because the request was sent to the endpoint that already exists in the voter
set (this would be incorrect), since both endpoints are the same.
This is different from the add voter tests, since add voter's `API_VERSIONS`
loop test a node that is not part of the voter set yet.
--
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]
