fix(inkless:switch): keep switched partitions eligible for ISR shrink

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

@@ -1220,17 +1220,27 @@ class Partition(val topicPartition: TopicPartition,
     delayedOperations.checkAndCompleteAll()
   }
 
-  def maybeShrinkIsr(): Unit = {
+  /**
+   * @param leaderEndOffsetCap highest offset a follower is expected to reach by fetching from this
+   *                           leader, or -1 for the leader's log end offset. Set below the LEO when
+   *                           part of the local log is replicated out of band and the leader therefore
+   *                           holds no fetch evidence of a healthy follower's progress past that point.
+   *                           Inkless (diskless tiered-storage consolidation): `ReplicaManager` passes
+   *                           the classic-to-diskless seal for a switched partition, because past the
+   *                           seal every replica appends the consolidated suffix from object storage
+   *                           instead of fetching it from this leader.
+   */
+  def maybeShrinkIsr(leaderEndOffsetCap: Long = -1L): Unit = {
     def needsIsrUpdate: Boolean = {
       !partitionState.isInflight && inReadLock(leaderIsrUpdateLock) {
-        needsShrinkIsr()
+        needsShrinkIsr(leaderEndOffsetCap)
       }
     }
 
     if (needsIsrUpdate) {
       val alterIsrUpdateOpt = inWriteLock(leaderIsrUpdateLock) {
         leaderLogIfLocal.flatMap { leaderLog =>
-          val outOfSyncReplicaIds = getOutOfSyncReplicas(replicaLagTimeMaxMs)
+          val outOfSyncReplicaIds = getOutOfSyncReplicas(replicaLagTimeMaxMs, leaderEndOffsetCap)
           partitionState match {
             case currentState: CommittedPartitionState if outOfSyncReplicaIds.nonEmpty =>
               val outOfSyncReplicaLog = outOfSyncReplicaIds.map { replicaId =>
@@ -1260,16 +1270,21 @@ class Partition(val topicPartition: TopicPartition,
     }
   }
 
-  private def needsShrinkIsr(): Boolean = {
-    leaderLogIfLocal.exists { _ => getOutOfSyncReplicas(replicaLagTimeMaxMs).nonEmpty }
+  private def needsShrinkIsr(leaderEndOffsetCap: Long): Boolean = {
+    leaderLogIfLocal.exists { _ => getOutOfSyncReplicas(replicaLagTimeMaxMs, leaderEndOffsetCap).nonEmpty }
   }
 
   private def isFollowerOutOfSync(replicaId: Int,
                                   leaderEndOffset: Long,
+                                  leaderEndOffsetCap: Long,
                                   currentTimeMs: Long,
                                   maxLagMs: Long): Boolean = {
     getReplica(replicaId).fold(true) { followerReplica =>
-      !followerReplica.stateSnapshot.isCaughtUp(leaderEndOffset, currentTimeMs, maxLagMs)
+      val followerState = followerReplica.stateSnapshot
+      // A follower recorded at or beyond the ceiling holds everything this leader supplies, so it
+      // cannot lag regardless of when it last fetched.
+      if (leaderEndOffsetCap >= 0 && followerState.logEndOffset >= leaderEndOffsetCap) false
+      else !followerState.isCaughtUp(leaderEndOffset, currentTimeMs, maxLagMs)
     }
   }
 
@@ -1285,14 +1300,19 @@ class Partition(val topicPartition: TopicPartition,
    * is violated, that replica is considered to be out of sync
    *
    * If an ISR update is in-flight, we will return an empty set here
+   *
+   * @param leaderEndOffsetCap see [[maybeShrinkIsr]]
    **/
-  def getOutOfSyncReplicas(maxLagMs: Long): Set[Int] = {
+  def getOutOfSyncReplicas(maxLagMs: Long, leaderEndOffsetCap: Long = -1L): Set[Int] = {
     val current = partitionState
     if (!current.isInflight) {
       val candidateReplicaIds = (current.isr.asScala.map(_.toInt) - localBrokerId).toSet
       val currentTimeMs = time.milliseconds()
-      val leaderEndOffset = localLogOrException.logEndOffset
-      candidateReplicaIds.filter(replicaId => isFollowerOutOfSync(replicaId, leaderEndOffset, currentTimeMs, maxLagMs))
+      val logEndOffset = localLogOrException.logEndOffset
+      val leaderEndOffset =
+        if (leaderEndOffsetCap >= 0) math.min(logEndOffset, leaderEndOffsetCap) else logEndOffset
+      candidateReplicaIds.filter(replicaId =>
+        isFollowerOutOfSync(replicaId, leaderEndOffset, leaderEndOffsetCap, currentTimeMs, maxLagMs))
     } else {
       Set.empty
     }

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -3239,13 +3239,29 @@ class ReplicaManager(val config: KafkaConfig,
       log.highWatermark
   }
 
-  private def maybeShrinkIsr(): Unit = {
+  // private[server] for testing: otherwise only reachable via the scheduled "isr-expiration" task.
+  private[server] def maybeShrinkIsr(): Unit = {
     trace("Evaluating ISR list of partitions to see which replicas can be removed from the ISR")
 
     // Shrink ISRs for non offline partitions
     allPartitions.forEach { (topicPartition, _) =>
-      if (!_inklessMetadataView.isDisklessTopic(topicPartition.topic()))
+      if (_inklessMetadataView.isDisklessTopic(topicPartition.topic())) {
+        val seal = _inklessMetadataView.getClassicToDisklessStartOffset(topicPartition)
+        // -1 means "no committed seal": born-diskless, or a switch aborted through
+        // AlterDisklessSwitch. Neither has a classic prefix replicated from this leader, so there
+        // is nothing to fall behind on.
+        if (seal != PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET) {
+          // Past a committed seal the local log only grows by consolidation, which every replica does
+          // from object storage without fetching from this leader (DisklessLeaderEndPoint), so the
+          // leader's recorded state for a healthy follower stops at the seal. Cap the lag comparison
+          // there, or the whole ISR is shrunk out as the consolidated suffix advances. Nothing
+          // re-expands it either, because expansion also needs a follower fetch.
+          val leaderEndOffsetCap = if (seal >= 0) seal else -1L
+          onlinePartition(topicPartition).foreach(_.maybeShrinkIsr(leaderEndOffsetCap))
+        }
+      } else {
         onlinePartition(topicPartition).foreach(_.maybeShrinkIsr())
+      }
     }
   }
 
@@ -4115,6 +4131,9 @@ class ReplicaManager(val config: KafkaConfig,
       "local followers.")
     val partitionsToStartFetching = new mutable.HashMap[TopicPartition, Partition]
     val partitionsToStopFetching = new mutable.HashMap[TopicPartition, Boolean]
+    // Consolidating followers that need one classic fetch to re-establish the new leader's record of
+    // their position. See the diversion below.
+    val consolidationLeaderEvidenceFetch = new mutable.HashSet[TopicPartition]
     val followerTopicSet = new mutable.HashSet[String]
     localFollowers.foreachEntry { (tp, info) =>
       val isConsolidatingDisklessTopic =
@@ -4139,14 +4158,19 @@ class ReplicaManager(val config: KafkaConfig,
           getOrCreatePartition(tp, delta, info.topicId).foreach { case (partition, isNew) =>
             try {
               val partitionAssignedDirectoryId = directoryIds.find(_._1.topicPartition() == tp).map(_._2)
+              val previousLeaderId = partition.leaderReplicaIdOpt
               val isNewLeaderEpoch = partition.makeFollower(info.partition, isNew, offsetCheckpoints, Some(info.topicId), partitionAssignedDirectoryId)
               partition.seal()
               changedPartitions.add(partition)
               val isOutOfIsr = !info.partition.isr.contains(config.brokerId)
+              // A new leader resets our recorded fetch state to UNKNOWN (Replica.resetReplicaState),
+              // so without a fetch to re-establish it the lastCaughtUpTimeMs grace period expires and
+              // the leader shrinks us out of ISR.
+              val leaderChanged = previousLeaderId.exists(_ != info.partition.leader)
               // Skip during controlled shutdown: the leader will not expand ISR for a shutting-down
               // broker (isReplicaIsrEligible), and this replica is about to stop serving.
               if (seal >= 0 && !isInControlledShutdown &&
-                (partition.localLogOrException.highWatermark < seal || isOutOfIsr)) {
+                (partition.localLogOrException.highWatermark < seal || isOutOfIsr || leaderChanged)) {
                 // Schedule a catch-up fetch when the local HW is below the seal -- either
                 // because we restarted with a stale HW (unclean shutdown) or because we
                 // were just added as a replica and have an empty local log.
@@ -4183,6 +4207,7 @@ class ReplicaManager(val config: KafkaConfig,
             //   is unavailable. This is required to ensure that we include the partition's
             //   high watermark in the checkpoint file (see KAFKA-1647).
             val partitionAssignedDirectoryId = directoryIds.find(_._1.topicPartition() == tp).map(_._2)
+            val previousLeaderId = partition.leaderReplicaIdOpt
             val isNewLeaderEpoch = partition.makeFollower(info.partition, isNew, offsetCheckpoints, Some(info.topicId), partitionAssignedDirectoryId)
 
             if (isInControlledShutdown && (info.partition.leader == NO_LEADER ||
@@ -4195,6 +4220,9 @@ class ReplicaManager(val config: KafkaConfig,
               partition.invokeOnBecomingFollowerListeners()
               // Otherwise, fetcher is restarted if the leader epoch has changed.
               partitionsToStartFetching.put(tp, partition)
+              if (isConsolidatingDisklessTopic && previousLeaderId.exists(_ != info.partition.leader)) {
+                consolidationLeaderEvidenceFetch.add(tp)
+              }
             }
 
             changedPartitions.add(partition)
@@ -4243,11 +4271,20 @@ class ReplicaManager(val config: KafkaConfig,
       // to the consolidation reconciler (startConsolidationFetchersForCaughtUpClassicPartitions).
       // Routing a below-seal/pending partition straight to the reconciler would strand it: the
       // reconciler returns Retry and no classic fetcher would ever bring it up to the seal.
+      // A consolidation-ready follower is also diverted to the classic fetcher for one round when the
+      // leader changed. Consolidation reads from object storage, so the new leader never sees a fetch
+      // from this replica and keeps the UNKNOWN state that makeLeader installed, which maybeShrinkIsr
+      // removes from ISR once the lag timeout passes. One classic fetch records the position; the
+      // fetcher then self-evicts at the seal and hands the partition back to consolidation.
       val (consolidatingDisklessPartitionsToStartFetching, classicPartitionsToStartFetching) = partitionsToStartFetching.partition { case (tp, partition) =>
-        isReadyForConsolidation(tp, partition)
+        isReadyForConsolidation(tp, partition) && !consolidationLeaderEvidenceFetch.contains(tp)
       }
       replicaFetcherManager.removeFetcherForPartitions(classicPartitionsToStartFetching.keySet)
-      consolidationFetcherManager.foreach(_.removeFetcherForPartitions(consolidatingDisklessPartitionsToStartFetching.keySet))
+      // Diverted partitions keep a consolidation fetcher from the previous leader epoch. Evict it as
+      // well, or both fetchers append to the same log and processPartitionData fails the
+      // fetchOffset == logEndOffset check.
+      consolidationFetcherManager.foreach(_.removeFetcherForPartitions(
+        consolidatingDisklessPartitionsToStartFetching.keySet ++ consolidationLeaderEvidenceFetch))
       stateChangeLogger.info(s"Stopped fetchers as part of become-follower for ${partitionsToStartFetching.size} partitions")
 
       val listenerName = config.interBrokerListenerName.value