add disabledFeatures for intergration tests

Author: mannoopj

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java

@@ -513,6 +513,9 @@ private void formatNode(
                         "bootstrap metadata. Use Formatter.setScramArguments() instead.");
                 }
             }
+            for (String disabledFeature : nodes.disabledFeatures()) {
+                formatter.setFeatureLevel(disabledFeature, (short) 0);
+            }
             formatter.setAdditionalBootstrapRecords(additionalRecords);
 
             StringBuilder dynamicVotersBuilder = new StringBuilder();

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java

@@ -36,6 +36,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
@@ -54,6 +55,7 @@ public static class Builder {
         private int numDisksPerBroker = 1;
         private Map<Integer, Map<String, String>> perServerProperties = Map.of();
         private BootstrapMetadata bootstrapMetadata;
+        private Set<String> disabledFeatures = Set.of();
 
         public Builder() {
             this(BootstrapMetadata.fromVersions(
@@ -93,6 +95,11 @@ public Builder setBootstrapMetadata(BootstrapMetadata bootstrapMetadata) {
             return this;
         }
 
+        public Builder setDisabledFeatures(Set<String> disabledFeatures) {
+            this.disabledFeatures = Collections.unmodifiableSet(disabledFeatures);
+            return this;
+        }
+
         public Builder setCombined(boolean combined) {
             this.combined = combined;
             return this;
@@ -215,14 +222,15 @@ public TestKitNodes build() {
                 brokerNodes.put(id, brokerNode);
             }
 
-            return new TestKitNodes(baseDirectory.toFile().getAbsolutePath(), clusterId, bootstrapMetadata, controllerNodes, brokerNodes,
-                brokerListenerName, brokerSecurityProtocol, controllerListenerName, controllerSecurityProtocol);
+            return new TestKitNodes(baseDirectory.toFile().getAbsolutePath(), clusterId, bootstrapMetadata, disabledFeatures, controllerNodes,
+                brokerNodes, brokerListenerName, brokerSecurityProtocol, controllerListenerName, controllerSecurityProtocol);
         }
     }
 
     private final String baseDirectory;
     private final String clusterId;
     private final BootstrapMetadata bootstrapMetadata;
+    private final Set<String> disabledFeatures;
     private final SortedMap<Integer, TestKitNode> controllerNodes;
     private final SortedMap<Integer, TestKitNode> brokerNodes;
     private final ListenerName brokerListenerName;
@@ -234,6 +242,7 @@ private TestKitNodes(
         String baseDirectory,
         String clusterId,
         BootstrapMetadata bootstrapMetadata,
+        Set<String> disabledFeatures,
         SortedMap<Integer, TestKitNode> controllerNodes,
         SortedMap<Integer, TestKitNode> brokerNodes,
         ListenerName brokerListenerName,
@@ -244,6 +253,7 @@ private TestKitNodes(
         this.baseDirectory = Objects.requireNonNull(baseDirectory);
         this.clusterId = Objects.requireNonNull(clusterId);
         this.bootstrapMetadata = Objects.requireNonNull(bootstrapMetadata);
+        this.disabledFeatures = Objects.requireNonNull(disabledFeatures);
         this.controllerNodes = Collections.unmodifiableSortedMap(new TreeMap<>(Objects.requireNonNull(controllerNodes)));
         this.brokerNodes = Collections.unmodifiableSortedMap(new TreeMap<>(Objects.requireNonNull(brokerNodes)));
         this.brokerListenerName = Objects.requireNonNull(brokerListenerName);
@@ -272,6 +282,10 @@ public BootstrapMetadata bootstrapMetadata() {
         return bootstrapMetadata;
     }
 
+    public Set<String> disabledFeatures() {
+        return disabledFeatures;
+    }
+
     public SortedMap<Integer, TestKitNode> brokerNodes() {
         return brokerNodes;
     }

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/junit/RaftClusterInvocationContext.java

@@ -336,8 +336,15 @@ public void format() throws Exception {
                     }
                 }
 
+                Set<String> disabledFeatures = newFeatureLevels.entrySet().stream()
+                    .filter(featureEntry -> featureEntry.getValue() == 0)
+                    .filter(featureEntry -> !featureEntry.getKey().equals(MetadataVersion.FEATURE_NAME))
+                    .map(Map.Entry::getKey)
+                    .collect(Collectors.toSet());
+
                 TestKitNodes nodes = new TestKitNodes.Builder()
                         .setBootstrapMetadata(BootstrapMetadata.fromVersions(clusterConfig.metadataVersion(), newFeatureLevels, "testkit"))
+                        .setDisabledFeatures(disabledFeatures)
                         .setCombined(isCombined)
                         .setNumBrokerNodes(clusterConfig.numBrokers())
                         .setNumDisksPerBroker(clusterConfig.numDisksPerBroker())