fix(inkless:switch): clear under-replicated partitions after classic-to-diskless switch

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

@@ -807,7 +807,7 @@ abstract class AbstractFetcherThread(name: String,
     }
   }
 
-  private def delayPartitions(partitions: Iterable[TopicPartition], delay: Long): Unit = {
+  protected def delayPartitions(partitions: Iterable[TopicPartition], delay: Long): Unit = {
     partitionMapLock.lockInterruptibly()
     try {
       for (partition <- partitions) {

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

@@ -51,6 +51,9 @@ class ReplicaFetcherThread(name: String,
   // and should be evicted from this fetcher.
   private[server] val partitionsToEvictAfterDisklessSwitch = mutable.Buffer[TopicPartition]()
 
+  // At the seal but not yet in metadata ISR, so we cannot evict. Visible for testing.
+  private[server] val partitionsAwaitingIsrRecovery = mutable.Buffer[TopicPartition]()
+
   override protected def latestEpoch(topicPartition: TopicPartition): Optional[Integer] = {
     replicaMgr.localLogOrException(topicPartition).latestEpoch
   }
@@ -100,6 +103,7 @@ class ReplicaFetcherThread(name: String,
     super.doWork()
     completeDelayedFetchRequests()
     evictFullySwitchedDisklessPartitions()
+    backOffPartitionsAwaitingIsrRecovery()
   }
 
   /**
@@ -168,14 +172,25 @@ class ReplicaFetcherThread(name: String,
     if (shouldRecordReplicationBytesIn)
       brokerTopicStats.updateReplicationBytesIn(records.sizeInBytes)
 
-    // Stop fetching after the switch from classic to diskless is completed: once the controller
-    // has committed a classicToDisklessStartOffset for this partition AND our local LEO has reached it,
-    // the follower is fully caught up to the leader's frozen classic log and must not keep fetching.
-    val classicToDisklessStartOffset = replicaMgr.inklessMetadataView().getClassicToDisklessStartOffset(topicPartition)
+    // Stop fetching once the switch is complete: seal is committed, local LEO has reached it,
+    // and this replica is in ISR. A consolidating partition evicts without waiting for ISR so
+    // it can hand off to the consolidation fetcher.
+    val inklessMetadataView = replicaMgr.inklessMetadataView()
+    val classicToDisklessStartOffset = inklessMetadataView.getClassicToDisklessStartOffset(topicPartition)
+    def isConsolidatingPartition: Boolean =
+      brokerConfig.disklessRemoteStorageConsolidationEnabled &&
+        inklessMetadataView.isConsolidatingDisklessTopic(topicPartition.topic)
     if (shouldEvictFullySwitchedDisklessPartitions &&
         classicToDisklessStartOffset >= 0 &&
         log.logEndOffset >= classicToDisklessStartOffset) {
-      partitionsToEvictAfterDisklessSwitch += topicPartition
+      if (isConsolidatingPartition || inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId)) {
+        partitionsToEvictAfterDisklessSwitch += topicPartition
+      } else {
+        // The leader answers this fetch from immediateFetchResponses and does not park it
+        // in the fetch purgatory, so maxWaitMs is ignored. Delay here or we re-fetch at
+        // network rate until the ISR expansion lands.
+        partitionsAwaitingIsrRecovery += topicPartition
+      }
     }
 
     logAppendInfo
@@ -188,6 +203,16 @@ class ReplicaFetcherThread(name: String,
     }
   }
 
+  // Visible for testing. Must run from doWork, not processPartitionData: processFetchRequest
+  // overwrites fetch state right after processPartitionData and would drop an inline delay.
+  private[server] def backOffPartitionsAwaitingIsrRecovery(): Unit = {
+    if (partitionsAwaitingIsrRecovery.nonEmpty) {
+      val toDelay = partitionsAwaitingIsrRecovery.toSet
+      partitionsAwaitingIsrRecovery.clear()
+      delayPartitions(toDelay, brokerConfig.replicaFetchBackoffMs.toLong)
+    }
+  }
+
   private def evictFullySwitchedDisklessPartitions(): Unit = {
     if (partitionsToEvictAfterDisklessSwitch.nonEmpty) {
       val toEvict = partitionsToEvictAfterDisklessSwitch.toSet

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

@@ -42,7 +42,7 @@ import org.apache.kafka.common.message.ListOffsetsResponseData.{ListOffsetsParti
 import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.{OffsetForLeaderPartition, OffsetForLeaderTopic}
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.{EpochEndOffset, OffsetForLeaderTopicResult}
 import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse
-import org.apache.kafka.common.message.{DescribeLogDirsResponseData, DescribeProducersResponseData}
+import org.apache.kafka.common.message.{DescribeLogDirsResponseData, DescribeProducersResponseData, FetchResponseData}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.protocol.Errors
@@ -91,6 +91,7 @@ import java.util.stream.Collectors
 import scala.collection.{Map, Seq, Set, immutable, mutable}
 import scala.jdk.CollectionConverters._
 import scala.jdk.OptionConverters.RichOptional
+import scala.util.control.NonFatal
 
 /*
  * Result metadata of a log append operation on the log
@@ -2508,6 +2509,41 @@ class ReplicaManager(val config: KafkaConfig,
         if (!partitionLookupFailed) {
           val disklessSwitchCompleted = !shouldReadFromUnifiedLog && classicToDisklessStartOffset >= 0
           if (params.isFromFollower && disklessSwitchCompleted) {
+            var fetchError = Errors.NONE
+            var divergingEpoch = Optional.empty[FetchResponseData.EpochEndOffset]
+            // A recovered follower for a switched partition may already be caught up to the
+            // seal offset but still be outside ISR. Record the seal-offset fetch so the normal
+            // ISR expansion path can observe that the follower is caught up without reading
+            // diskless data into the local log.
+            if (fetchPartitionData.fetchOffset >= classicToDisklessStartOffset) {
+              getPartitionOrError(tp.topicPartition) match {
+                case Right(partition) =>
+                  try {
+                    // Use the classic follower-read validation without returning any records.
+                    val fetchAtSeal = new PartitionData(
+                      fetchPartitionData.topicId,
+                      classicToDisklessStartOffset,
+                      fetchPartitionData.logStartOffset,
+                      0,
+                      fetchPartitionData.currentLeaderEpoch,
+                      fetchPartitionData.lastFetchedEpoch
+                    )
+                    val readInfo = partition.fetchRecords(
+                      fetchParams = params,
+                      fetchPartitionData = fetchAtSeal,
+                      fetchTimeMs = time.milliseconds,
+                      maxBytes = 0,
+                      minOneMessage = false,
+                      updateFetchState = true
+                    )
+                    divergingEpoch = readInfo.divergingEpoch
+                  } catch {
+                    case NonFatal(e) =>
+                      fetchError = Errors.forException(e)
+                  }
+                case Left(error) => fetchError = error
+              }
+            }
             // The partition has fully switched to diskless and the follower is asking for an offset at or beyond it.
             // Followers must never replicate diskless records into their local log. Return
             // an empty response with HW clamped to the seal offset so the fetcher loop sees the
@@ -2517,11 +2553,11 @@ class ReplicaManager(val config: KafkaConfig,
             // local data intact and remains able to serve consumer reads from the local log.
             immediateFetchResponses += tp ->
               new FetchPartitionData(
-                Errors.NONE,
-                classicToDisklessStartOffset,
-                0L,
+                fetchError,
+                if (fetchError == Errors.NONE) classicToDisklessStartOffset else UnifiedLog.UNKNOWN_OFFSET,
+                if (fetchError == Errors.NONE) 0L else UnifiedLog.UNKNOWN_OFFSET,
                 MemoryRecords.EMPTY,
-                Optional.empty(),
+                divergingEpoch,
                 OptionalLong.empty(),
                 Optional.empty(),
                 OptionalInt.empty(),
@@ -4102,11 +4138,14 @@ class ReplicaManager(val config: KafkaConfig,
               val isNewLeaderEpoch = partition.makeFollower(info.partition, isNew, offsetCheckpoints, Some(info.topicId), partitionAssignedDirectoryId)
               partition.seal()
               changedPartitions.add(partition)
-              if (seal >= 0 && partition.localLogOrException.highWatermark < seal) {
+              val isOutOfIsr = !info.partition.isr.contains(config.brokerId)
+              if (seal >= 0 && (partition.localLogOrException.highWatermark < seal || isOutOfIsr)) {
                 // 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. The
                 // ReplicaFetcher self-evicts once the follower has read past the seal.
+                // Also schedule one fetch when this replica is already caught up but out
+                // of ISR, so the leader observes its fetch state and can expand ISR.
                 partitionsToStartFetching.put(tp, partition)
               } else if (seal == PartitionRegistration.CLASSIC_TO_DISKLESS_SWITCH_PENDING && isNewLeaderEpoch) {
                 // Switch is in flight: the leader has already sealed its log and

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

@@ -102,6 +102,12 @@ class InklessMetadataView(val metadataCache: KRaftMetadataCache, val defaultConf
       .getOrElse(PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET)
   }
 
+  def isReplicaInIsr(topicPartition: TopicPartition, replicaId: Int): Boolean = {
+    Option(metadataCache.currentImage().topics().getTopic(topicPartition.topic()))
+      .flatMap(topicImage => Option(topicImage.partitions().get(topicPartition.partition())))
+      .exists(_.isr.contains(replicaId))
+  }
+
   /**
    * The diskless leader epoch (E_d) captured at the classic-to-diskless switch, or
    * [[PartitionRegistration.NO_DISKLESS_LEADER_EPOCH]] when the partition never switched (born-diskless

core/src/test/scala/kafka/server/metadata/InklessMetadataViewTest.scala

@@ -296,6 +296,21 @@ class InklessMetadataViewTest {
     assertEquals(PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET, metadataView.getClassicToDisklessStartOffset(tp))
   }
 
+  @Test
+  def testIsReplicaInIsrUsesImageState(): Unit = {
+    val tp = new TopicPartition("switched", 0)
+    stubImageTopic(tp.topic(), util.Map.of(Integer.valueOf(0), partitionRegistration()))
+    assertTrue(metadataView.isReplicaInIsr(tp, 1))
+    assertFalse(metadataView.isReplicaInIsr(tp, 2))
+  }
+
+  @Test
+  def testIsReplicaInIsrReturnsFalseWhenTopicMissing(): Unit = {
+    val tp = new TopicPartition("missing", 0)
+    stubImageWithoutTopic(tp.topic())
+    assertFalse(metadataView.isReplicaInIsr(tp, 1))
+  }
+
   @Nested
   class TopicConfigCacheTest {
     @Test

core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala

@@ -759,6 +759,74 @@ class ReplicaFetcherThreadTest {
       expectEviction = true)
   }
 
+  @Test
+  def shouldNotEvictPartitionAtSealUntilMetadataIsrContainsReplica(): Unit = {
+    verifyDisklessSwitchEviction(
+      classicToDisklessStartOffset = 100L,
+      logEndOffsetAfterAppend = 100L,
+      expectEviction = false,
+      replicaInIsr = false)
+  }
+
+  @Test
+  def shouldDelayPartitionAtSealWhileMetadataIsrDoesNotContainReplica(): Unit = {
+    val props = TestUtils.createBrokerConfig(1)
+    val config = KafkaConfig.fromProps(props)
+
+    val mockBlockingSend: BlockingSend = mock(classOf[BlockingSend])
+    when(mockBlockingSend.brokerEndPoint()).thenReturn(brokerEndPoint)
+
+    val log: UnifiedLog = mock(classOf[UnifiedLog])
+    when(log.logEndOffset).thenReturn(100L)
+    when(log.latestEpoch).thenReturn(Optional.of(Integer.valueOf(1)))
+    when(log.maybeUpdateHighWatermark(anyLong())).thenReturn(Optional.empty)
+
+    val partition: Partition = mock(classOf[Partition])
+    when(partition.localLogOrException).thenReturn(log)
+    when(partition.appendRecordsToFollowerOrFutureReplica(any[MemoryRecords], any[Boolean], any[Int]))
+      .thenReturn(Some(mock(classOf[LogAppendInfo])))
+
+    val inklessMetadataView: InklessMetadataView = mock(classOf[InklessMetadataView])
+    when(inklessMetadataView.getClassicToDisklessStartOffset(t1p0)).thenReturn(100L)
+    when(inklessMetadataView.isReplicaInIsr(t1p0, config.brokerId)).thenReturn(false)
+
+    val replicaFetcherManager: ReplicaFetcherManager = mock(classOf[ReplicaFetcherManager])
+    val replicaManager: ReplicaManager = mock(classOf[ReplicaManager])
+    when(replicaManager.getPartitionOrException(any[TopicPartition])).thenReturn(partition)
+    when(replicaManager.localLogOrException(t1p0)).thenReturn(log)
+    when(replicaManager.brokerTopicStats).thenReturn(new BrokerTopicStats)
+    when(replicaManager.inklessMetadataView()).thenReturn(inklessMetadataView)
+    when(replicaManager.replicaFetcherManager).thenReturn(replicaFetcherManager)
+
+    val thread = createReplicaFetcherThread(
+      name = "replica-fetcher",
+      fetcherId = 0,
+      brokerConfig = config,
+      failedPartitions = failedPartitions,
+      replicaMgr = replicaManager,
+      quota = mock(classOf[ReplicaQuota]),
+      leaderEndpointBlockingSend = mockBlockingSend)
+
+    thread.addPartitions(Map(t1p0 -> initialFetchState(Some(topicId1), 100L)))
+    val partitionData = new FetchResponseData.PartitionData()
+      .setPartitionIndex(t1p0.partition)
+      .setRecords(MemoryRecords.withRecords(Compression.NONE,
+        new SimpleRecord(1000, "foo".getBytes(StandardCharsets.UTF_8))))
+    thread.processPartitionData(t1p0, 100L, Int.MaxValue, partitionData)
+
+    assertFalse(thread.fetchState(t1p0).get.isDelayed,
+      "Delay must be applied from doWork, not inline in processPartitionData")
+    verify(replicaFetcherManager, times(0)).removeFetcherForPartitions(any())
+
+    thread.backOffPartitionsAwaitingIsrRecovery()
+
+    val fetchState = thread.fetchState(t1p0).get
+    assertTrue(fetchState.isDelayed,
+      s"Partition must be delayed after the at-seal fetch, got $fetchState")
+    assertEquals(config.replicaFetchBackoffMs.toLong, fetchState.delay.orElse(0L))
+    assertEquals(mutable.Buffer.empty, thread.partitionsAwaitingIsrRecovery)
+  }
+
   @Test
   def shouldNotEvictPartitionWhenLogEndOffsetBelowClassicToDisklessSealOffset(): Unit = {
     verifyDisklessSwitchEviction(
@@ -786,7 +854,8 @@ class ReplicaFetcherThreadTest {
   private def verifyDisklessSwitchEviction(
     classicToDisklessStartOffset: Long,
     logEndOffsetAfterAppend: Long,
-    expectEviction: Boolean
+    expectEviction: Boolean,
+    replicaInIsr: Boolean = true
   ): Unit = {
     val props = TestUtils.createBrokerConfig(1)
     val config = KafkaConfig.fromProps(props)
@@ -802,11 +871,13 @@ class ReplicaFetcherThreadTest {
 
     val partition: Partition = mock(classOf[Partition])
     when(partition.localLogOrException).thenReturn(log)
+    when(partition.inSyncReplicaIds).thenReturn(Set.empty)
     when(partition.appendRecordsToFollowerOrFutureReplica(any[MemoryRecords], any[Boolean], any[Int]))
       .thenReturn(Some(mock(classOf[LogAppendInfo])))
 
     val inklessMetadataView: InklessMetadataView = mock(classOf[InklessMetadataView])
     when(inklessMetadataView.getClassicToDisklessStartOffset(t1p0)).thenReturn(classicToDisklessStartOffset)
+    when(inklessMetadataView.isReplicaInIsr(t1p0, config.brokerId)).thenReturn(replicaInIsr)
 
     val replicaFetcherManager: ReplicaFetcherManager = mock(classOf[ReplicaFetcherManager])
 
@@ -849,6 +920,7 @@ class ReplicaFetcherThreadTest {
       verify(replicaManager, times(0)).startConsolidationFetchersForCaughtUpClassicPartitions(any())
     }
     assertEquals(mutable.Buffer.empty, thread.partitionsToEvictAfterDisklessSwitch)
+    assertEquals(mutable.Buffer.empty, thread.partitionsAwaitingIsrRecovery)
   }
 
   private def newOffsetForLeaderPartitionResult(