jsancio commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3468321019
##########
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:
Good point. I updated the test that confirm a successful change to use a new
hostname for the default listener. This should check that kraft is sending the
RPC the new default listener's host and port. I don't think we need to check on
every single tests.
--
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]
