aiven · giuseppelillo · Jul 14, 2026
diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -163,12 +163,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
     }
 

diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -2218,6 +2218,24 @@ 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 =>
+                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
@@ -3746,11 +3764,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

diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/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])))
 

diff --git a/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py b/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,58 @@ 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."""
+        key = "%s:Value" % obj_name
+        total = 0.0
+        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)
+        return int(total)
+
+    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=%d, expected=%d",
+                             count, expected_count)
+            return 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=%d, expected_at_least=%d",
+                             count, min_count)
+            return 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 +1285,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: