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

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

@@ -169,12 +169,18 @@ class ReplicaFetcherThread(name: String,
       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)
+    // has committed a classicToDisklessStartOffset for this partition, our local LEO has reached it,
+    // and this replica is in ISR, the follower is fully caught up to the leader's frozen classic log
+    // and must not keep fetching.
+    val inklessMetadataView = replicaMgr.inklessMetadataView()
+    val classicToDisklessStartOffset = inklessMetadataView.getClassicToDisklessStartOffset(topicPartition)
+    val isConsolidatingPartition =
+      brokerConfig.disklessRemoteStorageConsolidationEnabled &&
+        inklessMetadataView.isConsolidatingDisklessTopic(topicPartition.topic)
     if (shouldEvictFullySwitchedDisklessPartitions &&
         classicToDisklessStartOffset >= 0 &&
-        log.logEndOffset >= classicToDisklessStartOffset) {
+        log.logEndOffset >= classicToDisklessStartOffset &&
+        (isConsolidatingPartition || inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId))) {
       partitionsToEvictAfterDisklessSwitch += topicPartition
     }
 

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
@@ -2508,6 +2508,52 @@ 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).foreach { partition =>
+                val requestEpochMatchesLeader =
+                  fetchPartitionData.currentLeaderEpoch.toScala.forall(_.intValue() == partition.getLeaderEpoch)
+                if (requestEpochMatchesLeader) {
+                  try {
+                    partition.getReplica(params.replicaId).foreach { _ =>
+                      // 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 e@(_: UnknownTopicOrPartitionException |
+                            _: NotLeaderOrFollowerException |
+                            _: UnknownLeaderEpochException |
+                            _: FencedLeaderEpochException |
+                            _: OffsetOutOfRangeException |
+                            _: ReplicaNotAvailableException |
+                            _: KafkaStorageException |
+                            _: InconsistentTopicIdException) =>
+                      fetchError = Errors.forException(e)
+                  }
+                }
+              }
+            }
             // 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 +2563,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 +4148,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,15 @@ class ReplicaFetcherThreadTest {
       expectEviction = true)
   }
 
+  @Test
+  def shouldNotEvictPartitionAtSealUntilMetadataIsrContainsReplica(): Unit = {
+    verifyDisklessSwitchEviction(
+      classicToDisklessStartOffset = 100L,
+      logEndOffsetAfterAppend = 100L,
+      expectEviction = false,
+      replicaInIsr = false)
+  }
+
   @Test
   def shouldNotEvictPartitionWhenLogEndOffsetBelowClassicToDisklessSealOffset(): Unit = {
     verifyDisklessSwitchEviction(
@@ -786,7 +795,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 +812,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])