fix(inkless:switch): keep switched partitions eligible for ISR shrink - #756
Draft
jeqo wants to merge 3 commits into
Draft
fix(inkless:switch): keep switched partitions eligible for ISR shrink#756jeqo wants to merge 3 commits into
jeqo wants to merge 3 commits into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Enables ISR shrinking for switched diskless partitions while preserving the exemption for born-diskless partitions.
Changes:
- Refines diskless ISR shrink eligibility.
- Starts a catch-up fetch after leader changes.
- Adds regression tests for both behaviors.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
core/src/main/scala/kafka/server/ReplicaManager.scala |
Updates ISR shrinking and follower fetch scheduling. |
core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala |
Adds ISR and leader-change regression tests. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.
Suppressed comments (2)
core/src/main/scala/kafka/server/ReplicaManager.scala:3254
- Switched consolidating partitions also satisfy
!isBornDiskless, but their leader LEO advances beyond the seal while followers hand off at the seal toConsolidationFetcher. That fetch path usesFUTURE_LOCAL_REPLICA_ID(DisklessLeaderEndPoint.scala:92-101), so it never refreshes the leader's followerReplicaState(Partition.scala:1444-1460). Once the lag timeout expires,Partition.maybeShrinkIsrcompares the stale seal offset with the leader's advancing LEO and removes healthy consolidating followers from ISR. Please preserve the liveness-based ISR behavior for the consolidated suffix, or evaluate lag only against the committed seal, and cover this mode with a regression test.
val isBornDiskless = _inklessMetadataView.isDisklessTopic(topicPartition.topic()) &&
_inklessMetadataView.getClassicToDisklessStartOffset(topicPartition) ==
PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET
if (!isBornDiskless)
core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala:7847
- This verifies only that the switched partition was submitted; Mockito still permits an additional submission for
bornDiskless, so a regression that shrinks both partitions would pass. Add an explicit negative verification for the born-diskless topic.
verify(alterPartitionManager, times(1)).submit(
ArgumentMatchers.eq(new org.apache.kafka.server.common.TopicIdPartition(
switched.topicId(), switched.partition())),
argThat[LeaderAndIsr](!_.isr.contains(followerId)))
jeqo
force-pushed
the
jeqo/shrink-switched-isr
branch
2 times, most recently
from
August 18, 2026 11:35
e4af171 to
8f5cd6f
Compare
Comment on lines
+4166
to
+4170
| val leaderChanged = previousLeaderId.exists(_ != info.partition.leader) | ||
| // Skip during controlled shutdown: the leader will not expand ISR for a shutting-down | ||
| // broker (isReplicaIsrEligible), and this replica is about to stop serving. | ||
| if (seal >= 0 && !isInControlledShutdown && | ||
| (partition.localLogOrException.highWatermark < seal || isOutOfIsr)) { | ||
| (partition.localLogOrException.highWatermark < seal || isOutOfIsr || leaderChanged)) { |
maybeShrinkIsr skipped every diskless topic, which includes a partition switched from classic. Those keep records below classicToDisklessStartOffset in the replicas' local logs, so a follower that stalls before reaching the seal is genuinely behind. Nothing removed it from ISR either: the controller shrinks only on fencing, and the seal is taken at the leader's LEO, so a lagging follower does not hold the switch back. The replica stayed in ISR without the prefix and was electable. Shrink switched partitions, exempting only those with no committed seal: born-diskless, or a switch aborted through AlterDisklessSwitch. Neither replicates a classic prefix from this leader, so neither has anything to fall behind on. Judge lag against the seal rather than the leader's log end offset. Past the seal the local log grows only by consolidation, which every replica appends from object storage through DisklessLeaderEndPoint -- a local fetch that never reaches the leader. The leader's recorded state for a healthy consolidating follower therefore stops at the seal while its own LEO keeps advancing, so comparing the two shrinks the whole ISR out, and nothing re-expands it because expansion also waits for a follower fetch. Partition takes the ceiling as an optional argument, defaulted off, so classic partitions keep comparing against the LEO. Shrinking alone would also regress after a leader election. A new leader resets an in-ISR follower's recorded state to UNKNOWN (Replica.resetReplicaState), and a follower at the seal and in ISR starts no fetcher, so the grace period from lastCaughtUpTimeMs would expire and the follower would be shrunk and re-admitted on every leader change. Schedule a catch-up fetch when the leader changed, so the new leader observes the follower's position before the lag timeout. Keyed on the leader moving rather than on a leader-epoch bump: a bump that keeps the same leader does not lose the recorded state, and keying on it would also schedule a pointless fetch on every fresh metadata delta.
The leader-change catch-up test picked broker 3 as the new leader, which ClusterImageTest.IMAGE1 does not register. The leader node therefore never resolved, applyLocalFollowersDelta called addFetcherForPartitions with an empty map, and the `times(1)` assertion held on that empty call: it proved the branch was entered but nothing about where or from which offset the follower would fetch. Move the leader from 0 to 2 (both registered in IMAGE1) and assert the exact InitialFetchState, matching the sibling switch-pending test. Name the leaders after the production variables the test pins (previousLeaderId / newLeaderId) and hoist the seal offset into a val.
jeqo
force-pushed
the
jeqo/shrink-switched-isr
branch
from
August 18, 2026 13:16
8f5cd6f to
90c6a49
Compare
…ting followers Enabling lag-based ISR shrink for switched partitions exposed a second way the leader loses its record of a healthy follower. A consolidating follower reads the consolidated suffix from object storage through DisklessLeaderEndPoint, a local fetch that never reaches the leader, so the leader records that follower only during the classic phase. makeLeader resets the record to UNKNOWN, which leaves a new leader with no evidence at all: the lag timeout expires and the ISR shrinks to the leader alone. Expansion also waits for a follower fetch, so nothing brings the follower back. Leader changes are routine, so this lands on every rolling restart. Divert a consolidation-ready follower to the classic ReplicaFetcher for one round when the leader changed. That fetch records the position on the new leader, then the fetcher self-evicts at the seal and hands the partition back to consolidation, which is the handoff a below-seal follower already takes. The diverted partition also leaves the consolidation fetcher manager: its fetcher from the previous leader epoch would otherwise keep appending to the same log and trip processPartitionData's fetchOffset == logEndOffset check. Cap both sides of the lag comparison. The diversion makes a recorded offset above the seal reachable, because the leader answers from its local log when its own consolidation runs ahead of the follower, recording the real offset instead of the seal. Judging that against a ceiling of the seal fails the offset leg and shrinks the follower one lag timeout later, so a follower recorded at or beyond the ceiling now counts as caught up. Testing: unit tests cover the diversion on a leader change and a follower recorded beyond the seal staying in ISR. Both fail without the corresponding change. The upstream PartitionTest suite passes unchanged.
jeqo
added a commit
that referenced
this pull request
Aug 18, 2026
…ranch The consolidating branch of fetchMessages is not gated on isFromConsumer, so a follower fetch enters it too. On a consolidating leader the local log end offset is the consolidated frontier, well past the seal, so a follower fetch at the seal satisfies fetchOffset < logEndOffset and reads the consolidated suffix from the leader's local log. That skips the at-seal branch entirely and contradicts the invariant stated in the same method: followers must never replicate diskless records into their local log. It also ships those records over the inter-broker path, billing them to ReplicationBytesInPerSec rather than ConsolidationFetchBytesInPerSec, which is the traffic consolidation exists to avoid, and defers ISR admission until the follower matches the frontier. Exclude a follower fetch at or above a committed seal from the local-log path, so the at-seal branch answers it with empty records and the high watermark clamped to the seal. The exclusion is inert today: a consolidating follower evicts from the classic fetcher on the same call that brings it to the seal, so it never issues a fetch at or above the seal. It becomes load-bearing as soon as a consolidating follower stays on the classic fetcher at the seal, which both the ISR wait in this branch and the leader-change diversion in #756 introduce. Testing testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse appends past the seal on the leader, then asserts a follower fetch at the seal receives empty records with the high watermark at the seal, and that the branch still records the follower's position. Without the exclusion the follower receives 85 bytes of the leader's consolidated log.
jeqo
added a commit
that referenced
this pull request
Aug 18, 2026
…ranch The consolidating branch of fetchMessages is not gated on isFromConsumer, so a follower fetch enters it too. On a consolidating leader the local log end offset is the consolidated frontier, well past the seal, so a follower fetch at the seal satisfies fetchOffset < logEndOffset and reads the consolidated suffix from the leader's local log. That skips the at-seal branch entirely and contradicts the invariant stated in the same method: followers must never replicate diskless records into their local log. It also ships those records over the inter-broker path, billing them to ReplicationBytesInPerSec rather than ConsolidationFetchBytesInPerSec, which is the traffic consolidation exists to avoid, and defers ISR admission until the follower matches the frontier. Exclude a follower fetch at or above a committed seal from the local-log path, so the at-seal branch answers it with empty records and the high watermark clamped to the seal. The exclusion is inert today: a consolidating follower evicts from the classic fetcher on the same call that brings it to the seal, so it never issues a fetch at or above the seal. It becomes load-bearing as soon as a consolidating follower stays on the classic fetcher at the seal, which both the ISR wait in this branch and the leader-change diversion in #756 introduce. Testing testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse appends past the seal on the leader, then asserts a follower fetch at the seal receives empty records with the high watermark at the seal, and that the branch still records the follower's position. Without the exclusion the follower receives 85 bytes of the leader's consolidated log.
jeqo
added a commit
that referenced
this pull request
Aug 18, 2026
…ranch The consolidating branch of fetchMessages is not gated on isFromConsumer, so a follower fetch enters it too. On a consolidating leader the local log end offset is the consolidated frontier, well past the seal, so a follower fetch at the seal satisfies fetchOffset < logEndOffset and reads the consolidated suffix from the leader's local log. That skips the at-seal branch entirely and contradicts the invariant stated in the same method: followers must never replicate diskless records into their local log. It also ships those records over the inter-broker path, billing them to ReplicationBytesInPerSec rather than ConsolidationFetchBytesInPerSec, which is the traffic consolidation exists to avoid, and defers ISR admission until the follower matches the frontier. Exclude a follower fetch at or above a committed seal from the local-log path, so the at-seal branch answers it with empty records and the high watermark clamped to the seal. The exclusion is inert today: a consolidating follower evicts from the classic fetcher on the same call that brings it to the seal, so it never issues a fetch at or above the seal. It becomes load-bearing as soon as a consolidating follower stays on the classic fetcher at the seal, which both the ISR wait in this branch and the leader-change diversion in #756 introduce. Testing testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse appends past the seal on the leader, then asserts a follower fetch at the seal receives empty records with the high watermark at the seal, and that the branch still records the follower's position. Without the exclusion the follower receives 85 bytes of the leader's consolidated log.
jeqo
added a commit
that referenced
this pull request
Aug 18, 2026
…ranch The consolidating branch of fetchMessages is not gated on isFromConsumer, so a follower fetch enters it too. On a consolidating leader the local log end offset is the consolidated frontier, well past the seal, so a follower fetch at the seal satisfies fetchOffset < logEndOffset and reads the consolidated suffix from the leader's local log. That skips the at-seal branch entirely and contradicts the invariant stated in the same method: followers must never replicate diskless records into their local log. It also ships those records over the inter-broker path, billing them to ReplicationBytesInPerSec rather than ConsolidationFetchBytesInPerSec, which is the traffic consolidation exists to avoid, and defers ISR admission until the follower matches the frontier. Exclude a follower fetch at or above a committed seal from the local-log path, so the at-seal branch answers it with empty records and the high watermark clamped to the seal. The exclusion is inert today: a consolidating follower evicts from the classic fetcher on the same call that brings it to the seal, so it never issues a fetch at or above the seal. It becomes load-bearing as soon as a consolidating follower stays on the classic fetcher at the seal, which both the ISR wait in this branch and the leader-change diversion in #756 introduce. Testing testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse appends past the seal on the leader, then asserts a follower fetch at the seal receives empty records with the high watermark at the seal, and that the branch still records the follower's position. Without the exclusion the follower receives 85 bytes of the leader's consolidated log.
jeqo
added a commit
that referenced
this pull request
Aug 19, 2026
…ranch The consolidating branch of fetchMessages is not gated on isFromConsumer, so a follower fetch enters it too. On a consolidating leader the local log end offset is the consolidated frontier, well past the seal, so a follower fetch at the seal satisfies fetchOffset < logEndOffset and reads the consolidated suffix from the leader's local log. That skips the at-seal branch entirely and contradicts the invariant stated in the same method: followers must never replicate diskless records into their local log. It also ships those records over the inter-broker path, billing them to ReplicationBytesInPerSec rather than ConsolidationFetchBytesInPerSec, which is the traffic consolidation exists to avoid, and defers ISR admission until the follower matches the frontier. Exclude a follower fetch at or above a committed seal from the local-log path, so the at-seal branch answers it with empty records and the high watermark clamped to the seal. The exclusion is inert today: a consolidating follower evicts from the classic fetcher on the same call that brings it to the seal, so it never issues a fetch at or above the seal. It becomes load-bearing as soon as a consolidating follower stays on the classic fetcher at the seal, which both the ISR wait in this branch and the leader-change diversion in #756 introduce. Testing testFollowerFetchAtSealOnConsolidatingLeaderGetsAtSealResponse appends past the seal on the leader, then asserts a follower fetch at the seal receives empty records with the high watermark at the seal, and that the branch still records the follower's position. Without the exclusion the follower receives 85 bytes of the leader's consolidated log.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
maybeShrinkIsrskipped every diskless topic, which includes a partition switched from classic. Those keep records belowclassicToDisklessStartOffsetin the replicas' local logs, so a follower that stalls before reaching the seal is genuinely behind. Nothing removed it from ISR either: the controller shrinks only on fencing, and the seal is taken at the leader's LEO, so a lagging follower does not hold the switch back. It stayed in ISR without the prefix and was electable.Switched partitions are now shrunk. Three things the shrink alone gets wrong:
DisklessLeaderEndPoint-- a local fetch that never reaches the leader. The leader's recorded state for a healthy consolidating follower therefore stops at the seal while its own LEO keeps advancing, so comparing the two would shrink the entire ISR out, with nothing to re-expand it (expansion also waits for a follower fetch).Partitiontakes the ceiling as an optional argument, defaulted off, so classic partitions are unaffected.UNKNOWN, and a follower at the seal starts no fetcher, so the lag grace period would expire and the follower would be shrunk and re-admitted on every leader change.UNKNOWNstate, which the shrink then acts on: the ISR collapses to the leader alone on every rolling restart, with nothing to re-expand it. One classic fetch records the position; the fetcher self-evicts at the seal and hands the partition back to consolidation. The diverted partition also leaves the consolidation fetcher manager, so two fetchers never append to the same log.Only partitions with no committed seal stay exempt: born-diskless, or a switch aborted through
AlterDisklessSwitch. Neither replicates a classic prefix from this leader.Commits
fix(inkless:switch): keep switched partitions eligible for ISR shrink-- the shrink predicate, the seal-capped lag comparison, and the leader-change catch-up fetch, with tests.test(inkless:switch): assert the catch-up fetch target on leader change-- test-only: the leader-change regression test picked a broker thatClusterImageTest.IMAGE1does not register, so the leader endpoint never resolved and the assertion held on an empty fetcher map. It now moves the leader between two registered brokers and asserts the exactInitialFetchState.fix(inkless:switch): re-establish leader fetch evidence for consolidating followers-- the classic-fetcher diversion on a leader change, and the ceiling applied to both sides of the lag comparison.See the commit messages for the reasoning behind each.
Operator impact
A switched partition whose follower never replicated the classic prefix now shows up as under-replicated and is logged by the usual
Shrinking ISR from ... to ...line, instead of silently staying in ISR. No new config or metric.Test plan
Unit tests in
ReplicaManagerInklessTestcover: a switched partition shrinking a follower that never reached the seal while a born-diskless partition is left alone; a consolidating switched partition keeping a follower recorded at the seal in ISR while the leader's log end offset advances past it; and the catch-up fetch scheduled on a leader change, asserted against the expectedInitialFetchState. Two further tests cover the consolidating leader change: the follower is scheduled on the classic fetcher rather than handed to the reconciler, and a follower recorded beyond the seal stays in ISR. Each fails without its production change. The upstreamPartitionTestsuite passes unchanged, which covers the classic path through the new default argument.Relation to #757
ISR correctness for a switched partition has two legs, and these PRs are one each. #757 closes admission:
changePartitionReassignmentapplied the target replica set in one step for any diskless topic, so a replica added by a reassignment entered ISR holding none of the classic prefix. This PR closes recovery: a replica that ended up in ISR without the prefix -- however it got there -- was never lag-shrunk out. They are independent (either can merge first) and complementary: with only #757, a replica that stalls mid-switch stays in ISR forever; with only this PR, a wrongly admitted replica is electable until the lag timeout expires.Both, together with the unfence guard from #754, key on
classicToDisklessStartOffset == -1to mean "born-diskless". That reading is imprecise for a partition whose switch was aborted viaAlterDisklessSwitch(-1)whilediskless.enableis still true -- it keeps a local classic prefix. Pre-existing (the predicate this replaces,!isDisklessTopic, skipped that partition too) and not fixable at any single one of these sites, sinceLogConfig.validateDisklessTransitionforbids disabling diskless afterwards, which makes that state terminal. Tracked as a follow-up.