showuon commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3479906876
##########
server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java:
##########
@@ -205,6 +208,51 @@ private void testDescribeMetadataQuorum(ClusterInstance
clusterInstance, boolean
}
}
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoterByControllers(ClusterInstance
clusterInstance) throws Exception {
+ testAddRemoveRaftVoter(clusterInstance, true);
+ }
+
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoter(ClusterInstance clusterInstance) throws
Exception {
+ testAddRemoveRaftVoter(clusterInstance, false);
+ }
+
+ private void testAddRemoveRaftVoter(ClusterInstance clusterInstance,
boolean usingBootstrapControllers) throws Exception {
+ try (Admin admin = Admin.create(adminConfig(clusterInstance,
usingBootstrapControllers))) {
+ Set<Integer> initialVoters =
voterIds(admin.describeMetadataQuorum().quorumInfo().get());
+ AtomicInteger voterId = new AtomicInteger();
Review Comment:
Let's make the variable name clear. Maybe `voterIdToBeAddedAndRemoved`?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2276,39 +2282,79 @@ private CompletionStage<AddRaftVoterResponseData>
handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid
voter")
- );
- }
-
- Endpoints newVoterEndpoints =
Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint
(%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(),
currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data,
requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return
completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST,
e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
+ AddRaftVoterRequestData data,
+ short apiVersion,
+ long currentTimeMs
+ ) {
+ if (apiVersion >= 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID))
{
+ List<ReplicaKey> matchingObservers = quorum.leaderStateOrThrow()
+ .observerStates(currentTimeMs)
+ .keySet()
+ .stream()
+ .filter(key -> key.id() == data.voterId())
+ .toList();
+ if (matchingObservers.size() > 1) {
+ throw new IllegalArgumentException(
+ String.format("Multiple observers with node ID %d detected
(%s).", data.voterId(), matchingObservers)
+ );
+ } else if (matchingObservers.isEmpty() ||
matchingObservers.get(0).directoryId().isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Could not find a unique observer with node ID %d to
derive its directory ID. " +
+ "Ensure the node is running and fetching before
adding it as a voter.",
+ data.voterId()
+ )
+ );
+ } else {
+ return matchingObservers.get(0);
+ }
+ }
+
+ return RaftUtil.addVoterRequestVoterKey(data)
+ .filter(voterKey -> voterKey.directoryId().isPresent())
+ .orElseThrow(() -> new IllegalArgumentException("Add voter request
didn't include a valid voter"));
+ }
+
+ private Endpoints resolveAddVoterEndpoints(
+ AddRaftVoterRequestData data,
+ short apiVersion
+ ) {
+ Endpoints endpoints = apiVersion < 2 || !data.listeners().isEmpty() ?
+ Endpoints.fromAddVoterRequest(data.listeners()) :
+ nodeEndpointProvider.endpointsOf(data.voterId());
+
+ if (endpoints.address(channel.listenerName()).isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Add voter request didn't include the endpoint (%s) for
the default listener %s",
Review Comment:
Is the error message correct for request without listener?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2276,39 +2282,79 @@ private CompletionStage<AddRaftVoterResponseData>
handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid
voter")
- );
- }
-
- Endpoints newVoterEndpoints =
Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint
(%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(),
currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data,
requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return
completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST,
e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
+ AddRaftVoterRequestData data,
+ short apiVersion,
+ long currentTimeMs
+ ) {
+ if (apiVersion >= 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID))
{
+ List<ReplicaKey> matchingObservers = quorum.leaderStateOrThrow()
+ .observerStates(currentTimeMs)
+ .keySet()
+ .stream()
+ .filter(key -> key.id() == data.voterId())
+ .toList();
+ if (matchingObservers.size() > 1) {
+ throw new IllegalArgumentException(
+ String.format("Multiple observers with node ID %d detected
(%s).", data.voterId(), matchingObservers)
Review Comment:
I would add some more instruction for this. Ex: Please remove one of the
node and then try it again.
##########
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:
TBH, I've read this test twice, I still don't know why it will fail. The
test name `DoesNotReplaceExplicitDirectoryId` is also not clear to me. Could we
add comments on each test like the `testRemoveVoterDerivesDirectoryId` test
does to make it clear what we're testing in each test?
##########
tools/src/test/java/org/apache/kafka/tools/MetadataQuorumCommandTest.java:
##########
@@ -59,9 +63,9 @@ public void
testDescribeQuorumReplicationSuccessful(ClusterInstance cluster) thr
assertTrue(header.matches("NodeId\\s+DirectoryId\\s+LogEndOffset\\s+Lag\\s+LastFetchTimestamp\\s+LastCaughtUpTimestamp\\s+Status\\s+"));
- if (cluster.type() == Type.CO_KRAFT)
+ if (cluster.type() == Type.CO_KRAFT)
assertEquals(Math.max(cluster.config().numControllers(),
cluster.config().numBrokers()), data.size());
- else
+ else
assertEquals(cluster.config().numBrokers() +
cluster.config().numControllers(), data.size());
Review Comment:
nit: Let's revert unneeded empty spaces.
##########
tools/src/test/java/org/apache/kafka/tools/MetadataQuorumCommandTest.java:
##########
@@ -194,4 +198,60 @@ private static void assertHumanReadable(String output) {
assertTrue(lastCaughtUpTimestamp.contains("ms ago"));
assertTrue(lastCaughtUpTimestampValue.matches("\\d*"));
}
+
+ /**
+ * Verify that add-controller and remove-controller work using only the
+ * controller node ID (no directory ID or endpoints required from the
user).
+ */
+ @ClusterTest(types = {Type.KRAFT}, controllers = 2, standalone = true)
+ public void testAddAndRemoveControllerByIdSuccessful(ClusterInstance
cluster) throws Exception {
+ AtomicInteger controllerId = new AtomicInteger();
+ try (Admin admin = cluster.admin(Map.of(), true)) {
+ TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
+ QuorumInfo info =
admin.describeMetadataQuorum().quorumInfo().get();
+ controllerId.set(info.observers().stream()
+ .mapToInt(QuorumInfo.ReplicaState::replicaId)
+ .filter(cluster.controllerIds()::contains)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No controller
observer found in quorum info " + info)));
Review Comment:
Let's also assert the voter size is 1 before adding voters here.
##########
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);
Review Comment:
Since we expect the "error message" should be different in difference cases,
could we also verify the it? Same as below.
##########
tools/src/test/java/org/apache/kafka/tools/MetadataQuorumCommandTest.java:
##########
@@ -194,4 +198,60 @@ private static void assertHumanReadable(String output) {
assertTrue(lastCaughtUpTimestamp.contains("ms ago"));
assertTrue(lastCaughtUpTimestampValue.matches("\\d*"));
}
+
+ /**
+ * Verify that add-controller and remove-controller work using only the
+ * controller node ID (no directory ID or endpoints required from the
user).
+ */
+ @ClusterTest(types = {Type.KRAFT}, controllers = 2, standalone = true)
+ public void testAddAndRemoveControllerByIdSuccessful(ClusterInstance
cluster) throws Exception {
Review Comment:
Question: Could we addVoter with [node ID, directory ID] or [node ID,
endpoint] ? If [node ID, directory ID] or [node ID, endpoint] doesn't match
with each other, ex: wrong directory ID/endpoint to the node, what will happen?
Could we also add tests for them?
--
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]
