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

clients/src/main/java/org/apache/kafka/common/errors/IneligibleReplica.java

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class IneligibleReplica extends ApiException {
+    public IneligibleReplica(String message) {
+        super(message);
+    }
+}

clients/src/main/java/org/apache/kafka/common/errors/NewLeaderElected.java

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class NewLeaderElected extends ApiException {
+    public NewLeaderElected(String message) {
+        super(message);
+    }
+}

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -52,6 +52,7 @@
 import org.apache.kafka.common.errors.InconsistentTopicIdException;
 import org.apache.kafka.common.errors.InconsistentVoterSetException;
 import org.apache.kafka.common.errors.InconsistentClusterIdException;
+import org.apache.kafka.common.errors.IneligibleReplica;
 import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException;
 import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.common.errors.InvalidFetchSessionEpochException;
@@ -77,6 +78,7 @@
 import org.apache.kafka.common.errors.LogDirNotFoundException;
 import org.apache.kafka.common.errors.MemberIdRequiredException;
 import org.apache.kafka.common.errors.NetworkException;
+import org.apache.kafka.common.errors.NewLeaderElected;
 import org.apache.kafka.common.errors.NoReassignmentInProgressException;
 import org.apache.kafka.common.errors.NotControllerException;
 import org.apache.kafka.common.errors.NotCoordinatorException;
@@ -364,7 +366,9 @@ public enum Errors {
     INCONSISTENT_TOPIC_ID(103, "The log's topic ID did not match the topic ID in the request", InconsistentTopicIdException::new),
     INCONSISTENT_CLUSTER_ID(104, "The clusterId in the request does not match that found on the server", InconsistentClusterIdException::new),
     TRANSACTIONAL_ID_NOT_FOUND(105, "The transactionalId could not be found", TransactionalIdNotFoundException::new),
-    FETCH_SESSION_TOPIC_ID_ERROR(106, "The fetch session encountered inconsistent topic ID usage", FetchSessionTopicIdException::new);
+    FETCH_SESSION_TOPIC_ID_ERROR(106, "The fetch session encountered inconsistent topic ID usage", FetchSessionTopicIdException::new),
+    INELIGIBLE_REPLICA(107, "The new ISR contains at least one ineligible replica.", IneligibleReplica::new),
+    NEW_LEADER_ELECTED(108, "The AlterPartition request successfully updated the partition state but the leader has changed.", NewLeaderElected::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java

@@ -45,8 +45,8 @@ public AlterPartitionRequestData data() {
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         return new AlterPartitionResponse(new AlterPartitionResponseData()
-                .setThrottleTimeMs(throttleTimeMs)
-                .setErrorCode(Errors.forException(e).code()));
+            .setThrottleTimeMs(throttleTimeMs)
+            .setErrorCode(Errors.forException(e).code()));
     }
 
     public static AlterPartitionRequest parse(ByteBuffer buffer, short version) {
@@ -57,8 +57,13 @@ public static class Builder extends AbstractRequest.Builder<AlterPartitionReques
 
         private final AlterPartitionRequestData data;
 
-        public Builder(AlterPartitionRequestData data) {
-            super(ApiKeys.ALTER_PARTITION);
+        public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {
+            super(
+                ApiKeys.ALTER_PARTITION,
+                ApiKeys.ALTER_PARTITION.oldestVersion(),
+                // Version 1 is the maximum version that can be used without topic ids.
+                canUseTopicIds ? ApiKeys.ALTER_PARTITION.latestVersion() : 1
+            );
             this.data = data;
         }
 

clients/src/main/resources/common/message/AlterPartitionRequest.json

@@ -18,16 +18,21 @@
   "type": "request",
   "listeners": ["zkBroker", "controller"],
   "name": "AlterPartitionRequest",
-  "validVersions": "0-1",
+  // Version 1 adds LeaderRecoveryState field (KIP-704).
+  //
+  // Version 2 adds TopicId field to replace TopicName field (KIP-841).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
       "about": "The ID of the requesting broker" },
     { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
       "about": "The epoch of the requesting broker" },
     { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
-      { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+      { "name":  "TopicName", "type": "string", "versions": "0-1", "ignorable": true, "entityType": "topicName",
         "about": "The name of the topic to alter ISRs for" },
+      { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
+        "about": "The ID of the topic to alter ISRs for" },
       { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
         { "name": "PartitionIndex", "type": "int32", "versions": "0+",
           "about": "The partition index" },

clients/src/main/resources/common/message/AlterPartitionResponse.json

@@ -17,16 +17,23 @@
   "apiKey": 56,
   "type": "response",
   "name": "AlterPartitionResponse",
-  "validVersions": "0-1",
+  // Version 1 adds LeaderRecoveryState field (KIP-704).
+  //
+  // Version 2 adds TopicId field to replace TopicName field, can return INELIGIBLE_REPLICA error
+  // when any replicas in the new ISR is fenced or in controlled shutdown, and can return
+  // NEW_LEADER_ELECTED error when the a new leader was elected after the partition state change (KIP-841).
+  "validVersions": "0-2",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name": "ErrorCode", "type": "int16", "versions": "0+",
       "about": "The top level response error code" },
     { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
-      { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+      { "name": "TopicName", "type": "string", "versions": "0-1", "ignorable": true, "entityType": "topicName",
         "about": "The name of the topic" },
+      { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
+        "about": "The ID of the topic" },
       { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
         { "name": "PartitionIndex", "type": "int32", "versions": "0+",
           "about": "The partition index" },

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

@@ -1321,9 +1321,10 @@ private AlterPartitionRequest createAlterPartitionRequest(short version) {
             .setBrokerEpoch(123L)
             .setBrokerId(1)
             .setTopics(singletonList(new AlterPartitionRequestData.TopicData()
-                .setName("topic1")
+                .setTopicName("topic1")
+                .setTopicId(Uuid.randomUuid())
                 .setPartitions(singletonList(partitionData))));
-        return new AlterPartitionRequest.Builder(data).build(version);
+        return new AlterPartitionRequest.Builder(data, version >= 1).build(version);
     }
 
     private AlterPartitionResponse createAlterPartitionResponse(int version) {
@@ -1343,8 +1344,9 @@ private AlterPartitionResponse createAlterPartitionResponse(int version) {
                 .setErrorCode(Errors.NONE.code())
                 .setThrottleTimeMs(123)
                 .setTopics(singletonList(new AlterPartitionResponseData.TopicData()
-                        .setName("topic1")
-                        .setPartitions(singletonList(partitionData))));
+                    .setTopicName("topic1")
+                    .setTopicId(Uuid.randomUuid())
+                    .setPartitions(singletonList(partitionData))));
         return new AlterPartitionResponse(data);
     }
 

core/src/main/scala/kafka/cluster/Partition.scala

@@ -19,7 +19,6 @@ package kafka.cluster
 import java.util.concurrent.locks.ReentrantReadWriteLock
 import java.util.Optional
 import java.util.concurrent.CompletableFuture
-
 import kafka.api.LeaderAndIsr
 import kafka.common.UnexpectedAppendOffsetException
 import kafka.controller.{KafkaController, StateChangeLogger}
@@ -30,6 +29,7 @@ import kafka.server.checkpoints.OffsetCheckpoints
 import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
 import kafka.utils._
 import kafka.zookeeper.ZooKeeperClientException
+import org.apache.kafka.common.TopicIdPartition
 import org.apache.kafka.common.errors._
 import org.apache.kafka.common.message.{DescribeProducersResponseData, FetchResponseData}
 import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
@@ -159,6 +159,7 @@ sealed trait PartitionState {
 }
 
 sealed trait PendingPartitionChange extends PartitionState {
+  def partitionStateToRollBackTo: PartitionState
   def sentLeaderAndIsr: LeaderAndIsr
 
   override val leaderRecoveryState: LeaderRecoveryState = LeaderRecoveryState.RECOVERED
@@ -167,7 +168,8 @@ sealed trait PendingPartitionChange extends PartitionState {
 case class PendingExpandIsr(
   isr: Set[Int],
   newInSyncReplicaId: Int,
-  sentLeaderAndIsr: LeaderAndIsr
+  sentLeaderAndIsr: LeaderAndIsr,
+  partitionStateToRollBackTo: PartitionState
 ) extends PendingPartitionChange {
   val maximalIsr = isr + newInSyncReplicaId
   val isInflight = true
@@ -177,14 +179,16 @@ case class PendingExpandIsr(
     s", newInSyncReplicaId=$newInSyncReplicaId" +
     s", sentLeaderAndIsr=$sentLeaderAndIsr" +
     s", leaderRecoveryState=$leaderRecoveryState" +
+    s", partitionStateToRollBackTo=$partitionStateToRollBackTo" +
     ")"
   }
 }
 
 case class PendingShrinkIsr(
   isr: Set[Int],
   outOfSyncReplicaIds: Set[Int],
-  sentLeaderAndIsr: LeaderAndIsr
+  sentLeaderAndIsr: LeaderAndIsr,
+  partitionStateToRollBackTo: PartitionState
 ) extends PendingPartitionChange  {
   val maximalIsr = isr
   val isInflight = true
@@ -194,13 +198,14 @@ case class PendingShrinkIsr(
     s", outOfSyncReplicaIds=$outOfSyncReplicaIds" +
     s", sentLeaderAndIsr=$sentLeaderAndIsr" +
     s", leaderRecoveryState=$leaderRecoveryState" +
+    s", partitionStateToRollBackTo=$partitionStateToRollBackTo" +
     ")"
   }
 }
 
 case class CommittedPartitionState(
   isr: Set[Int],
-  override val leaderRecoveryState: LeaderRecoveryState
+  leaderRecoveryState: LeaderRecoveryState
 ) extends PartitionState {
   val maximalIsr = isr
   val isInflight = false
@@ -847,21 +852,29 @@ class Partition(val topicPartition: TopicPartition,
   }
 
   private def needsExpandIsr(followerReplica: Replica): Boolean = {
-    canAddReplicaToIsr(followerReplica.brokerId) && isFollowerAtHighwatermark(followerReplica)
+    canAddReplicaToIsr(followerReplica.brokerId) && isFollowerInSync(followerReplica)
   }
 
   private def canAddReplicaToIsr(followerReplicaId: Int): Boolean = {
     val current = partitionState
-    !current.isInflight && !current.isr.contains(followerReplicaId)
+    !current.isInflight &&
+      !current.isr.contains(followerReplicaId) &&
+      isBrokerIsrEligible(followerReplicaId)
   }
 
-  private def isFollowerAtHighwatermark(followerReplica: Replica): Boolean = {
+  private def isFollowerInSync(followerReplica: Replica): Boolean = {
     leaderLogIfLocal.exists { leaderLog =>
       val followerEndOffset = followerReplica.stateSnapshot.logEndOffset
       followerEndOffset >= leaderLog.highWatermark && leaderEpochStartOffsetOpt.exists(followerEndOffset >= _)
     }
   }
 
+  private def isBrokerIsrEligible(brokerId: Int): Boolean = {
+    // In KRaft mode, only replicas which are not fenced nor in controlled shutdown are
+    // allowed to join the ISR. This does not apply to ZK mode.
+    !metadataCache.isBrokerFenced(brokerId) && !metadataCache.isBrokerInControlledShutdown(brokerId)
+  }
+
   /*
    * Returns a tuple where the first element is a boolean indicating whether enough replicas reached `requiredOffset`
    * and the second element is an error (which would be `Errors.NONE` for no error).
@@ -1503,8 +1516,19 @@ class Partition(val topicPartition: TopicPartition,
     // Alternatively, if the update fails, no harm is done since the expanded ISR puts
     // a stricter requirement for advancement of the HW.
     val isrToSend = partitionState.isr + newInSyncReplicaId
-    val newLeaderAndIsr = LeaderAndIsr(localBrokerId, leaderEpoch, isrToSend.toList, partitionState.leaderRecoveryState, partitionEpoch)
-    val updatedState = PendingExpandIsr(partitionState.isr, newInSyncReplicaId, newLeaderAndIsr)
+    val newLeaderAndIsr = LeaderAndIsr(
+      localBrokerId,
+      leaderEpoch,
+      isrToSend.toList,
+      partitionState.leaderRecoveryState,
+      partitionEpoch
+    )
+    val updatedState = PendingExpandIsr(
+      partitionState.isr,
+      newInSyncReplicaId,
+      newLeaderAndIsr,
+      partitionState
+    )
     partitionState = updatedState
     updatedState
   }
@@ -1514,15 +1538,30 @@ class Partition(val topicPartition: TopicPartition,
     // erroneously advance the HW if the `AlterPartition` were to fail. Hence the "maximal ISR"
     // for `PendingShrinkIsr` is the the current ISR.
     val isrToSend = partitionState.isr -- outOfSyncReplicaIds
-    val newLeaderAndIsr = LeaderAndIsr(localBrokerId, leaderEpoch, isrToSend.toList, partitionState.leaderRecoveryState, partitionEpoch)
-    val updatedState = PendingShrinkIsr(partitionState.isr, outOfSyncReplicaIds, newLeaderAndIsr)
+    val newLeaderAndIsr = LeaderAndIsr(
+      localBrokerId,
+      leaderEpoch,
+      isrToSend.toList,
+      partitionState.leaderRecoveryState,
+      partitionEpoch
+    )
+    val updatedState = PendingShrinkIsr(
+      partitionState.isr,
+      outOfSyncReplicaIds,
+      newLeaderAndIsr,
+      partitionState
+    )
     partitionState = updatedState
     updatedState
   }
 
   private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
     debug(s"Submitting ISR state change $proposedIsrState")
-    val future = alterIsrManager.submit(topicPartition, proposedIsrState.sentLeaderAndIsr, controllerEpoch)
+    val future = alterIsrManager.submit(
+      new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+      proposedIsrState.sentLeaderAndIsr,
+      controllerEpoch
+    )
     future.whenComplete { (leaderAndIsr, e) =>
       var hwIncremented = false
       var shouldRetry = false
@@ -1571,10 +1610,18 @@ class Partition(val topicPartition: TopicPartition,
     error match {
       case Errors.OPERATION_NOT_ATTEMPTED =>
         // Since the operation was not attempted, it is safe to reset back to the committed state.
-        partitionState = CommittedPartitionState(proposedIsrState.isr, LeaderRecoveryState.RECOVERED)
+        partitionState = proposedIsrState.partitionStateToRollBackTo
         debug(s"Failed to alter partition to $proposedIsrState since there is a pending AlterPartition still inflight. " +
           s"partition state has been reset to the latest committed state $partitionState")
         false
+      case Errors.INELIGIBLE_REPLICA =>
+        // Since the operation was rejected, it is safe to reset back to the committed state. This
+        // assumes that the current state was still the correct expected state.
+        // This is only raised in KRaft mode.
+        partitionState = proposedIsrState.partitionStateToRollBackTo
+        debug(s"Failed to alter partition to $proposedIsrState since the controller rejected at least one replica " +
+          s"because it is ineligible to join the ISR. partition state has been reset to the latest committed state $partitionState.")
+        false
       case Errors.UNKNOWN_TOPIC_OR_PARTITION =>
         debug(s"Failed to alter partition to $proposedIsrState since the controller doesn't know about " +
           "this topic or partition. Giving up.")
@@ -1588,6 +1635,10 @@ class Partition(val topicPartition: TopicPartition,
       case Errors.INVALID_REQUEST =>
         debug(s"Failed to alter partition to $proposedIsrState because the request is invalid. Giving up.")
         false
+      case Errors.NEW_LEADER_ELECTED =>
+        // This is only raised in KRaft mode.
+        debug(s"ISR updated to ${partitionState.isr.mkString(",")} but this broker is not longer the leader.")
+        false
       case _ =>
         warn(s"Failed to update ISR to $proposedIsrState due to unexpected $error. Retrying.")
         true
@@ -1625,8 +1676,8 @@ class Partition(val topicPartition: TopicPartition,
       info(s"ISR updated to ${partitionState.isr.mkString(",")} and version updated to $partitionEpoch")
 
       proposedIsrState match {
-        case PendingExpandIsr(_, _, _) => alterPartitionListener.markIsrExpand()
-        case PendingShrinkIsr(_, _, _) => alterPartitionListener.markIsrShrink()
+        case PendingExpandIsr(_, _, _, _) => alterPartitionListener.markIsrExpand()
+        case PendingShrinkIsr(_, _, _, _) => alterPartitionListener.markIsrShrink()
       }
 
       // we may need to increment high watermark since ISR could be down to 1