From acdcb9ef74997de5c9da2202a0049408b34a23ad Mon Sep 17 00:00:00 2001 From: Giuseppe Lillo Date: Tue, 14 Jul 2026 15:45:33 +0200 Subject: [PATCH 1/8] fix(inkless:switch): clear under-replicated partitions after classic-to-diskless switch A follower that dropped out of ISR and recovered after a classic-to-diskless switch stayed under-replicated forever, keeping UnderReplicatedPartitions stuck. - Leader fetch handler: record a switched follower's fetch state at the seal so ISR can re-expand, gated on a leader-epoch check (no diskless data read locally). - makeFollower: give a switched, at-seal, out-of-ISR follower a catch-up fetcher. - ReplicaFetcherThread: don't self-evict a switched partition until it's in ISR. Adds unit tests for the epoch-gated seal fetch and a URP-recovery system test. --- .../kafka/server/ReplicaFetcherThread.scala | 14 +- .../scala/kafka/server/ReplicaManager.scala | 33 +++- .../server/ReplicaFetcherThreadTest.scala | 1 + .../server/ReplicaManagerInklessTest.scala | 156 ++++++++++++++++++ .../inkless/inkless_topic_switch_test.py | 90 ++++++++++ 5 files changed, 289 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 4052f2ef736..899f36d1243 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/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 } diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index d6a95b2711f..4102a3143cf 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/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 diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala index eedaaff32ad..586a9e53876 100644 --- 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/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 8d0e41bdc72..58f42c840ca 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/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 diff --git a/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py b/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py index ef8dfc19b3f..f8aab12afb0 100644 --- 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,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: From 59dc03b2bcb1a17644b56316c5300fbd89c7b39e Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Tue, 11 Aug 2026 16:09:06 +0200 Subject: [PATCH 2/8] fix(inkless:switch): validate follower epochs at seal Co-authored-by: Cursor --- .../scala/kafka/server/ReplicaManager.scala | 37 +++++++----- .../server/ReplicaManagerInklessTest.scala | 59 +++++++++++++++++++ 2 files changed, 80 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 4102a3143cf..979896eca6e 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/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,30 +2508,35 @@ class ReplicaManager(val config: KafkaConfig, if (!partitionLookupFailed) { val disklessSwitchCompleted = !shouldReadFromUnifiedLog && classicToDisklessStartOffset >= 0 if (params.isFromFollower && disklessSwitchCompleted) { + 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 => - // 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 + 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 } } } @@ -2549,7 +2554,7 @@ class ReplicaManager(val config: KafkaConfig, classicToDisklessStartOffset, 0L, MemoryRecords.EMPTY, - Optional.empty(), + divergingEpoch, OptionalLong.empty(), Optional.empty(), OptionalInt.empty(), diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 58f42c840ca..2834c55335e 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -7146,6 +7146,65 @@ class ReplicaManagerInklessTest { } } + @Test + def testFollowerFetchAtSealReturnsDivergingEpochAndSkipsIsrExpansion(): 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) + val leaderId = replicaManager.config.brokerId + val leaderEpoch = partition.getLeaderEpoch + 2 + partition.makeLeader( + partitionRegistration( + leaderId, + leaderEpoch, + isr = Array(leaderId), + partitionEpoch = partition.getPartitionEpoch + 1, + replicas = Array(leaderId, followerId)), + isNew = false, + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava), + None) + + doReturn(new CompletableFuture[Any]()) + .when(alterPartitionManager).submit(any(), any()) + clearInvocations(alterPartitionManager) + + 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(leaderEpoch), + Optional.of(leaderEpoch - 1))) + + @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null + replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, + response => responseData = response.toMap) + + val data = responseData(disklessTopicPartition) + assertEquals(Errors.NONE, data.error) + assertEquals(MemoryRecords.EMPTY, data.records) + assertTrue(data.divergingEpoch.isPresent) + assertEquals(0, data.divergingEpoch.get.epoch) + assertEquals(sealOffset, data.divergingEpoch.get.endOffset) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, + partition.getReplica(followerId).get.stateSnapshot.logEndOffset) + verify(alterPartitionManager, never()).submit(any(), any()) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + @Test def testFollowerFetchAtSealSkipsFetchStateAndIsrExpansionWhenLeaderEpochStale(): Unit = { val followerId = 2 From 88d5b944c743e04745699679f4f8589b39d78abc Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Tue, 11 Aug 2026 16:32:13 +0200 Subject: [PATCH 3/8] fix(inkless:switch): stop fetchers after ISR recovery Co-authored-by: Cursor --- .../scala/kafka/server/ReplicaFetcherThread.scala | 2 +- .../server/metadata/InklessMetadataView.scala | 6 ++++++ .../server/metadata/InklessMetadataViewTest.scala | 15 +++++++++++++++ .../kafka/server/ReplicaFetcherThreadTest.scala | 15 +++++++++++++-- 4 files changed, 35 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 899f36d1243..54d02678021 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -180,7 +180,7 @@ class ReplicaFetcherThread(name: String, if (shouldEvictFullySwitchedDisklessPartitions && classicToDisklessStartOffset >= 0 && log.logEndOffset >= classicToDisklessStartOffset && - (isConsolidatingPartition || partition.inSyncReplicaIds.contains(brokerConfig.brokerId))) { + (isConsolidatingPartition || inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId))) { partitionsToEvictAfterDisklessSwitch += topicPartition } diff --git a/core/src/main/scala/kafka/server/metadata/InklessMetadataView.scala b/core/src/main/scala/kafka/server/metadata/InklessMetadataView.scala index abfd75aa23f..0cdc0828401 100644 --- a/core/src/main/scala/kafka/server/metadata/InklessMetadataView.scala +++ b/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 diff --git a/core/src/test/scala/kafka/server/metadata/InklessMetadataViewTest.scala b/core/src/test/scala/kafka/server/metadata/InklessMetadataViewTest.scala index edb02ffb009..68acef2901a 100644 --- a/core/src/test/scala/kafka/server/metadata/InklessMetadataViewTest.scala +++ b/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 diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala index 586a9e53876..76a0395ac33 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala +++ b/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,12 +812,13 @@ class ReplicaFetcherThreadTest { val partition: Partition = mock(classOf[Partition]) when(partition.localLogOrException).thenReturn(log) - when(partition.inSyncReplicaIds).thenReturn(Set(config.brokerId)) + 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]) From 67f081f0f9acda04ffe1b356378570b2eec20cfe Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Wed, 12 Aug 2026 11:35:12 +0200 Subject: [PATCH 4/8] fix(inkless:switch): isolate stale follower fetch errors Co-authored-by: Cursor --- .../scala/kafka/server/ReplicaManager.scala | 56 +++++++++++-------- .../server/ReplicaManagerInklessTest.scala | 45 +++++++++++++++ 2 files changed, 79 insertions(+), 22 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 979896eca6e..8289c478abe 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -2508,6 +2508,7 @@ 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 @@ -2518,25 +2519,36 @@ class ReplicaManager(val config: KafkaConfig, val requestEpochMatchesLeader = fetchPartitionData.currentLeaderEpoch.toScala.forall(_.intValue() == partition.getLeaderEpoch) if (requestEpochMatchesLeader) { - 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 + 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 | + _: ReplicaNotAvailableException | + _: KafkaStorageException | + _: InconsistentTopicIdException) => + fetchError = Errors.forException(e) } } } @@ -2550,9 +2562,9 @@ 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, divergingEpoch, OptionalLong.empty(), diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 2834c55335e..23b45e2dec7 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -7146,6 +7146,51 @@ class ReplicaManagerInklessTest { } } + @Test + def testFollowerFetchAtSealReturnsPartitionErrorWhenBrokerEpochStale(): 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) + val currentBrokerEpoch = replicaManager.metadataCache.getAliveBrokerEpoch(followerId).get.longValue() + val staleBrokerEpoch = currentBrokerEpoch - 1L + assertTrue(staleBrokerEpoch >= 0L) + + doReturn(new CompletableFuture[Any]()) + .when(alterPartitionManager).submit(any(), any()) + clearInvocations(alterPartitionManager) + + val fetchParams = new FetchParams( + followerId, staleBrokerEpoch, 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) + + assertNotNull(responseData) + val data = responseData(disklessTopicPartition) + assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, data.error) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, data.highWatermark) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, data.logStartOffset) + assertEquals(MemoryRecords.EMPTY, data.records) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, + partition.getReplica(followerId).get.stateSnapshot.logEndOffset) + verify(alterPartitionManager, never()).submit(any(), any()) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + @Test def testFollowerFetchAtSealReturnsDivergingEpochAndSkipsIsrExpansion(): Unit = { val followerId = 2 From 772f9500fae5b7b3e1da4b2d4523a99fc6fc53d8 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Wed, 12 Aug 2026 11:35:23 +0200 Subject: [PATCH 5/8] test(inkless:switch): strengthen recovery coverage Co-authored-by: Cursor --- .../server/ReplicaManagerInklessTest.scala | 43 ++++++++++++++++++- .../inkless/inkless_topic_switch_test.py | 24 ++++++----- 2 files changed, 55 insertions(+), 12 deletions(-) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 23b45e2dec7..971489e1f1d 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -6423,7 +6423,8 @@ class ReplicaManagerInklessTest { brokerId: Int, leaderId: Int, classicToDisklessStartOffset: Long = PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET, - disklessLeaderEpoch: Int = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH + disklessLeaderEpoch: Int = PartitionRegistration.NO_DISKLESS_LEADER_EPOCH, + followerInIsr: Boolean = true ): TopicsDelta = { val delta = new TopicsDelta(TopicsImage.EMPTY) delta.replay(new TopicRecord().setName(topicName).setTopicId(topicId)) @@ -6431,7 +6432,7 @@ class ReplicaManagerInklessTest { .setPartitionId(0) .setTopicId(topicId) .setReplicas(util.Arrays.asList(brokerId, leaderId)) - .setIsr(util.Arrays.asList(brokerId, leaderId)) + .setIsr(if (followerInIsr) util.Arrays.asList(brokerId, leaderId) else util.Arrays.asList(leaderId)) .setLeader(leaderId) .setLeaderEpoch(0) .setPartitionEpoch(0) @@ -6578,6 +6579,44 @@ class ReplicaManagerInklessTest { } } + @Test + def testApplyDeltaStartsCatchUpFetcherWhenDisklessFollowerAtSealButOutOfIsr(): Unit = { + val topicName = "switched-topic" + val topicId = Uuid.randomUuid() + val tp = new TopicPartition(topicName, 0) + val brokerId = 1 + val leaderId = 2 + val sealOffset = 10L + + val mockFetcherManager = mock(classOf[ReplicaFetcherManager]) + when(mockFetcherManager.removeFetcherForPartitions(any())).thenReturn(Map.empty[TopicPartition, PartitionFetchState]) + + val replicaManager = spy(createReplicaManager( + List(topicName), + mockReplicaFetcherManager = Some(mockFetcherManager) + )) + try { + val log = replicaManager.logManager.getOrCreateLog(tp, isNew = true, topicId = Optional.of(topicId)) + populateLocalLogAtLeoAndCheckpointedHwm( + replicaManager, tp, log, leo = sealOffset, hw = sealOffset) + when(replicaManager.inklessMetadataView().getClassicToDisklessStartOffset(tp)).thenReturn(sealOffset) + + val delta = disklessFollowerDelta( + topicName, topicId, brokerId, leaderId, followerInIsr = false) + replicaManager.applyDelta(delta, imageFromTopics(delta.apply())) + + val leaderEndpoint = ClusterImageTest.IMAGE1.broker(leaderId).listeners().get("PLAINTEXT") + verify(mockFetcherManager).addFetcherForPartitions(Map(tp -> InitialFetchState( + topicId = Some(topicId), + leader = new BrokerEndPoint(leaderId, leaderEndpoint.host(), leaderEndpoint.port()), + currentLeaderEpoch = 0, + initOffset = sealOffset + ))) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + @Test def testApplyDeltaRestoresStaleHwmWhenSwitchedFollowerBecomesLeader(): Unit = { val topicName = "switched-topic" diff --git a/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py b/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py index f8aab12afb0..90f3e293f4b 100644 --- a/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py +++ b/tests/kafkatest/tests/inkless/inkless_topic_switch_test.py @@ -382,16 +382,17 @@ def check(): 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. + Returns None unless every live broker reported the gauge. Leader-owned + gauges such as UnderReplicatedPartitions cannot be treated as zero when + the leader's scrape is missing. """ key = "%s:Value" % obj_name total = 0.0 - observed = False - for node in self.kafka.nodes: - if not self.kafka.pids(node): - continue + observed_nodes = 0 + live_nodes = [node for node in self.kafka.nodes if self.kafka.pids(node)] + if not live_nodes: + return None + for node in live_nodes: idx = self.kafka.idx(node) try: self.kafka.read_jmx_output(idx, node) @@ -404,9 +405,12 @@ def _live_cluster_jmx_sum(self, obj_name): 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 + latest_stats = time_to_stats[latest] + if key not in latest_stats: + continue + total += latest_stats[key] + observed_nodes += 1 + return int(total) if observed_nodes == len(live_nodes) else None def _wait_for_under_replicated_partitions(self, expected_count, timeout_sec=120): def check(): From 8d5373720523067b41dc309f600c9b576940aeed Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Wed, 12 Aug 2026 15:58:07 +0200 Subject: [PATCH 6/8] fix(inkless:switch): isolate seal fetch offset errors Co-authored-by: Cursor --- .../scala/kafka/server/ReplicaManager.scala | 1 + .../server/ReplicaManagerInklessTest.scala | 68 +++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 8289c478abe..95ff28334ef 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -2545,6 +2545,7 @@ class ReplicaManager(val config: KafkaConfig, _: NotLeaderOrFollowerException | _: UnknownLeaderEpochException | _: FencedLeaderEpochException | + _: OffsetOutOfRangeException | _: ReplicaNotAvailableException | _: KafkaStorageException | _: InconsistentTopicIdException) => diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 971489e1f1d..bf92639ad04 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -7230,6 +7230,74 @@ class ReplicaManagerInklessTest { } } + @Test + def testFollowerFetchAtSealIsolatesOffsetOutOfRangeError(): Unit = { + val followerId = 2 + val sealOffset = 5L + val validTopicPartition = new TopicIdPartition( + disklessTopicPartition.topicId(), 1, disklessTopicPartition.topic()) + val replicaManager = createReplicaManager( + List(disklessTopicPartition.topic()), + topicIdMapping = Map(disklessTopicPartition.topic() -> disklessTopicPartition.topicId()), + disklessManagedReplicasEnabled = true, + ) + try { + val invalidEpochPartition = setupSwitchedLeaderWithOutOfSyncFollower( + replicaManager, disklessTopicPartition, followerId, sealOffset) + val validPartition = setupSwitchedLeaderWithOutOfSyncFollower( + replicaManager, validTopicPartition, followerId, sealOffset) + val brokerEpoch = replicaManager.metadataCache.getAliveBrokerEpoch(followerId).get.longValue() + + doReturn(new CompletableFuture[Any]()) + .when(alterPartitionManager).submit(any(), any()) + clearInvocations(alterPartitionManager) + + val fetchParams = new FetchParams( + followerId, brokerEpoch, 0L, 1, 1024 * 1024, FetchIsolation.LOG_END, Optional.empty()) + val fetchInfos = Seq( + disklessTopicPartition -> + new PartitionData( + disklessTopicPartition.topicId(), + sealOffset, + 0L, + 1024 * 1024, + Optional.of(invalidEpochPartition.getLeaderEpoch), + Optional.of(invalidEpochPartition.getLeaderEpoch + 1)), + validTopicPartition -> + new PartitionData( + validTopicPartition.topicId(), + sealOffset, + 0L, + 1024 * 1024, + Optional.of(validPartition.getLeaderEpoch)) + ) + + @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null + replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, + response => responseData = response.toMap) + + assertNotNull(responseData) + assertEquals(2, responseData.size) + + val invalidEpochData = responseData(disklessTopicPartition) + assertEquals(Errors.OFFSET_OUT_OF_RANGE, invalidEpochData.error) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, invalidEpochData.highWatermark) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, invalidEpochData.logStartOffset) + assertEquals(MemoryRecords.EMPTY, invalidEpochData.records) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, + invalidEpochPartition.getReplica(followerId).get.stateSnapshot.logEndOffset) + + val validData = responseData(validTopicPartition) + assertEquals(Errors.NONE, validData.error) + assertEquals(sealOffset, validData.highWatermark) + assertEquals(MemoryRecords.EMPTY, validData.records) + assertEquals(sealOffset, validPartition.getReplica(followerId).get.stateSnapshot.logEndOffset) + verify(alterPartitionManager, times(1)).submit(any(), any()) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + @Test def testFollowerFetchAtSealReturnsDivergingEpochAndSkipsIsrExpansion(): Unit = { val followerId = 2 From 7fbf9c332e8d456b7abe04f17eebce641e5067eb Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Fri, 14 Aug 2026 09:57:29 +0200 Subject: [PATCH 7/8] resolve code review comments --- .../scala/kafka/server/ReplicaManager.scala | 56 ++-- .../server/ReplicaManagerInklessTest.scala | 263 ++++++++++++------ 2 files changed, 204 insertions(+), 115 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 95ff28334ef..ce0e8db30ba 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -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 @@ -2515,43 +2516,32 @@ class ReplicaManager(val config: KafkaConfig, // 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) { + getPartitionOrError(tp.topicPartition) match { + case Right(partition) => 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 - } + // 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) => + 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. diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index bf92639ad04..e82f7eba330 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -7046,46 +7046,26 @@ class ReplicaManagerInklessTest { @Test def testFollowerFetchAtClassicToDisklessStartOffsetReturnsEmptyAndIdle(): Unit = { + val followerId = 2 + val sealOffset = 100L val fetchHandlerCtor = mockFetchHandler(Map.empty) val cp = mock(classOf[ControlPlane]) val replicaManager = spy(createReplicaManager( List(disklessTopicPartition.topic()), controlPlane = Some(cp), + topicIdMapping = Map(disklessTopicPartition.topic() -> disklessTopicPartition.topicId()), disklessManagedReplicasEnabled = true, )) try { - // Given a fully-switched diskless topic with classicToDisklessStartOffset = 100 - when(replicaManager.inklessMetadataView().getClassicToDisklessStartOffset(disklessTopicPartition.topicPartition())) - .thenReturn(100L) - - // When a follower fetches at offset >= classicToDisklessStartOffset - val fetchParams = new FetchParams( - 1, 1L, // follower fetch - 0L, 1, 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty() - ) - val fetchInfos = Seq( - disklessTopicPartition -> new PartitionData(disklessTopicPartition.topicId(), 100L, 0L, 1024, Optional.empty()) - ) + val partition = setupSwitchedLeaderWithOutOfSyncFollower( + replicaManager, disklessTopicPartition, followerId, sealOffset) + prepareAlterPartitionSubmit() - @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null - val responseCallback = (response: Seq[(TopicIdPartition, FetchPartitionData)]) => { - responseData = response.toMap - } - replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, responseCallback) + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(partition.getLeaderEpoch)) - // Then the response is empty with HW clamped to the seal offset and we never touched - // diskless storage or the local log on behalf of the follower. - assertNotNull(responseData) - assertEquals(1, responseData.size) - val data = responseData(disklessTopicPartition) - assertEquals(Errors.NONE, data.error) - assertEquals(MemoryRecords.EMPTY, data.records) - assertEquals(100L, data.highWatermark) - // logStartOffset must NOT advance the follower's local log start offset (would delete classic data). - assertEquals(0L, data.logStartOffset) - verify(replicaManager, never()).readFromLog(any(), any(), any(), any()) - verify(fetchHandlerCtor.constructed().get(0), never()).handle(any(), any()) - verify(cp, never()).findBatches(any(), any(), any()) + assertEmptySealResponseWithoutDisklessIo(data, replicaManager, fetchHandlerCtor, cp, sealOffset) } finally { replicaManager.shutdown(checkpointHW = false) fetchHandlerCtor.close() @@ -7094,41 +7074,28 @@ class ReplicaManagerInklessTest { @Test def testFollowerFetchAtClassicToDisklessStartOffsetEmptyEvenWhenManagedReplicasDisabled(): Unit = { + val followerId = 2 + val sealOffset = 100L val fetchHandlerCtor = mockFetchHandler(Map.empty) val cp = mock(classOf[ControlPlane]) val replicaManager = spy(createReplicaManager( List(disklessTopicPartition.topic()), controlPlane = Some(cp), + topicIdMapping = Map(disklessTopicPartition.topic() -> disklessTopicPartition.topicId()), disklessManagedReplicasEnabled = false, )) try { - when(replicaManager.inklessMetadataView().getClassicToDisklessStartOffset(disklessTopicPartition.topicPartition())) - .thenReturn(100L) + val partition = setupSwitchedLeaderWithOutOfSyncFollower( + replicaManager, disklessTopicPartition, followerId, sealOffset) + prepareAlterPartitionSubmit() - val fetchParams = new FetchParams( - 1, 1L, // follower fetch - 0L, 1, 1024, FetchIsolation.HIGH_WATERMARK, Optional.empty() - ) - val fetchInfos = Seq( - disklessTopicPartition -> new PartitionData(disklessTopicPartition.topicId(), 150L, 0L, 1024, Optional.empty()) - ) + // Fetch past the seal: still the empty placeholder, not INVALID_REQUEST. + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(partition.getLeaderEpoch), + fetchOffset = Some(150L)) - @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null - val responseCallback = (response: Seq[(TopicIdPartition, FetchPartitionData)]) => { - responseData = response.toMap - } - replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, responseCallback) - - // Same outcome regardless of managedReplicasEnabled: follower never sees diskless data. - assertNotNull(responseData) - assertEquals(1, responseData.size) - val data = responseData(disklessTopicPartition) - assertEquals(Errors.NONE, data.error) - assertEquals(MemoryRecords.EMPTY, data.records) - assertEquals(100L, data.highWatermark) - verify(replicaManager, never()).readFromLog(any(), any(), any(), any()) - verify(fetchHandlerCtor.constructed().get(0), never()).handle(any(), any()) - verify(cp, never()).findBatches(any(), any(), any()) + assertEmptySealResponseWithoutDisklessIo(data, replicaManager, fetchHandlerCtor, cp, sealOffset) } finally { replicaManager.shutdown(checkpointHW = false) fetchHandlerCtor.close() @@ -7358,7 +7325,7 @@ class ReplicaManagerInklessTest { } @Test - def testFollowerFetchAtSealSkipsFetchStateAndIsrExpansionWhenLeaderEpochStale(): Unit = { + def testFollowerFetchAtSealWithNewerLeaderEpochSkipsFetchState(): Unit = { val followerId = 2 val sealOffset = 5L val replicaManager = createReplicaManager( @@ -7369,39 +7336,119 @@ class ReplicaManagerInklessTest { try { val partition = setupSwitchedLeaderWithOutOfSyncFollower( replicaManager, disklessTopicPartition, followerId, sealOffset) - assertFalse(partition.inSyncReplicaIds.contains(followerId), - "Follower must start outside the ISR") + prepareAlterPartitionSubmit() - doReturn(new CompletableFuture[Any]()) - .when(alterPartitionManager).submit(any(), any()) - clearInvocations(alterPartitionManager) + // Ordinary leader-election race: the follower applied epoch N+1, this broker still has N. + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(partition.getLeaderEpoch + 1)) - // 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))) + assertRejectedSealFetch(data, Errors.UNKNOWN_LEADER_EPOCH) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, + partition.getReplica(followerId).get.stateSnapshot.logEndOffset) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } - @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null - replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, - response => responseData = response.toMap) + @Test + def testFollowerFetchAtSealWithFencedLeaderEpochSkipsFetchState(): 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) + val fencedEpoch = partition.getLeaderEpoch + val leaderId = replicaManager.config.brokerId + partition.makeLeader( + partitionRegistration( + leaderId, + leaderEpoch = fencedEpoch + 1, + isr = Array(leaderId), + partitionEpoch = partition.getPartitionEpoch + 1, + replicas = Array(leaderId, followerId)), + isNew = false, + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava), + None) + prepareAlterPartitionSubmit() - // 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) + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(fencedEpoch)) - // ...but the epoch guard refused to record its fetch state, so it stays out of the ISR and no - // AlterPartition is submitted. + assertRejectedSealFetch(data, Errors.FENCED_LEADER_EPOCH) 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 testFollowerFetchAtSealFromRemovedReplicaReturnsUnknownLeaderEpoch(): 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) + val leaderId = replicaManager.config.brokerId + val leaderEpoch = partition.getLeaderEpoch + // Reassignment drops the follower and leaves leaderEpoch unchanged, so the request + // epoch still matches and only replica membership fails. + partition.makeLeader( + partitionRegistration( + leaderId, + leaderEpoch, + isr = Array(leaderId), + partitionEpoch = partition.getPartitionEpoch + 1, + replicas = Array(leaderId)), + isNew = false, + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints.asJava), + None) + assertTrue(partition.getReplica(followerId).isEmpty) + prepareAlterPartitionSubmit() + + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(leaderEpoch)) + + assertRejectedSealFetch(data, Errors.UNKNOWN_LEADER_EPOCH) + assertTrue(partition.getReplica(followerId).isEmpty) + } finally { + replicaManager.shutdown(checkpointHW = false) + } + } + + @Test + def testFollowerFetchAtSealReturnsStorageErrorWhenLogDirOffline(): 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) + prepareAlterPartitionSubmit() + replicaManager.markPartitionOffline(disklessTopicPartition.topicPartition()) + + val data = fetchFollowerAtSeal( + replicaManager, disklessTopicPartition, followerId, sealOffset, + Optional.of(partition.getLeaderEpoch)) + + assertRejectedSealFetch(data, Errors.KAFKA_STORAGE_ERROR) } finally { replicaManager.shutdown(checkpointHW = false) } @@ -7661,6 +7708,58 @@ class ReplicaManagerInklessTest { * 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 prepareAlterPartitionSubmit(): Unit = { + doReturn(new CompletableFuture[Any]()) + .when(alterPartitionManager).submit(any(), any()) + clearInvocations(alterPartitionManager) + } + + private def fetchFollowerAtSeal( + replicaManager: ReplicaManager, + topicIdPartition: TopicIdPartition, + followerId: Int, + sealOffset: Long, + currentLeaderEpoch: Optional[Integer], + fetchOffset: Option[Long] = None + ): FetchPartitionData = { + val offset = fetchOffset.getOrElse(sealOffset) + val fetchParams = new FetchParams( + followerId, -1L, 0L, 1, 1024 * 1024, FetchIsolation.LOG_END, Optional.empty()) + val fetchInfos = Seq( + topicIdPartition -> + new PartitionData(topicIdPartition.topicId(), offset, 0L, 1024 * 1024, currentLeaderEpoch)) + @volatile var responseData: Map[TopicIdPartition, FetchPartitionData] = null + replicaManager.fetchMessages(fetchParams, fetchInfos, QuotaFactory.UNBOUNDED_QUOTA, + response => responseData = response.toMap) + assertNotNull(responseData) + responseData(topicIdPartition) + } + + private def assertEmptySealResponseWithoutDisklessIo( + data: FetchPartitionData, + replicaManager: ReplicaManager, + fetchHandlerCtor: MockedConstruction[FetchHandler], + cp: ControlPlane, + sealOffset: Long + ): Unit = { + assertEquals(Errors.NONE, data.error) + assertEquals(MemoryRecords.EMPTY, data.records) + assertEquals(sealOffset, data.highWatermark) + // logStartOffset=0 so the follower does not advance its local start and drop classic data. + assertEquals(0L, data.logStartOffset) + verify(replicaManager, never()).readFromLog(any(), any(), any(), any()) + verify(fetchHandlerCtor.constructed().get(0), never()).handle(any(), any()) + verify(cp, never()).findBatches(any(), any(), any()) + } + + private def assertRejectedSealFetch(data: FetchPartitionData, expectedError: Errors): Unit = { + assertEquals(expectedError, data.error) + assertEquals(MemoryRecords.EMPTY, data.records) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, data.highWatermark) + assertEquals(UnifiedLog.UNKNOWN_OFFSET, data.logStartOffset) + verify(alterPartitionManager, never()).submit(any(), any()) + } + private def setupSwitchedLeaderWithOutOfSyncFollower(replicaManager: ReplicaManager, topicIdPartition: TopicIdPartition, followerId: Int, From 96dad1d8cb1f15aa6ee1ba975f617cfc53d39369 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Fri, 14 Aug 2026 11:07:04 +0200 Subject: [PATCH 8/8] delay requests at seal waiting for recovery --- .../kafka/server/AbstractFetcherThread.scala | 2 +- .../kafka/server/ReplicaFetcherThread.scala | 35 ++++++++--- .../server/ReplicaFetcherThreadTest.scala | 60 +++++++++++++++++++ 3 files changed, 88 insertions(+), 9 deletions(-) diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala index 1e8841df0ca..cead22165f3 100755 --- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala +++ b/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) { diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 54d02678021..dc2cf592a98 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/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,20 +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, 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. + // 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) - val isConsolidatingPartition = + def isConsolidatingPartition: Boolean = brokerConfig.disklessRemoteStorageConsolidationEnabled && inklessMetadataView.isConsolidatingDisklessTopic(topicPartition.topic) if (shouldEvictFullySwitchedDisklessPartitions && classicToDisklessStartOffset >= 0 && - log.logEndOffset >= classicToDisklessStartOffset && - (isConsolidatingPartition || inklessMetadataView.isReplicaInIsr(topicPartition, brokerConfig.brokerId))) { - partitionsToEvictAfterDisklessSwitch += topicPartition + log.logEndOffset >= classicToDisklessStartOffset) { + 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 @@ -194,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 diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala index 76a0395ac33..eeb6c88f607 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala @@ -768,6 +768,65 @@ class ReplicaFetcherThreadTest { 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( @@ -861,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(