From acdcb9ef74997de5c9da2202a0049408b34a23ad Mon Sep 17 00:00:00 2001 From: Giuseppe Lillo Date: Tue, 14 Jul 2026 15:45:33 +0200 Subject: [PATCH 1/6] 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/6] 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/6] 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/6] 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/6] 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/6] 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