Handling min.insync.replicas deletion operation when possible based on ELR version (#11857)

Signed-off-by: Paolo Patierno <ppatierno@live.com>

Author: ppatierno

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaBrokerConfigurationDiff.java

@@ -42,6 +42,7 @@ public class KafkaBrokerConfigurationDiff extends AbstractJsonDiff {
     private final Reconciliation reconciliation;
     private final Collection<AlterConfigOp> brokerConfigDiff;
     private final Map<String, ConfigModel> configModel;
+    private final short elrVersion;
 
     /**
      * These options are skipped because they contain placeholders
@@ -73,9 +74,11 @@ public class KafkaBrokerConfigurationDiff extends AbstractJsonDiff {
      * @param desired           Desired configuration
      * @param kafkaVersion      Kafka version
      * @param brokerNodeRef     Broker node reference
+     * @param elrVersion        If Eligible Leader Replicas (ELR) is enabled on the Kafka cluster
      */
-    protected KafkaBrokerConfigurationDiff(Reconciliation reconciliation, Config brokerConfigs, String desired, KafkaVersion kafkaVersion, NodeRef brokerNodeRef) {
+    protected KafkaBrokerConfigurationDiff(Reconciliation reconciliation, Config brokerConfigs, String desired, KafkaVersion kafkaVersion, NodeRef brokerNodeRef, short elrVersion) {
         this.reconciliation = reconciliation;
+        this.elrVersion = elrVersion;
         this.configModel = KafkaConfiguration.readConfigModel(kafkaVersion);
         this.brokerConfigDiff = diff(brokerNodeRef, desired, brokerConfigs, configModel);
     }
@@ -215,7 +218,9 @@ private void removeProperty(Map<String, ConfigModel> configModel, Collection<Alt
         } else {
             // entry is in current, is not in desired, is not default -> it was using non-default value and was removed
             // if the entry was custom, it should be deleted
-            if (!isIgnorableProperty(pathValueWithoutSlash, nodeIsController)) {
+            if ("min.insync.replicas".equals(entry.name()) && elrVersion >= 1) {
+                LOGGER.traceCr(reconciliation, "min.insync.replicas cannot be deleted when ELR is enabled");
+            } else if (!isIgnorableProperty(pathValueWithoutSlash, nodeIsController)) {
                 updatedCE.add(new AlterConfigOp(new ConfigEntry(pathValueWithoutSlash, null), AlterConfigOp.OpType.DELETE));
                 LOGGER.infoCr(reconciliation, "{} not set in desired, unsetting back to default {}", entry.name(), "deleted entry");
             } else {

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaRoller.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.AlterConfigsResult;
 import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.FeatureMetadata;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.config.ConfigException;
@@ -623,8 +624,12 @@ private void checkIfRestartOrReconfigureRequired(NodeRef nodeRef, boolean isCont
             // Always get the broker config. This request gets sent to that specific broker, so it's a proof that we can
             // connect to the broker and that it's capable of responding.
             Config brokerConfig;
+            short elrVersion = 0;
             try {
                 brokerConfig = brokerConfig(nodeRef);
+                FeatureMetadata featureMetadata = featureMetadata();
+                elrVersion = featureMetadata.finalizedFeatures().get("eligible.leader.replicas.version") == null ?
+                        0 : featureMetadata.finalizedFeatures().get("eligible.leader.replicas.version").maxVersionLevel();
             } catch (ForceableProblem e) {
                 if (restartContext.backOff.done()) {
                     needsRestart = true;
@@ -636,7 +641,7 @@ private void checkIfRestartOrReconfigureRequired(NodeRef nodeRef, boolean isCont
 
             if (!needsRestart && allowReconfiguration) {
                 LOGGER.traceCr(reconciliation, "Pod {}: description {}", nodeRef, brokerConfig);
-                brokerConfigDiff = new KafkaBrokerConfigurationDiff(reconciliation, brokerConfig, kafkaConfigProvider.apply(nodeRef.nodeId()), kafkaVersion, nodeRef);
+                brokerConfigDiff = new KafkaBrokerConfigurationDiff(reconciliation, brokerConfig, kafkaConfigProvider.apply(nodeRef.nodeId()), kafkaVersion, nodeRef, elrVersion);
 
                 if (brokerConfigDiff.getDiffSize() > 0) {
                     if (brokerConfigDiff.canBeUpdatedDynamically()) {
@@ -657,6 +662,17 @@ private void checkIfRestartOrReconfigureRequired(NodeRef nodeRef, boolean isCont
         restartContext.brokerConfigDiff = brokerConfigDiff;
     }
 
+    /**
+     * Returns the information about the features within the Kafka Cluster
+     * @return information about the features
+     */
+    /* test */ FeatureMetadata featureMetadata() throws ForceableProblem, InterruptedException {
+        return await(VertxUtil.kafkaFutureToVertxFuture(reconciliation, vertx, brokerAdminClient.describeFeatures().featureMetadata()),
+                30, TimeUnit.SECONDS,
+                error -> new ForceableProblem("Error getting feature metadata", error)
+        );
+    }
+
     /**
      * Returns a config of the given broker.
      * @param nodeRef The reference of the broker.