junrao commented on code in PR #22191:
URL: https://github.com/apache/kafka/pull/22191#discussion_r3678121550
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -610,6 +628,17 @@ public void replay(UnregisterBrokerRecord record) {
}
}
+ public void replay(UnregisterControllerRecord record) {
+ int controllerId = record.controllerId();
+ ControllerRegistration registration =
controllerRegistrations.remove(controllerId);
+ if (registration == null) {
+ throw new RuntimeException(String.format("Unable to replay %s: no
controller " +
Review Comment:
Finding from Claude.
UnregisterBrokerRecord has a brokerEpoch field, which is checked during
replay. UnregisterControllerRecord doesn't have a brokerEpoch field. Why the
inconsistency?
##########
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) {
+ throw new InvalidRequestException("Controller cannot
unregister itself while it is active.");
+ }
+ return clusterControl.unregisterController(controllerId);
Review Comment:
Finding from Claude.
Should we remove the controllerId from FeatureControlManager's
quorumFeatures.quorumNodeIds()? Otherwise, this unregistered controller can
block future MV upgrades.
##########
server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java:
##########
@@ -675,6 +678,53 @@ public void testUnregisterBroker(boolean
usingBootstrapControllers) throws Excep
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testUnregisterController(boolean usingBootstrapControllers)
throws Exception {
+ try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder()
+ .setNumBrokerNodes(3)
+ .setNumControllerNodes(3)
+ .build()).build()) {
+ cluster.format();
+ cluster.startup();
+ int controllerIdToUnregister =
cluster.controllers().keySet().iterator().next();
+ cluster.controllers().get(controllerIdToUnregister).shutdown();
+
+ try (Admin admin = createAdminClient(cluster,
usingBootstrapControllers)) {
Review Comment:
Do we guarantee that the active controller is elected when we get here?
##########
core/src/test/scala/unit/kafka/server/ControllerRegistrationManagerTest.scala:
##########
@@ -318,4 +340,86 @@ class ControllerRegistrationManagerTest {
manager.close()
}
}
+
+ @Test
+ def testReRegistrationAfterUnregister(): Unit = {
+ val context = new RegistrationTestContext(configProperties)
+ val manager = newControllerRegistrationManager(context)
+ try {
+ context.controllerNodeProvider.node.set(controller1)
+ manager.start(context.mockChannelManager)
+
+ // initial state with controller registered in log
+ val image = doMetadataUpdate(MetadataImage.EMPTY,
+ manager,
+ MetadataVersion.IBP_3_7_IV0,
Review Comment:
Why do we use an old MV? Ditto below.
##########
clients/src/main/resources/common/message/UnregisterControllerRequest.json:
##########
@@ -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.
+
+{
+ "apiKey": 94,
+ "type": "request",
+ "listeners": ["broker", "controller"],
+ "name": "UnregisterControllerRequest",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "ControllerId", "type": "int32", "versions": "0+",
Review Comment:
Should we add "entityType" following UnregisterBrokerRequest?
##########
clients/src/main/java/org/apache/kafka/clients/admin/Admin.java:
##########
@@ -1682,6 +1682,48 @@ 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>
Review Comment:
Should we add `</li> `for the remaining `<li>`?
##########
clients/src/main/java/org/apache/kafka/common/requests/UnregisterControllerResponse.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.common.requests;
+
+import org.apache.kafka.common.message.UnregisterControllerResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.Readable;
+
+import java.util.EnumMap;
+import java.util.Map;
+
+public class UnregisterControllerResponse extends AbstractResponse {
+ private final UnregisterControllerResponseData data;
+
+ public UnregisterControllerResponse(UnregisterControllerResponseData data)
{
+ super(ApiKeys.UNREGISTER_CONTROLLER);
+ this.data = data;
+ }
+
+ @Override
+ public UnregisterControllerResponseData data() {
+ return data;
+ }
+
+ @Override
+ public int throttleTimeMs() {
+ return data.throttleTimeMs();
+ }
+
+ @Override
+ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
+ data.setThrottleTimeMs(throttleTimeMs);
+ }
+
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
+ if (data.errorCode() != 0) {
+ errorCounts.put(Errors.forCode(data.errorCode()), 1);
+ }
+ return errorCounts;
+ }
+
+ public static UnregisterControllerResponse parse(Readable readable, short
version) {
+ return new UnregisterControllerResponse(new
UnregisterControllerResponseData(readable, version));
+ }
+
+ @Override
+ public boolean shouldClientThrottle(short version) {
+ return true;
Review Comment:
We don't need to override the method now that we merged
https://github.com/apache/kafka/pull/22908
--
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]
