brandboat commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3480209815
##########
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:
I'll add more comments in the test, hope it could be helpful.
##########
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:
I'll add more comments in the test, hope it helps
--
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]
