WIP

Author: frankvicky

Diff for: metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -89,7 +89,7 @@ static class Builder {
         private SnapshotRegistry snapshotRegistry = null;
         private long sessionTimeoutNs = DEFAULT_SESSION_TIMEOUT_NS;
         private ReplicaPlacer replicaPlacer = null;
-        private FeatureControlManager featureControl = null;
+        private LimitedFeatureControlManager featureControl = null;
         private BrokerShutdownHandler brokerShutdownHandler = null;
         private String interBrokerListenerName = "PLAINTEXT";
 
@@ -123,7 +123,7 @@ Builder setReplicaPlacer(ReplicaPlacer replicaPlacer) {
             return this;
         }
 
-        Builder setFeatureControlManager(FeatureControlManager featureControl) {
+        Builder setFeatureControlManager(LimitedFeatureControlManager featureControl) {
             this.featureControl = featureControl;
             return this;
         }
@@ -250,7 +250,7 @@ boolean check() {
     /**
      * The feature control manager.
      */
-    private final FeatureControlManager featureControl;
+    private final LimitedFeatureControlManager featureControl;
 
     private final BrokerShutdownHandler brokerShutdownHandler;
 
@@ -276,7 +276,7 @@ private ClusterControlManager(
         SnapshotRegistry snapshotRegistry,
         long sessionTimeoutNs,
         ReplicaPlacer replicaPlacer,
-        FeatureControlManager featureControl,
+        LimitedFeatureControlManager featureControl,
         BrokerShutdownHandler brokerShutdownHandler,
         String interBrokerListenerName
     ) {

Diff for: metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java

@@ -75,7 +75,7 @@ public class ConfigurationControlManager {
     private final TimelineHashSet<Integer> brokersWithConfigs;
     private final Map<String, Object> staticConfig;
     private final ConfigResource currentController;
-    private final FeatureControlManager featureControl;
+    private final LimitedFeatureControlManager featureControl;
 
     static class Builder {
         private LogContext logContext = null;
@@ -86,7 +86,7 @@ static class Builder {
         private ConfigurationValidator validator = ConfigurationValidator.NO_OP;
         private Map<String, Object> staticConfig = Map.of();
         private int nodeId = 0;
-        private FeatureControlManager featureControl = null;
+        private LimitedFeatureControlManager featureControl = null;
 
         Builder setLogContext(LogContext logContext) {
             this.logContext = logContext;
@@ -128,7 +128,7 @@ Builder setNodeId(int nodeId) {
             return this;
         }
 
-        Builder setFeatureControl(FeatureControlManager featureControl) {
+        Builder setFeatureControl(LimitedFeatureControlManager featureControl) {
             this.featureControl = featureControl;
             return this;
         }
@@ -156,14 +156,14 @@ ConfigurationControlManager build() {
     }
 
     private ConfigurationControlManager(LogContext logContext,
-            SnapshotRegistry snapshotRegistry,
-            KafkaConfigSchema configSchema,
-            Consumer<ConfigResource> existenceChecker,
-            Optional<AlterConfigPolicy> alterConfigPolicy,
-            ConfigurationValidator validator,
-            Map<String, Object> staticConfig,
-            int nodeId,
-            FeatureControlManager featureControl
+                                        SnapshotRegistry snapshotRegistry,
+                                        KafkaConfigSchema configSchema,
+                                        Consumer<ConfigResource> existenceChecker,
+                                        Optional<AlterConfigPolicy> alterConfigPolicy,
+                                        ConfigurationValidator validator,
+                                        Map<String, Object> staticConfig,
+                                        int nodeId,
+                                        LimitedFeatureControlManager featureControl
     ) {
         this.log = logContext.logger(ConfigurationControlManager.class);
         this.snapshotRegistry = snapshotRegistry;

Diff for: metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java

@@ -50,7 +50,7 @@
 import static org.apache.kafka.controller.QuorumController.MAX_RECORDS_PER_USER_OP;
 
 
-public class FeatureControlManager {
+public class FeatureControlManager implements LimitedFeatureControlManager {
     public static class Builder {
         private LogContext logContext = null;
         private SnapshotRegistry snapshotRegistry = null;
@@ -141,7 +141,8 @@ private FeatureControlManager(
         this.clusterSupportDescriber = clusterSupportDescriber;
     }
 
-    ControllerResult<ApiError> updateFeatures(
+    @Override
+    public ControllerResult<ApiError> updateFeatures(
         Map<String, Short> updates,
         Map<String, FeatureUpdate.UpgradeType> upgradeTypes,
         boolean validateOnly
@@ -168,11 +169,13 @@ ControllerResult<ApiError> updateFeatures(
         }
     }
 
-    Optional<MetadataVersion> metadataVersion() {
+    @Override
+    public Optional<MetadataVersion> metadataVersion() {
         return metadataVersion.get();
     }
 
-    MetadataVersion metadataVersionOrThrow() {
+    @Override
+    public MetadataVersion metadataVersionOrThrow() {
         return metadataVersionOrThrow(SnapshotRegistry.LATEST_EPOCH);
     }
 
@@ -361,6 +364,7 @@ FinalizedControllerFeatures finalizedFeatures(long epoch) {
         return new FinalizedControllerFeatures(features, epoch);
     }
 
+    @Override
     public void replay(FeatureLevelRecord record) {
         VersionRange range = quorumFeatures.localSupportedFeature(record.name());
         if (!range.contains(record.featureLevel())) {
@@ -387,11 +391,13 @@ public void replay(FeatureLevelRecord record) {
         }
     }
 
-    boolean isControllerId(int nodeId) {
+    @Override
+    public boolean isControllerId(int nodeId) {
         return quorumFeatures.isControllerId(nodeId);
     }
 
-    boolean isElrFeatureEnabled() {
+    @Override
+    public boolean isElrFeatureEnabled() {
         return finalizedVersions.getOrDefault(EligibleLeaderReplicasVersion.FEATURE_NAME, (short) 0) >=
             EligibleLeaderReplicasVersion.ELRV_1.featureLevel();
     }

Diff for: metadata/src/main/java/org/apache/kafka/controller/LimitedFeatureControlManager.java

@@ -0,0 +1,28 @@
+package org.apache.kafka.controller;
+
+import org.apache.kafka.clients.admin.FeatureUpdate;
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.requests.ApiError;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.Map;
+import java.util.Optional;
+
+public interface LimitedFeatureControlManager {
+
+    ControllerResult<ApiError> updateFeatures(
+        Map<String, Short> updates,
+        Map<String, FeatureUpdate.UpgradeType> upgradeTypes,
+        boolean validateOnly
+    );
+
+    Optional<MetadataVersion> metadataVersion();
+
+    MetadataVersion metadataVersionOrThrow();
+
+    void replay(FeatureLevelRecord record);
+
+    boolean isControllerId(int nodeId);
+
+    boolean isElrFeatureEnabled();
+}