brandboat commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3488048447
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java:
##########
@@ -364,6 +364,170 @@ public void testAddVoter() throws Exception {
context.assertSentAddVoterResponse(Errors.NONE);
}
+ @Test
+ public void testAddVoterDerivesDirectoryIdAndEndpoints() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+ InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
+ "localhost",
+ 9990 + newVoter.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), newAddress)
+ );
+ context.client.setNodeEndpointProvider(nodeId ->
+ nodeId == newVoter.id() ? newListeners : Endpoints.empty()
+ );
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ completeApiVersionsForAddVoter(context, newVoter, newAddress);
+
+ // Handle the API_VERSIONS response
+ context.client.poll();
+ // Append new VotersRecord to log
+ context.client.poll();
+
+ commitNewVoterSetForAddVoter(context, local, follower, newVoter,
epoch);
+
+ // Expect reply for AddVoter request
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.NONE);
+ }
+
+ @Test
+ public void testAddVoterFailsToDeriveDefaultListenerEndpoint() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.INVALID_REQUEST);
+ }
+
+ @Test
+ public void testAddVoterDoesNotReplaceExplicitDirectoryId() throws
Exception {
Review Comment:
In AddVoterRequest v1, if the voter ID is valid but the directory UUID is
incorrect, we return REQUEST_TIMEOUT. This test verifies that v2 preserves the
same behavior as v1.
That said, I'm not sure we actually need to keep this behavior in v2. It
might be perfectly reasonable to throw an IllegalArgumentException when a user
tries to add a voter that doesn't exist in the quorum's observerStates. That
behavior would be much easier to understand and would make it clearer to users
what went wrong.
--
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]
