KAFKA-16540: enforce min.insync.replicas config invariants for ELR (apache#17952)

If ELR is enabled, we need to set a cluster-level min.insync.replicas, and remove all broker-level overrides. The reason for this is that if brokers disagree about which partitions are under min ISR, it breaks the KIP-966 replication invariants. In order to enforce this, when the eligible.leader.replicas.version feature is turned on, we automatically remove all broker-level min.insync.replicas overrides, and create the required cluster-level override if needed. Similarly, if the cluster was created with eligible.leader.replicas.version enabled, we create a similar cluster-level record. In both cases, we don't allow setting overrides for individual brokers afterwards, or removing the cluster-level override.

Split ActivationRecordsGeneratorTest up into multiple test cases rather than having it be one giant test case.

Fix a bug in QuorumControllerTestEnv where we would replay records manually on objects, racing with the active controller thread. Instead, we should simply ensure that the initial bootstrap records contains what we want.

Reviewers: Colin P. McCabe <[email protected]>

Author: CalvinConfluent

checkstyle/suppressions.xml

@@ -305,7 +305,7 @@
     <suppress checks="(ParameterNumber|ClassDataAbstractionCoupling)"
               files="(QuorumController).java"/>
     <suppress checks="(CyclomaticComplexity|NPathComplexity)"
-              files="(PartitionRegistration|PartitionChangeBuilder|ScramParser).java"/>
+              files="(ConfigurationControlManager|PartitionRegistration|PartitionChangeBuilder|ScramParser).java"/>
     <suppress checks="CyclomaticComplexity"
               files="(ClientQuotasImage|KafkaEventQueue|MetadataDelta|QuorumController|ReplicationControlManager|KRaftMigrationDriver|ClusterControlManager|MetaPropertiesEnsemble).java"/>
     <suppress checks="NPathComplexity"

metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java

@@ -17,25 +17,31 @@
 
 package org.apache.kafka.controller;
 
+import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.metadata.AbortTransactionRecord;
 import org.apache.kafka.common.metadata.BeginTransactionRecord;
+import org.apache.kafka.common.metadata.ConfigRecord;
 import org.apache.kafka.common.metadata.EndTransactionRecord;
 import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.EligibleLeaderReplicasVersion;
 import org.apache.kafka.server.common.MetadataVersion;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
 
+import static org.apache.kafka.common.config.ConfigResource.Type.BROKER;
+
 
 public class ActivationRecordsGenerator {
 
     static ControllerResult<Void> recordsForEmptyLog(
         Consumer<String> activationMessageConsumer,
         long transactionStartOffset,
         BootstrapMetadata bootstrapMetadata,
-        MetadataVersion metadataVersion
+        MetadataVersion metadataVersion,
+        int defaultMinInSyncReplicas
     ) {
         StringBuilder logMessageBuilder = new StringBuilder("Performing controller activation. ");
         List<ApiMessageAndVersion> records = new ArrayList<>();
@@ -87,6 +93,15 @@ static ControllerResult<Void> recordsForEmptyLog(
         // initialization, etc.
         records.addAll(bootstrapMetadata.records());
 
+        // If ELR is enabled, we need to set a cluster-level min.insync.replicas.
+        if (bootstrapMetadata.featureLevel(EligibleLeaderReplicasVersion.FEATURE_NAME) > 0) {
+            records.add(new ApiMessageAndVersion(new ConfigRecord().
+                setResourceType(BROKER.id()).
+                setResourceName("").
+                setName(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG).
+                setValue(Integer.toString(defaultMinInSyncReplicas)), (short) 0));
+        }
+
         activationMessageConsumer.accept(logMessageBuilder.toString().trim());
         if (metadataVersion.isMetadataTransactionSupported()) {
             records.add(new ApiMessageAndVersion(new EndTransactionRecord(), (short) 0));
@@ -148,13 +163,19 @@ static ControllerResult<Void> generate(
         boolean isEmpty,
         long transactionStartOffset,
         BootstrapMetadata bootstrapMetadata,
-        MetadataVersion curMetadataVersion
+        MetadataVersion curMetadataVersion,
+        int defaultMinInSyncReplicas
     ) {
         if (isEmpty) {
-            return recordsForEmptyLog(activationMessageConsumer, transactionStartOffset,
-                    bootstrapMetadata, bootstrapMetadata.metadataVersion());
+            return recordsForEmptyLog(activationMessageConsumer,
+                    transactionStartOffset,
+                    bootstrapMetadata,
+                    bootstrapMetadata.metadataVersion(),
+                    defaultMinInSyncReplicas);
         } else {
-            return recordsForNonEmptyLog(activationMessageConsumer, transactionStartOffset, curMetadataVersion);
+            return recordsForNonEmptyLog(activationMessageConsumer,
+                    transactionStartOffset,
+                    curMetadataVersion);
         }
     }
 }