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

[giuseppelillo]

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: Validate a switched follower's seal fetch through the classic follower checks, then record its fetch state so ISR can re-expand without reading diskless data locally. Divergence and stale broker epochs produce partition-level responses.
• makeFollower: Give a switched, at-seal, out-of-ISR follower a catch-up fetcher.
• ReplicaFetcherThread: Keep fetching until controller metadata shows the replica back in ISR.
• Add unit coverage for seal fetch validation, fetcher scheduling and eviction, plus a URP-recovery system test with complete live-broker JMX observations.

[Copilot]

Pull request overview

This PR fixes an operational issue where ReplicaManager’s UnderReplicatedPartitions aggregate could remain non-zero (or otherwise not reflect live ISR state) after switching a topic from classic to diskless, by ensuring the leader continues to observe the follower’s fetch state until the replica is back in ISR.

Changes:

• Update follower fetch routing to treat fetchOffset == classicToDisklessStartOffset as still eligible for unified-log reads for follower requests, enabling the leader to observe fetch progress at the seal offset.
• Ensure diskless followers schedule a fetch even when caught up to the seal but currently out of ISR, so the leader can expand ISR again.
• Add a system test that stops a follower after switch, asserts URP rises, then restarts the follower and asserts URP clears after ISR fully recovers.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
tests/kafkatest/tests/inkless/inkless_topic_switch_test.py	Adds URP JMX scraping + a new test to verify URP rises on replica loss and clears after replica recovery post-switch.
core/src/main/scala/kafka/server/ReplicaManager.scala	Adjusts diskless follower fetch eligibility at the seal offset and ensures fetching is scheduled when out of ISR even if already caught up.
core/src/main/scala/kafka/server/ReplicaFetcherThread.scala	Prevents evicting switched partitions from the replica fetcher until the replica is back in ISR.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (3)

tests/kafkatest/tests/inkless/inkless_topic_switch_test.py:409

• _live_cluster_jmx_sum can return 0 even when one or more live brokers failed to report the metric (e.g., JMX read failed or no samples yet). That can make wait-for-zero checks pass prematurely and mask a real UnderReplicatedPartitions > 0 on an unsampled broker. Consider treating partial observations as a scrape miss and returning None unless every live broker produced a value for the requested gauge.

            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

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

• The comment describes the follower's leader epoch as "STALE (ahead-of-leader)", but an epoch greater than the leader's is not stale; it's mismatched/ahead-of-leader. Tightening the wording makes the test intent clearer.

      // 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.

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

• The eviction comment says the fetcher stops only once the replica is in ISR, but the condition also allows eviction for consolidating topics even if not in ISR. Updating the comment avoids a mismatch between documentation and behavior.

    // 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.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Suppressed comments (2)

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

• Prechecking with getReplica(...).foreach suppresses the classic invalid-replica response: an unassigned follower skips fetchRecords and receives Errors.NONE. Call fetchRecords directly so followerReplicaOrThrow returns UNKNOWN_LEADER_EPOCH/NOT_LEADER_OR_FOLLOWER as appropriate.

                    partition.getReplica(params.replicaId).foreach { _ =>

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

• This equality guard bypasses Partition.fetchRecords precisely when the request epoch is invalid. Classic follower reads return FENCED_LEADER_EPOCH for an older epoch and UNKNOWN_LEADER_EPOCH for a newer one, but this path returns NONE; invoke the classic validation and let the existing catch convert those exceptions to partition errors.

                val requestEpochMatchesLeader =
                  fetchPartitionData.currentLeaderEpoch.toScala.forall(_.intValue() == partition.getLeaderEpoch)
                if (requestEpochMatchesLeader) {

[jeqo]

Looks good, just a few more comments and a finding:

ReplicationControlManager#expandIsrForDisklessManagedPartitions re-adds an unfenced broker to the ISR of every diskless.enable=true partition without consulting classicToDisklessStartOffset. I wrote that in #643 for born-diskless managed-RF, where any live replica is current because all the data is in object storage but missed the switched case, where the classic prefix below the seal exists only in the replicas' local logs. So today a returning replica can enter a switched partition's ISR with no evidence its prefix is complete, and is then electable.

I can open a follow-up PR to skip partitions whose seal isn't NO_CLASSIC_TO_DISKLESS_START_OFFSET to let them earn ISR through the path this PR builds, if this feels like stretching the scope of this PR.