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 || partition.inSyncReplicaIds.contains(brokerConfig.brokerId))) {
       partitionsToEvictAfterDisklessSwitch += topicPartition
     }
 

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

@@ -2508,6 +2508,34 @@ class ReplicaManager(val config: KafkaConfig,
         if (!partitionLookupFailed) {
           val disklessSwitchCompleted = !shouldReadFromUnifiedLog && classicToDisklessStartOffset >= 0
           if (params.isFromFollower && disklessSwitchCompleted) {
+            // 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 =>
+                // This short-circuit bypasses the normal fetch read, which is where the request's
+                // leader epoch is validated before follower state is updated. Re-check it here so a
+                // stale-epoch fetch cannot push a follower into ISR. A divergent log *below* the seal
+                // is not a concern: a follower can only reach the frozen seal LEO through the classic
+                // ReplicaFetcher, which reconciles leader epochs and truncates any divergent suffix
+                // before it gets there. An absent request epoch (older fetch protocol) is a match.
+                val requestEpochMatchesLeader =
+                  fetchPartitionData.currentLeaderEpoch.toScala.forall(_.intValue() == partition.getLeaderEpoch)
+                if (requestEpochMatchesLeader) {
+                  partition.getReplica(params.replicaId).foreach { replica =>
+                    partition.updateFollowerFetchState(
+                      replica,
+                      followerFetchOffsetMetadata = new LogOffsetMetadata(classicToDisklessStartOffset),
+                      followerStartOffset = fetchPartitionData.logStartOffset,
+                      followerFetchTimeMs = time.milliseconds,
+                      leaderEndOffset = partition.localLogOrException.logEndOffset,
+                      params.replicaEpoch
+                    )
+                  }
+                }
+              }
+            }
             // 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
@@ -4102,11 +4130,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/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala

@@ -802,6 +802,7 @@ class ReplicaFetcherThreadTest {
 
     val partition: Partition = mock(classOf[Partition])
     when(partition.localLogOrException).thenReturn(log)
+    when(partition.inSyncReplicaIds).thenReturn(Set(config.brokerId))
     when(partition.appendRecordsToFollowerOrFutureReplica(any[MemoryRecords], any[Boolean], any[Int]))
       .thenReturn(Some(mock(classOf[LogAppendInfo])))
 

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

@@ -7096,6 +7096,106 @@ class ReplicaManagerInklessTest {
     }
   }
 
+  @Test
+  def testFollowerFetchAtSealRecordsFetchStateAndAllowsIsrExpansionWhenEpochMatches(): Unit = {
+    val followerId = 2
+    val sealOffset = 5L
+    val replicaManager = createReplicaManager(
+      List(disklessTopicPartition.topic()),
+      topicIdMapping = Map(disklessTopicPartition.topic() -> disklessTopicPartition.topicId()),
+      disklessManagedReplicasEnabled = true,
+    )
+    try {
+      // Leader broker (1) hosts the fully-switched partition; follower (2) is assigned but out of ISR.
+      val partition = setupSwitchedLeaderWithOutOfSyncFollower(
+        replicaManager, disklessTopicPartition, followerId, sealOffset)
+      assertFalse(partition.inSyncReplicaIds.contains(followerId),
+        "Follower must start outside the ISR")
+
+      // AlterPartition submission is async; hand back an uncompleted future so the ISR-expansion path
+      // can run without the mock returning null (which would NPE inside submitAlterPartition).
+      doReturn(new CompletableFuture[Any]())
+        .when(alterPartitionManager).submit(any(), any())
+      clearInvocations(alterPartitionManager)
+
+      // Follower fetches at the seal, carrying the CURRENT leader epoch.
+      val fetchParams = new FetchParams(
+        followerId, -1L, 0L, 1, 1024 * 1024, FetchIsolation.LOG_END, Optional.empty())
+      val fetchInfos = Seq(
+        disklessTopicPartition ->
+          new PartitionData(disklessTopicPartition.topicId(), sealOffset, 0L, 1024 * 1024,
+            Optional.of(partition.getLeaderEpoch)))
+
+      @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null
+      replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA,
+        response => responseData = response.toMap)
+
+      // The follower still gets the empty, HW-clamped placeholder (never reads diskless data)...
+      val data = responseData(disklessTopicPartition)
+      assertEquals(Errors.NONE, data.error)
+      assertEquals(MemoryRecords.EMPTY, data.records)
+      assertEquals(sealOffset, data.highWatermark)
+
+      // ...but because the request epoch matched the leader's, the leader recorded the follower's
+      // fetch state at the seal...
+      assertEquals(sealOffset, partition.getReplica(followerId).get.stateSnapshot.logEndOffset)
+      // ...which drives the normal ISR-expansion path for the now caught-up follower.
+      verify(alterPartitionManager).submit(any(), any())
+    } finally {
+      replicaManager.shutdown(checkpointHW = false)
+    }
+  }
+
+  @Test
+  def testFollowerFetchAtSealSkipsFetchStateAndIsrExpansionWhenLeaderEpochStale(): Unit = {
+    val followerId = 2
+    val sealOffset = 5L
+    val replicaManager = createReplicaManager(
+      List(disklessTopicPartition.topic()),
+      topicIdMapping = Map(disklessTopicPartition.topic() -> disklessTopicPartition.topicId()),
+      disklessManagedReplicasEnabled = true,
+    )
+    try {
+      val partition = setupSwitchedLeaderWithOutOfSyncFollower(
+        replicaManager, disklessTopicPartition, followerId, sealOffset)
+      assertFalse(partition.inSyncReplicaIds.contains(followerId),
+        "Follower must start outside the ISR")
+
+      doReturn(new CompletableFuture[Any]())
+        .when(alterPartitionManager).submit(any(), any())
+      clearInvocations(alterPartitionManager)
+
+      // Follower fetches at the seal, but carries a STALE (ahead-of-leader) leader epoch. This mirrors
+      // the classic read path, which validates the request epoch before touching follower state.
+      val staleEpoch = partition.getLeaderEpoch + 1
+      val fetchParams = new FetchParams(
+        followerId, -1L, 0L, 1, 1024 * 1024, FetchIsolation.LOG_END, Optional.empty())
+      val fetchInfos = Seq(
+        disklessTopicPartition ->
+          new PartitionData(disklessTopicPartition.topicId(), sealOffset, 0L, 1024 * 1024,
+            Optional.of(staleEpoch)))
+
+      @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null
+      replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA,
+        response => responseData = response.toMap)
+
+      // The follower still gets the same empty, HW-clamped response...
+      val data = responseData(disklessTopicPartition)
+      assertEquals(Errors.NONE, data.error)
+      assertEquals(MemoryRecords.EMPTY, data.records)
+      assertEquals(sealOffset, data.highWatermark)
+
+      // ...but the epoch guard refused to record its fetch state, so it stays out of the ISR and no
+      // AlterPartition is submitted.
+      assertEquals(UnifiedLog.UNKNOWN_OFFSET,
+        partition.getReplica(followerId).get.stateSnapshot.logEndOffset)
+      assertFalse(partition.inSyncReplicaIds.contains(followerId))
+      verify(alterPartitionManager, never()).submit(any(), any())
+    } finally {
+      replicaManager.shutdown(checkpointHW = false)
+    }
+  }
+
   @Test
   def testFollowerFetchBelowClassicToDisklessStartOffsetReadsFromClassicLog(): Unit = {
     val fetchHandlerCtor = mockFetchHandler(Map.empty)
@@ -7343,6 +7443,62 @@ class ReplicaManagerInklessTest {
     partition
   }
 
+  /**
+   * Like [[setupHybridLeaderPartition]] but assigns an extra follower replica that starts OUT of the
+   * ISR, and seals the local classic log at `sealOffset` (LEO == HW == sealOffset). Used to exercise
+   * the switched-follower fetch path on the leader, where a caught-up follower must be re-admitted to
+   * the ISR via `updateFollowerFetchState`. Registers the cluster brokers in the metadata cache so the
+   * follower can pass the leader's ISR-eligibility (alive, unfenced) check.
+   */
+  private def setupSwitchedLeaderWithOutOfSyncFollower(replicaManager: ReplicaManager,
+                                                       topicIdPartition: TopicIdPartition,
+                                                       followerId: Int,
+                                                       sealOffset: Long): Partition = {
+    val leaderId = replicaManager.config.brokerId
+    // ClusterImageTest.IMAGE1 registers brokers 1 (leader) and 2 (follower) as alive/unfenced, which
+    // is what isReplicaIsrEligible consults via metadataCache.getAliveBrokerEpoch on the leader.
+    replicaManager.metadataCache.asInstanceOf[KRaftMetadataCache].setImage(imageFromTopics(TopicsImage.EMPTY))
+
+    val topicDelta = new TopicsDelta(TopicsImage.EMPTY)
+    topicDelta.replay(new TopicRecord()
+      .setName(topicIdPartition.topic())
+      .setTopicId(topicIdPartition.topicId()))
+    topicDelta.replay(new PartitionRecord()
+      .setTopicId(topicIdPartition.topicId())
+      .setPartitionId(topicIdPartition.partition())
+      .setLeader(leaderId)
+      .setLeaderEpoch(0)
+      .setPartitionEpoch(0)
+      .setReplicas(List[Integer](leaderId, followerId).asJava)
+      .setIsr(List[Integer](leaderId).asJava))
+
+    val (partition, _) = replicaManager.getOrCreatePartition(
+      topicIdPartition.topicPartition(),
+      topicDelta,
+      topicIdPartition.topicId()).get
+    partition.makeLeader(
+      partitionRegistration(
+        leaderId,
+        leaderEpoch = 0,
+        isr = Array(leaderId),
+        partitionEpoch = 0,
+        replicas = Array(leaderId, followerId)),
+      isNew = false,
+      new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava),
+      None)
+
+    val records = (0L until sealOffset).map { i =>
+      new SimpleRecord(s"key-$i".getBytes, s"value-$i".getBytes)
+    }.toArray
+    val log = partition.localLogOrException
+    log.appendAsLeader(MemoryRecords.withRecords(0L, Compression.NONE, 0, records: _*), 0)
+    log.updateHighWatermark(sealOffset)
+
+    when(replicaManager.inklessMetadataView().getClassicToDisklessStartOffset(topicIdPartition.topicPartition()))
+      .thenReturn(sealOffset)
+    partition
+  }
+
   private def mockFetchHandler(disklessResponse: Map[TopicIdPartition, FetchPartitionData]) = {
     // We use constructor mocking here to inject a FetchHandler mock into ReplicaManager,
     // because ReplicaManager internally constructs its own FetchHandler instance and does not

tests/kafkatest/tests/inkless/inkless_topic_switch_test.py

@@ -121,9 +121,11 @@ def __init__(self, test_context: TestContext) -> None:
         ),
     }
     SEALED_LEADER_PARTITIONS_JMX_OBJECT = "kafka.server:type=ReplicaManager,name=SealedPartitionsCount"
+    UNDER_REPLICATED_PARTITIONS_JMX_OBJECT = "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
     INIT_DISKLESS_IN_FLIGHT_PARTITIONS_JMX_OBJECT = _IDLM_OBJ % "InFlightPartitions"
     SWITCH_COMPLETION_JMX_OBJECT_NAMES = [
         SEALED_LEADER_PARTITIONS_JMX_OBJECT,
+        UNDER_REPLICATED_PARTITIONS_JMX_OBJECT,
         INIT_DISKLESS_IN_FLIGHT_PARTITIONS_JMX_OBJECT,
     ]
     SWITCH_STATE_JMX_OBJECT_NAMES = [obj for pair in SWITCH_STATE_GAUGES.values() for obj in pair]
@@ -377,6 +379,65 @@ def check():
                      (topic, timeout_sec, expected_sealed_leader_count))
         )
 
+    def _live_cluster_jmx_sum(self, obj_name):
+        """Read and sum one JMX gauge across live broker nodes.
+
+        Returns None when no live broker reported the gauge so callers can tell a
+        genuine zero apart from a scrape miss, rather than reading a failed scrape
+        as 0 and passing a wait-for-zero check prematurely.
+        """
+        key = "%s:Value" % obj_name
+        total = 0.0
+        observed = False
+        for node in self.kafka.nodes:
+            if not self.kafka.pids(node):
+                continue
+            idx = self.kafka.idx(node)
+            try:
+                self.kafka.read_jmx_output(idx, node)
+            except Exception as e:
+                self.logger.debug("Failed to read JMX from live broker %s: %s",
+                                  node.account.hostname, e)
+                continue
+            if idx - 1 >= len(self.kafka.jmx_stats):
+                continue
+            time_to_stats = self.kafka.jmx_stats[idx - 1]
+            if time_to_stats:
+                latest = max(time_to_stats.keys())
+                total += time_to_stats[latest].get(key, 0)
+                observed = True
+        return int(total) if observed else None
+
+    def _wait_for_under_replicated_partitions(self, expected_count, timeout_sec=120):
+        def check():
+            count = self._live_cluster_jmx_sum(self.UNDER_REPLICATED_PARTITIONS_JMX_OBJECT)
+            self.logger.info("Cluster UnderReplicatedPartitions=%s, expected=%d",
+                             count, expected_count)
+            return count is not None and count == expected_count
+
+        wait_until(
+            check,
+            timeout_sec=timeout_sec,
+            backoff_sec=2,
+            err_msg="UnderReplicatedPartitions did not become %d within %ds" %
+                    (expected_count, timeout_sec)
+        )
+
+    def _wait_for_under_replicated_partitions_at_least(self, min_count, timeout_sec=120):
+        def check():
+            count = self._live_cluster_jmx_sum(self.UNDER_REPLICATED_PARTITIONS_JMX_OBJECT)
+            self.logger.info("Cluster UnderReplicatedPartitions=%s, expected_at_least=%d",
+                             count, min_count)
+            return count is not None and count >= min_count
+
+        wait_until(
+            check,
+            timeout_sec=timeout_sec,
+            backoff_sec=2,
+            err_msg="UnderReplicatedPartitions did not reach at least %d within %ds" %
+                    (min_count, timeout_sec)
+        )
+
     # -----------------------------------------------------------------------
     # Helpers: produce / consume
     # -----------------------------------------------------------------------
@@ -1231,6 +1292,35 @@ def test_classic_data_available_after_restarts(self, metadata_quorum) -> None:
                                                     wait_for_completion=True)
         assert consumed == total, "Expected exactly %d messages after rolling restart but got %d" % (total, consumed)
 
+    @cluster(num_nodes=5)
+    @matrix(metadata_quorum=[quorum.isolated_kraft])
+    def test_switched_topic_urp_clears_after_replica_recovery(self, metadata_quorum) -> None:
+        """A switched hybrid partition should report URP only while ISR is short.
+
+        This guards the operational contract that the ReplicaManager aggregate
+        gauge is live state, not a sticky artifact of switching to diskless:
+        stopping one replica raises URP, and the same replica catching back up
+        clears it.
+        """
+        self.num_partitions = 1
+        self._create_kafka()
+        self.kafka.start()
+        self._create_classic_topic(num_partitions=1)
+
+        self._produce_messages(num_messages=5000)
+
+        self._switch_topic_to_diskless()
+        self._wait_for_switch_complete()
+        self._wait_for_under_replicated_partitions(0)
+
+        follower = self._get_follower_nodes(partition=0)[0]
+        self._stop_broker(follower, clean_shutdown=False)
+        self._wait_for_under_replicated_partitions_at_least(1)
+
+        self._start_broker(follower)
+        self._wait_for_all_partitions_isr_full(num_partitions=1)
+        self._wait_for_under_replicated_partitions(0)
+
     @cluster(num_nodes=5)
     @matrix(metadata_quorum=[quorum.isolated_kraft])
     def test_classic_data_available_after_leader_failures(self, metadata_quorum) -> None: