KAFKA-13916; Fenced replicas should not be allowed to join the ISR in KRaft (KIP-841, Part 1)

core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala

@@ -17,15 +17,13 @@
 package kafka.server.metadata
 
 import java.util.concurrent.RejectedExecutionException
-
 import kafka.utils.Logging
 import org.apache.kafka.image.MetadataImage
 import org.apache.kafka.common.utils.{LogContext, Time}
 import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
 import org.apache.kafka.server.common.ApiMessageAndVersion
 import org.apache.kafka.snapshot.SnapshotWriter
 
-
 trait SnapshotWriterBuilder {
   def build(committedOffset: Long,
             committedEpoch: Int,

core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala

@@ -98,11 +98,11 @@ class BrokerMetadataListenerTest {
           assertEquals(200L, newImage.highestOffsetAndEpoch().offset)
           assertEquals(new BrokerRegistration(0, 100L,
             Uuid.fromString("GFBwlTcpQUuLYQ2ig05CSg"), Collections.emptyList[Endpoint](),
-            Collections.emptyMap[String, VersionRange](), Optional.empty[String](), false),
+            Collections.emptyMap[String, VersionRange](), Optional.empty[String](), false, false),
             delta.clusterDelta().broker(0))
           assertEquals(new BrokerRegistration(1, 200L,
             Uuid.fromString("QkOQtNKVTYatADcaJ28xDg"), Collections.emptyList[Endpoint](),
-            Collections.emptyMap[String, VersionRange](), Optional.empty[String](), true),
+            Collections.emptyMap[String, VersionRange](), Optional.empty[String](), true, false),
             delta.clusterDelta().broker(1))
         }
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.controller;
 
+import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.common.Endpoint;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.DuplicateBrokerRegistrationException;
@@ -39,13 +40,15 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.metadata.BrokerRegistration;
 import org.apache.kafka.metadata.BrokerRegistrationFencingChange;
+import org.apache.kafka.metadata.BrokerRegistrationInControlledShutdownChange;
 import org.apache.kafka.metadata.BrokerRegistrationReply;
 import org.apache.kafka.metadata.FinalizedControllerFeatures;
 import org.apache.kafka.metadata.VersionRange;
 import org.apache.kafka.metadata.placement.ReplicaPlacer;
 import org.apache.kafka.metadata.placement.StripedReplicaPlacer;
 import org.apache.kafka.metadata.placement.UsableBroker;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.timeline.SnapshotRegistry;
 import org.apache.kafka.timeline.TimelineHashMap;
 import org.slf4j.Logger;
@@ -64,8 +67,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static java.util.Collections.singletonList;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
-import static org.apache.kafka.common.metadata.MetadataRecordType.REGISTER_BROKER_RECORD;
 
 
 /**
@@ -84,6 +87,7 @@ static class Builder {
         private long sessionTimeoutNs = DEFAULT_SESSION_TIMEOUT_NS;
         private ReplicaPlacer replicaPlacer = null;
         private ControllerMetrics controllerMetrics = null;
+        private FeatureControlManager featureControl = null;
 
         Builder setLogContext(LogContext logContext) {
             this.logContext = logContext;
@@ -120,8 +124,15 @@ Builder setControllerMetrics(ControllerMetrics controllerMetrics) {
             return this;
         }
 
+        Builder setFeatureControlManager(FeatureControlManager featureControl) {
+            this.featureControl = featureControl;
+            return this;
+        }
+
         ClusterControlManager build() {
-            if (logContext == null) logContext = new LogContext();
+            if (logContext == null) {
+                logContext = new LogContext();
+            }
             if (clusterId == null) {
                 clusterId = Uuid.randomUuid().toString();
             }
@@ -132,15 +143,27 @@ ClusterControlManager build() {
                 replicaPlacer = new StripedReplicaPlacer(new Random());
             }
             if (controllerMetrics == null) {
-                throw new RuntimeException("You must specify controllerMetrics");
+                throw new RuntimeException("You must specify ControllerMetrics");
+            }
+            if (featureControl == null) {
+                featureControl = new FeatureControlManager.Builder().
+                    setLogContext(logContext).
+                    setSnapshotRegistry(snapshotRegistry).
+                    setQuorumFeatures(new QuorumFeatures(0, new ApiVersions(),
+                        QuorumFeatures.defaultFeatureMap(),
+                        singletonList(0))).
+                    setMetadataVersion(MetadataVersion.latest()).
+                    build();
             }
             return new ClusterControlManager(logContext,
                 clusterId,
                 time,
                 snapshotRegistry,
                 sessionTimeoutNs,
                 replicaPlacer,
-                controllerMetrics);
+                controllerMetrics,
+                featureControl
+            );
         }
     }
 
@@ -218,14 +241,20 @@ boolean check() {
      */
     private Optional<ReadyBrokersFuture> readyBrokersFuture;
 
+    /**
+     * The feature control manager.
+     */
+    private final FeatureControlManager featureControl;
+
     private ClusterControlManager(
         LogContext logContext,
         String clusterId,
         Time time,
         SnapshotRegistry snapshotRegistry,
         long sessionTimeoutNs,
         ReplicaPlacer replicaPlacer,
-        ControllerMetrics metrics
+        ControllerMetrics metrics,
+        FeatureControlManager featureControl
     ) {
         this.logContext = logContext;
         this.clusterId = clusterId;
@@ -237,6 +266,7 @@ private ClusterControlManager(
         this.heartbeatManager = null;
         this.readyBrokersFuture = Optional.empty();
         this.controllerMetrics = metrics;
+        this.featureControl = featureControl;
     }
 
     ReplicaPlacer replicaPlacer() {
@@ -279,6 +309,13 @@ Set<Integer> fencedBrokerIds() {
             .collect(Collectors.toSet());
     }
 
+    private short registerBrokerRecordVersion() {
+        if (featureControl.metadataVersion().isInControlledShutdownStateSupported())
+            return (short) 1;
+        else
+            return (short) 0;
+    }
+
     /**
      * Process an incoming broker registration request.
      */
@@ -339,7 +376,7 @@ public ControllerResult<BrokerRegistrationReply> registerBroker(
         heartbeatManager.register(brokerId, record.fenced());
 
         List<ApiMessageAndVersion> records = new ArrayList<>();
-        records.add(new ApiMessageAndVersion(record, REGISTER_BROKER_RECORD.highestSupportedVersion()));
+        records.add(new ApiMessageAndVersion(record, registerBrokerRecordVersion()));
         return ControllerResult.atomicOf(records, new BrokerRegistrationReply(brokerEpoch));
     }
 
@@ -361,7 +398,8 @@ public void replay(RegisterBrokerRecord record) {
         BrokerRegistration prevRegistration = brokerRegistrations.put(brokerId,
                 new BrokerRegistration(brokerId, record.brokerEpoch(),
                     record.incarnationId(), listeners, features,
-                    Optional.ofNullable(record.rack()), record.fenced()));
+                    Optional.ofNullable(record.rack()), record.fenced(),
+                    record.inControlledShutdown()));
         updateMetrics(prevRegistration, brokerRegistrations.get(brokerId));
         if (heartbeatManager != null) {
             if (prevRegistration != null) heartbeatManager.remove(brokerId);
@@ -395,12 +433,14 @@ public void replay(UnregisterBrokerRecord record) {
 
     public void replay(FenceBrokerRecord record) {
         replayRegistrationChange(record, record.id(), record.epoch(),
-            BrokerRegistrationFencingChange.UNFENCE);
+            BrokerRegistrationFencingChange.UNFENCE,
+            BrokerRegistrationInControlledShutdownChange.NONE);
     }
 
     public void replay(UnfenceBrokerRecord record) {
         replayRegistrationChange(record, record.id(), record.epoch(),
-            BrokerRegistrationFencingChange.FENCE);
+            BrokerRegistrationFencingChange.FENCE,
+            BrokerRegistrationInControlledShutdownChange.NONE);
     }
 
     public void replay(BrokerRegistrationChangeRecord record) {
@@ -410,15 +450,22 @@ public void replay(BrokerRegistrationChangeRecord record) {
             throw new RuntimeException(String.format("Unable to replay %s: unknown " +
                 "value for fenced field: %d", record.toString(), record.fenced()));
         }
+        Optional<BrokerRegistrationInControlledShutdownChange> inControlledShutdownChange =
+            BrokerRegistrationInControlledShutdownChange.fromValue(record.inControlledShutdown());
+        if (!inControlledShutdownChange.isPresent()) {
+            throw new RuntimeException(String.format("Unable to replay %s: unknown " +
+                "value for inControlledShutdown field: %d", record.toString(), record.inControlledShutdown()));
+        }
         replayRegistrationChange(record, record.brokerId(), record.brokerEpoch(),
-            fencingChange.get());
+            fencingChange.get(), inControlledShutdownChange.get());
     }
 
     private void replayRegistrationChange(
         ApiMessage record,
         int brokerId,
         long brokerEpoch,
-        BrokerRegistrationFencingChange fencingChange
+        BrokerRegistrationFencingChange fencingChange,
+        BrokerRegistrationInControlledShutdownChange inControlledShutdownChange
     ) {
         BrokerRegistration curRegistration = brokerRegistrations.get(brokerId);
         if (curRegistration == null) {
@@ -429,8 +476,10 @@ private void replayRegistrationChange(
                 "registration with that epoch found", record.toString()));
         } else {
             BrokerRegistration nextRegistration = curRegistration;
-            if (fencingChange != BrokerRegistrationFencingChange.NONE) {
-                nextRegistration = nextRegistration.cloneWithFencing(fencingChange.asBoolean().get());
+            if (fencingChange != BrokerRegistrationFencingChange.NONE
+                || inControlledShutdownChange != BrokerRegistrationInControlledShutdownChange.NONE) {
+                nextRegistration = nextRegistration.cloneWith(
+                    fencingChange.asBoolean(), inControlledShutdownChange.asBoolean());
             }
             if (!curRegistration.equals(nextRegistration)) {
                 brokerRegistrations.put(brokerId, nextRegistration);
@@ -491,6 +540,18 @@ public boolean unfenced(int brokerId) {
         return !registration.fenced();
     }
 
+    public boolean inControlledShutdown(int brokerId) {
+        BrokerRegistration registration = brokerRegistrations.get(brokerId);
+        if (registration == null) return false;
+        return registration.inControlledShutdown();
+    }
+
+    public boolean active(int brokerId) {
+        BrokerRegistration registration = brokerRegistrations.get(brokerId);
+        if (registration == null) return false;
+        return !registration.inControlledShutdown() && !registration.fenced();
+    }
+
     BrokerHeartbeatManager heartbeatManager() {
         if (heartbeatManager == null) {
             throw new RuntimeException("ClusterControlManager is not active.");
@@ -549,16 +610,18 @@ public List<ApiMessageAndVersion> next() {
                     setMaxSupportedVersion(featureEntry.getValue().max()).
                     setMinSupportedVersion(featureEntry.getValue().min()));
             }
-            List<ApiMessageAndVersion> batch = new ArrayList<>();
-            batch.add(new ApiMessageAndVersion(new RegisterBrokerRecord().
+            RegisterBrokerRecord record = new RegisterBrokerRecord().
                 setBrokerId(brokerId).
                 setIncarnationId(registration.incarnationId()).
                 setBrokerEpoch(registration.epoch()).
                 setEndPoints(endpoints).
                 setFeatures(features).
                 setRack(registration.rack().orElse(null)).
-                setFenced(registration.fenced()), REGISTER_BROKER_RECORD.highestSupportedVersion()));
-            return batch;
+                setFenced(registration.fenced());
+            if (featureControl.metadataVersion().isInControlledShutdownStateSupported()) {
+                record.setInControlledShutdown(registration.inControlledShutdown());
+            }
+            return singletonList(new ApiMessageAndVersion(record, registerBrokerRecordVersion()));
         }
     }
 

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

@@ -1557,6 +1557,11 @@ private QuorumController(LogContext logContext,
             setNodeId(nodeId).
             build();
         this.clientQuotaControlManager = new ClientQuotaControlManager(snapshotRegistry);
+        this.featureControl = new FeatureControlManager.Builder().
+                setLogContext(logContext).
+                setQuorumFeatures(quorumFeatures).
+                setSnapshotRegistry(snapshotRegistry).
+                build();
         this.clusterControl = new ClusterControlManager.Builder().
             setLogContext(logContext).
             setClusterId(clusterId).
@@ -1565,12 +1570,8 @@ private QuorumController(LogContext logContext,
             setSessionTimeoutNs(sessionTimeoutNs).
             setReplicaPlacer(replicaPlacer).
             setControllerMetrics(controllerMetrics).
+            setFeatureControlManager(featureControl).
             build();
-        this.featureControl = new FeatureControlManager.Builder().
-                setLogContext(logContext).
-                setQuorumFeatures(quorumFeatures).
-                setSnapshotRegistry(snapshotRegistry).
-                build();
         this.producerIdControlManager = new ProducerIdControlManager(clusterControl, snapshotRegistry);
         this.snapshotMaxNewRecordBytes = snapshotMaxNewRecordBytes;
         this.leaderImbalanceCheckIntervalNs = leaderImbalanceCheckIntervalNs;