Re: [PR] KAFKA-20395: Support unregistering controllers [kafka]

["josefk31 (via GitHub)" <gi...@apache.org>]

josefk31 commented on code in PR #22191:
URL: https://github.com/apache/kafka/pull/22191#discussion_r3335734708


##########
metadata/src/main/java/org/apache/kafka/controller/Controller.java:
##########
@@ -412,6 +412,20 @@ CompletableFuture<Void> registerController(
         ControllerRegistrationRequestData request
     );
 
+    /**
+     * Attempt to unregister the given controller.
+     *
+     * @param context       The controller request context.
+     * @param controllerId  The controller id to unregister.
+     *
+     * @return              A future that is completed successfully when the controller is
+     *                      unregistered.
+     */

Review Comment:
   Nit: there are two expected failure modes which we validate for:
   
   1. Don't unregister active controller
   2. Don't unregister controller which doesn't exist (controllerId not found).
   
   Should we document that implementors are expected to throw exceptions for these cases? 



##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -498,6 +500,22 @@ ControllerResult<Void> registerController(ControllerRegistrationRequestData requ
         return ControllerResult.atomicOf(records, null);
     }
 
+    ControllerResult<Void> unregisterController(int controllerId) {
+        if (!featureControl.metadataVersionOrThrow().isControllerUnregistrationSupported()) {
+            throw new UnsupportedVersionException("The current MetadataVersion is too old to " +
+                    "support controller unregistration.");
+        }
+        if (controllerRegistrations.get(controllerId) == null) {
+            throw new ControllerIdNotRegisteredException("Controller ID " + controllerId +
+                " is not currently registered.");
+        }
+        List<ApiMessageAndVersion> records = new ArrayList<>();
+        records.add(new ApiMessageAndVersion(new UnregisterControllerRecord().
+            setControllerId(controllerId),
+                (short) 0));

Review Comment:
   Nit - the indentation is a bit confusing about what function each object belongs to. Might be a bit clearer if
   
   ```
   new ApiMessageAndVersion(
        new UnregisterControllerRecord().setControllerId(controllerId),
        (short) 0));
   ```
   



##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -2128,6 +2132,19 @@ public CompletableFuture<Void> registerController(
             EnumSet.noneOf(ControllerOperationFlag.class));
     }
 
+    @Override
+    public CompletableFuture<Void> unregisterController(
+        ControllerRequestContext context,
+        int controllerId
+    ) {
+        if (nodeId == controllerId && isActiveController()) {

Review Comment:
   Nit: is the `isActiveController()` check necessary? `ControllerWriteEvents` can only be run on the active controller. In other words we probably always want to reject an attempt to unregister the current controller (whatever is `nodeId`). 



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org