junrao commented on code in PR #22191:
URL: https://github.com/apache/kafka/pull/22191#discussion_r3633679511
##########
tools/src/main/java/org/apache/kafka/tools/ClusterTool.java:
##########
@@ -104,6 +106,11 @@ static void execute(String... args) throws Exception {
.action(store())
.required(true)
.help("The ID of the broker to unregister.");
+ unregisterControllerParser.addArgument("--id", "-i")
Review Comment:
The KIP uses --controller-id. Should we update the KIP?
##########
clients/src/main/java/org/apache/kafka/clients/admin/UnregisterControllerOptions.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+
+/**
+ * Options for {@link Admin#unregisterController(int,
UnregisterControllerOptions)}.
+ */
[email protected]
+public class UnregisterControllerOptions extends
AbstractOptions<UnregisterControllerOptions> {
Review Comment:
This is not defined in the KIP. Could you add it there?
##########
docs/operations/kraft.md:
##########
@@ -232,6 +232,26 @@ When using controller endpoints use the
--bootstrap-controller flag:
$ bin/kafka-metadata-quorum.sh --bootstrap-controller localhost:9093
remove-controller --controller-id <id> --controller-directory-id <directory-id>
```
+To also unregister the controller from the cluster metadata after it has been
removed from the voter set, pass the `--unregister` flag:
Review Comment:
This is a bit confusing. It reads as if one can call
kafka-metadata-quorum.sh first to remove the node from the voter set and then
call it again to unregister it.
##########
tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java:
##########
@@ -494,12 +503,55 @@ static void handleRemoveController(
throw new TerseException("Failed to parse
--controller-directory-id: " + e.getMessage());
}
if (!dryRun) {
- admin.removeRaftVoter(controllerId, directoryId).
- all().get();
+ removeRaftVoter(admin, controllerId, directoryId, unregister);
}
- System.out.printf("%s KRaft controller %d with directory id %s%n",
+ System.out.printf("%sKRaft controller %d with directory id %s%n",
dryRun ? "DRY RUN of removing " : "Removed ",
controllerId,
directoryId);
+ if (unregister) {
+ if (!dryRun) {
+ unregisterController(admin, controllerId);
+ }
+ System.out.printf("%sKRaft controller %d%n",
+ dryRun ? "DRY RUN of unregistering " : "Unregistered ",
+ controllerId);
+ }
+ }
+
+ private static void removeRaftVoter(
+ Admin admin,
+ int controllerId,
+ Uuid directoryId,
+ boolean unregister
+ ) throws TerseException, ExecutionException, InterruptedException {
+ try {
+ admin.removeRaftVoter(controllerId, directoryId).all().get();
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (unregister && (cause instanceof UnsupportedVersionException ||
+ cause instanceof VoterNotFoundException)) {
+ throw new TerseException("Failed to remove KRaft voter " +
controllerId
+ + ": " + cause.getMessage()
+ + ". To unregister the controller from the cluster, run "
+ + "`kafka-cluster.sh unregister-controller --id "
+ + controllerId + "`.");
+ }
+ throw e;
+ }
+ }
+
+ private static void unregisterController(Admin admin, int controllerId)
+ throws TerseException, InterruptedException {
+ try {
+ admin.unregisterController(controllerId).all().get();
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ throw new TerseException("Failed to unregister controller " +
controllerId
+ + ": " + (cause != null ? cause.getMessage() : e.getMessage())
+ + ". To unregister the controller from the cluster, run "
+ + "`kafka-cluster.sh unregister-controller --id "
Review Comment:
Hmm, if the unregister fails here, will it ever be successful using
`kafka-cluster.sh unregister-controller`?
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -610,6 +628,18 @@ public void replay(UnregisterBrokerRecord record) {
}
}
+ public void replay(UnregisterControllerRecord record) {
+ int controllerId = record.controllerId();
+ ControllerRegistration registration =
controllerRegistrations.get(controllerId);
Review Comment:
We can just call `remove()` here since it returns the previous value if
present.
##########
clients/src/main/java/org/apache/kafka/clients/admin/Admin.java:
##########
@@ -1682,6 +1682,41 @@ default UnregisterBrokerResult unregisterBroker(int
brokerId) {
@InterfaceStability.Unstable
UnregisterBrokerResult unregisterBroker(int brokerId,
UnregisterBrokerOptions options);
+ /**
+ * Unregister a controller.
+ *
+ * This is a convenience method for {@link #unregisterController(int,
UnregisterControllerOptions)}
+ *
+ * @param controllerId the controller id to unregister.
+ *
+ * @return the {@link UnregisterControllerResult} containing the result
+ */
+ @InterfaceStability.Unstable
+ default UnregisterControllerResult unregisterController(int controllerId) {
+ return unregisterController(controllerId, new
UnregisterControllerOptions());
+ }
+
+ /**
+ * Unregister a controller.
+ *
+ * The following exceptions can be anticipated when calling {@code get()}
on the future from the
+ * returned {@link UnregisterControllerResult}:
+ * <ul>
+ * <li>{@link org.apache.kafka.common.errors.TimeoutException}
+ * If the request timed out before the unregister operation could
finish.</li>
+ * <li>{@link org.apache.kafka.common.errors.UnsupportedVersionException}
+ * If the software is too old to support the unregistration API.
Review Comment:
The KIP also includes NOT_CONTROLLER and INVALID_REQUEST.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -2153,6 +2157,21 @@ public CompletableFuture<Void> registerController(
EnumSet.noneOf(ControllerOperationFlag.class));
}
+ @Override
+ public CompletableFuture<Void> unregisterController(
+ ControllerRequestContext context,
+ int controllerId
+ ) {
+ return appendWriteEvent("unregisterController", context.deadlineNs(),
+ () -> {
+ if (nodeId == controllerId && isActiveController()) {
Review Comment:
`isActiveController()` seems redundant since this code runs under
ControllerWriteEvent.run() that checks activeController already.
##########
tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java:
##########
@@ -494,12 +503,55 @@ static void handleRemoveController(
throw new TerseException("Failed to parse
--controller-directory-id: " + e.getMessage());
}
if (!dryRun) {
- admin.removeRaftVoter(controllerId, directoryId).
- all().get();
+ removeRaftVoter(admin, controllerId, directoryId, unregister);
}
- System.out.printf("%s KRaft controller %d with directory id %s%n",
+ System.out.printf("%sKRaft controller %d with directory id %s%n",
dryRun ? "DRY RUN of removing " : "Removed ",
controllerId,
directoryId);
+ if (unregister) {
+ if (!dryRun) {
+ unregisterController(admin, controllerId);
+ }
+ System.out.printf("%sKRaft controller %d%n",
+ dryRun ? "DRY RUN of unregistering " : "Unregistered ",
+ controllerId);
+ }
+ }
+
+ private static void removeRaftVoter(
+ Admin admin,
+ int controllerId,
+ Uuid directoryId,
+ boolean unregister
+ ) throws TerseException, ExecutionException, InterruptedException {
+ try {
+ admin.removeRaftVoter(controllerId, directoryId).all().get();
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (unregister && (cause instanceof UnsupportedVersionException ||
+ cause instanceof VoterNotFoundException)) {
+ throw new TerseException("Failed to remove KRaft voter " +
controllerId
+ + ": " + cause.getMessage()
+ + ". To unregister the controller from the cluster, run "
+ + "`kafka-cluster.sh unregister-controller --id "
+ + controllerId + "`.");
+ }
+ throw e;
+ }
+ }
+
+ private static void unregisterController(Admin admin, int controllerId)
+ throws TerseException, InterruptedException {
+ try {
+ admin.unregisterController(controllerId).all().get();
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ throw new TerseException("Failed to unregister controller " +
controllerId
+ + ": " + (cause != null ? cause.getMessage() : e.getMessage())
Review Comment:
We check for null here, but not in line 535. Could we be consistent?
--
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]
