You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
feat(inkless:switch): gate consolidation on the switch phase - #840
A committed seal doesn't mean the classic prefix is safe to drop from local disks. Until the controller commits classicPrefixInRemoteStorage, the local logs of the ISR members store the only complete copy. In phase 1, a replica that starts consolidating advances the cross-tier log start offset over a prefix that remote storage doesn't hold yet. A client routed to a replica outside the ISR can read a truncated prefix.
In phase 2 the opposite holds. In the record that sets the flag, the controller adds every live replica to the ISR without checking its local copy. From then on, brokers trust only the remote copy.
Phase 1: treat the partition as not consolidating
consolidationActiveFor(tp) gates consolidation per partition and is off in phase 1. isReadyForConsolidation and ConsolidationReconciler wait for the flag.
The classic fetcher applies the ISR rule of a topic that doesn't consolidate. A follower at the seal leaves the classic fetcher once it's in the ISR, and it starts consolidating when the flag commits. That commit always bumps the leader epoch. The follower path then runs again.
The broker fences a leader whose local log ends below the seal, as on a topic that doesn't consolidate. Nothing rebuilds the prefix in phase 1, so only phase 2 keeps that leader online for consolidation to rebuild the prefix from remote storage.
The metadata transformer advertises only alive ISR members for a pending switch and in phase 1.
InklessMetadataView exposes the flag, isSwitchedPhaseOne, and isSwitchedWithLocalClassicPrefix.
Phase 2: delete the local classic prefix
When a replica applies the flag, it deletes every local segment below the seal through UnifiedLog.forceDeleteLocalSegmentsBelow. If the active segment contains the end of the prefix, the replica rolls it first. The deletion skips the retention checks. It advances the local log start offset and keeps the log start offset and the leader epoch cache. Reads below the seal take the tiered path, and the leader can still validate follower fetches.
A follower's remote log metadata can trail the leader's, and a tiered read of a segment that the metadata doesn't list resets the consumer. So forced deletion removes only the segments that this broker's highest offset in remote storage covers, and ReplicaManager retries the rest once per second.
The classic fetcher runs the same idempotent step when it evicts a partition at the seal. That covers a follower that fetched part of the prefix after it applied the flag. Because the metadata delta and the eviction both run the deletion, the reconciler has no deletion step or failure path.
readFromLog never serves local records below the seal. If this broker's highest offset in remote storage covers the offset, a consumer read takes the tiered path. Otherwise, and for followers, the broker answers NOT_LEADER_OR_FOLLOWER, and the client retries. The check runs on the records that the read returned, so one rule covers both fetch branches, a fetch that waited in purgatory since phase 1, a read that races the deletion, and a log directory move that swaps the log. A read that hits the local prefix also queues the deletion again.
The tiered path matters most after an upgrade. On earlier releases, a follower at the seal started consolidating in phase 1 and appended the first diskless records to the segment that ends the prefix. Forced deletion can't remove that segment, so without the tiered path, consumers routed to that follower would get errors until local retention deletes it.
A consumer read below the seal never takes a diskless supplement. That read returns nothing when the request's byte budget is already spent, and a supplement from the local log end offset would skip the rest of the prefix.
A log directory move copies the prefix into the future log. The alter-log-dirs thread rolls the future log before it copies the first record at the seal. Without the roll, one segment would contain both the end of the prefix and the start of the diskless records, and forced deletion couldn't remove that segment. Before it promotes, shouldDelayFutureReplicaPromotion runs forced deletion on the caught-up future log and delays promotion while that log still holds prefix records.
If the deletion fails with a storage error, the broker reports the log directory to the log directory failure handler instead of marking the partition offline. The classic fetcher thread runs the deletion, and applyDelta waits for idle fetcher threads to stop while it holds replicaStateChangeLock, so taking that lock from the fetcher thread could deadlock.
UnifiedLog.java and ReplicaAlterLogDirsThread.scala become interleaved paths in INKLESS_OWNERSHIP.
Known limitations
If a replica with a segment that spans the seal leads the partition, a follower below the seal can't fetch that part of the prefix until local retention removes the segment. A log directory move on such a replica waits for the same deletion. Fixing both needs the leader to report the seal as its earliest local offset in phase 2. The follower rebuild uses that ListOffsets response. KC-714 tracks the fix.
If remote storage is off or remote.log.copy.disable is true, the broker keeps the prefix and logs a warning. Consumer reads succeed only for offsets that this broker already sees in remote storage.
This change leaves DisklessFetchOffsetRouter and the pending-state handling as they are. ISR shrink (KC-633), DeleteRecords, and RetentionEnforcer are out of scope.
Testing
ReplicaManagerInklessTest covers per-partition gating, the fence of a phase 1 leader below the seal, the phase 1 to phase 2 hand-off for leaders and followers, deletion on the flag commit and its retry, the read check in both phases with and without purgatory, the tiered path for a consumer read of a local prefix that remote storage covers, a phase 2 read below the seal that returns nothing, a storage error during the deletion, the promotion decision, and a log directory move that spans the flag commit.
ReplicaFetcherThreadTest checks that a consolidating follower at the seal waits for the ISR in phase 1, and that eviction deletes the prefix before consolidation starts.
ConsolidationReconcilerTest covers the phase 1 retry.
UnifiedLogTest checks that forced deletion keeps the log start offset and the epoch cache. Other cases cover the active-segment roll, the remote-copy precondition, and segments that remote storage doesn't cover.
ReplicaAlterLogDirsThreadTest checks that a delayed promotion keeps the partition on the thread, and that the thread rolls the future log at the seal before it appends.
InklessMetadataViewTest and InklessTopicMetadataTransformerTest cover the phase predicates and ISR-only routing.
InklessTopicTypeSwitcherClusterTest runs on the tiered storage stack and waits until every broker deletes its local prefix before it consumes. The consumer then reads the switched prefix from remote storage. With the no-op remote storage manager, the switch test fails.
ReplicaManagerTest, DelayedFetchTest, DisklessSwitchFlowTest, InitDisklessLogManagerTest, and DisklessFetchOffsetRouterTest pass unchanged.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
Unresolved read-safety defects can make retained records appear out of range and trigger consumer data skips, while some phase-2 reads bypass the required local-prefix guard.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
Two critical concurrency issues remain: post-promotion cleanup can deadlock through reversed lock ordering, and a promotion race can let fetches serve unchecked prefix records instead of returning NOT_LEADER_OR_FOLLOWER.
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
Critical races and a delayed-fetch guard bypass can expose unchecked local prefix records during forced deletion or log promotion, requiring enforcement in the shared read path and coordination with deletion.
No open issues 🎉
3 previously missed issues
In code that hasn't changed since last review
Delayed fetch bypasses guard and reads unchecked prefix
A consumer fetch admitted in phase 1 can remain in purgatory until phase 2, while deletion waits for remote metadata. DelayedFetch.onComplete() calls readFromLog() directly and bypasses this guard, so a request without currentLeaderEpoch can read the retained, unchecked prefix. Enforce the guard against the serving log in the shared read path under the partition read lock, and test a flag commit while such a fetch is parked.
Forced deletion exposes unchecked records during phase 2 fetch
During forced deletion, deleteSegments() advances localLogStartOffset before removing the segments from the lookup map, so this guard can admit a phase 2 fetch while the unchecked prefix is still readable. UnifiedLog.read() checks only logStartOffset and can return those local records instead of triggering a tiered read. Enforce the remote-only boundary in the actual read path, or synchronize the guard and read with deletion.
Prefix deletion race exposes unchecked records after log promotion
When both logs retain the prefix, this check permits promotion, but the deferred-deletion scheduler can delete the current log's prefix before the swap. A fetch can then pass the guard on the current log and read unchecked prefix records from the promoted log; the promotion's log-end-offset recheck doesn't detect deletion. Coordinate the final check and swap with current-prefix deletion, or require the future prefix to be gone regardless of the current log's start offset, and test this interleaving.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
Unresolved critical issues with divergent replica epoch caches and directory-move segments that cross the seal can indefinitely block consumer reads or replica promotion.
The reason will be displayed to describe this comment to others. Learn more.
🟡 Changes recommended
A stale classic epoch cache can leave an out-of-ISR replica retaining the prefix indefinitely while reads are rejected, and the cluster test can pass without exercising phase 2.
The reason will be displayed to describe this comment to others. Learn more.
Added some suggestions on the naming to avoid sticking with "switch phases" on functions that then could collide or need be renamed when diskless-to-classic switch is supported. If we could also make sure that comments specifically refer to "classic to diskless switch" would be great.
The reason will be displayed to describe this comment to others. Learn more.
Marking the partition offline is still correct here but we could name the switch phase on the log:
Suggested change
valreason=
if (consolidationActiveFor(tp.topic) && log.remoteLogEnabled())
"the switch is in phase 1, and only a leader that holds the whole classic prefix can request "+
"the classicPrefixInRemoteStorage flag. Move leadership to a replica that holds the prefix "+
"(this needs an unclean election if this broker is the only ISR member)"
else
"the prefix is not recoverable from remote storage (no consolidation or remote storage on this broker)"
stateChangeLogger.error(s"Leader partition $tp has LEO ${log.logEndOffset} below the "+
s"classic-to-diskless seal $seal and the classic prefix [0, $seal) is locally incomplete: $reason. "+
s"Marking the partition offline.")
markPartitionOffline(tp)
Also, if we have unclean leader election on phase 1 then it may not have a way out of the switching without operator. Should we ack this as a known limitation and if there will be a follow-up for this?
The reason will be displayed to describe this comment to others. Learn more.
Applied, without the unclean-election hint. In phase 1 with consolidation and remote storage, the error tells the operator to move leadership to a replica that holds the prefix. Otherwise it says remote storage can't rebuild the prefix on this broker.
On the limitation: yes, and #849 is the follow-up. It blocks unclean leader election while the classic prefix is only on local disks (pending and phase 1). That covers the topic config check, ElectLeaders, broker fencing, the leaderless-partition check, and reassignment cancellation. Once it lands, the controller rejects the election the hint suggests. Leadership can only move to an ISR member, and every ISR member holds the whole prefix.
I've also opened a new ticket for implementing operator tooling to recover a phase 1 partition that lost its classic prefix.
…e switch phase [KC-589]
A committed seal doesn't mean the classic prefix is safe to drop from local
disks. Until the controller commits `classicPrefixInRemoteStorage`, the local
logs of the ISR members are the only complete copy. A replica that starts
consolidating in phase 1 advances the cross-tier log start offset over a
prefix that remote storage doesn't hold yet. A client routed to a non-ISR
replica can read a truncated prefix.
The broker now treats phase 1 like a switched partition on a topic that
doesn't consolidate:
- The consolidation gate is per partition. `consolidationActiveFor(tp)` is
off in phase 1, and `isReadyForConsolidation` and the reconciler wait for
the flag.
- The classic fetcher evicts a consolidating partition at the seal without
waiting for the ISR only in phase 2. In phase 1 it applies the ISR rule of
a topic that doesn't consolidate. The flag commit bumps the leader epoch,
so the follower path re-runs and starts consolidation.
- The broker fences a leader whose local log ends below the seal. Nothing
rebuilds the prefix in phase 1, so only phase 2 keeps that leader online
for consolidation to rebuild the prefix from remote storage.
- The metadata transformer advertises only alive ISR members while the
switch is pending and in phase 1.
`InklessMetadataView` exposes the flag, along with `isSwitchedPhaseOne` and
`isSwitchedWithLocalClassicPrefix`.
- New and updated tests in `ReplicaManagerInklessTest`,
`ReplicaFetcherThreadTest`, `ConsolidationReconcilerTest`,
`InklessMetadataViewTest`, and `InklessTopicMetadataTransformerTest`.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
…arts [KC-589]
In the record that starts phase 2, the controller adds every live replica to
the ISR without checking its local copy of the classic prefix. A replica that
was below the seal, or that diverged, can then serve or consolidate from a
local prefix that nobody validated. From this record on, brokers trust only
the remote copy.
When a replica applies `classicPrefixInRemoteStorage`, it deletes every local
segment below the seal through `UnifiedLog.forceDeleteLocalSegmentsBelow`. If
the active segment contains the end of the prefix, the replica rolls it first.
The deletion skips the retention checks. It advances the local log start
offset and keeps the log start offset and the leader epoch cache, so reads
below the seal take the tiered path, and the leader can still validate
follower fetches. The classic fetcher runs the same idempotent step when it
evicts a partition at the seal. That covers a follower that fetched part of
the prefix after it applied the flag.
The leader requests the flag from its own view of remote storage. A
follower's remote log metadata can trail it. A tiered read of a segment that
the metadata doesn't list fails with `OFFSET_OUT_OF_RANGE` and resets the
consumer. So forced deletion removes only the segments that this
broker's highest offset in remote storage covers, and `ReplicaManager`
retries the rest once per second.
`readFromLog` never serves local records below the seal in phase 2. If this
broker's highest offset in remote storage covers the offset, a consumer read
takes the tiered path. Otherwise, and for followers, the broker answers
`NOT_LEADER_OR_FOLLOWER`, and the client retries. The check runs on the
records that the read returned, so one rule covers both fetch branches, a
fetch that waited in purgatory since phase 1, a read that races the
deletion, and a log directory move that swaps the log. A read that hits the
local prefix also queues the deletion again.
The tiered path matters most after an upgrade. On earlier releases, a
follower at the seal started consolidating in phase 1 and appended the first
diskless records to the segment that ends the prefix. Forced deletion can't
remove that segment, so without the tiered path, consumers routed to that
follower would get errors until local retention deletes it.
A consumer read below the seal never takes a diskless supplement. That read
returns nothing when the request's byte budget is already spent, and a
supplement from the local log end offset would skip the rest of the prefix.
If the deletion fails with a storage error, the broker reports the log
directory to the log directory failure handler instead of marking the
partition offline. The classic fetcher thread runs the deletion when it
evicts a partition, and `applyDelta` waits for idle fetcher threads to stop
while it holds `replicaStateChangeLock`. Taking that lock from the fetcher
thread could deadlock. The failure handler also stops the partition's
fetchers and notifies the controller.
A log directory move copies the prefix into the future log, and the flag
commit deletes it only from the current log. The alter-log-dirs thread rolls
the future log before it copies the first record at the seal. Without the
roll, one segment would contain both the end of the prefix and the start of the
diskless records, and forced deletion couldn't remove that segment. Before it
promotes, the thread asks `shouldDelayFutureReplicaPromotion`. That check
runs forced deletion on the caught-up future log with the current log's
remote storage coverage. Promotion waits while the future log still holds
prefix records that the current log no longer serves locally.
Known limitations:
- If a replica with a segment that spans the seal leads the partition, a
follower below the seal can't fetch that part of the prefix until local
retention removes the segment. A log directory move on such a replica
waits for the same deletion.
- If remote storage or `remote.log.copy.disable` keeps the broker from
copying, the broker keeps the prefix and logs a warning. Consumer reads
succeed only for offsets that this broker already sees in remote storage.
`UnifiedLog.java` and `ReplicaAlterLogDirsThread.scala` become interleaved
paths in `INKLESS_OWNERSHIP`.
## Testing
- `UnifiedLogTest` cases for the log start offset, the epoch cache, the
active-segment roll, the remote-copy precondition, and segments that
remote storage doesn't cover.
- `ReplicaManagerInklessTest` cases for the leader and follower deletion on
the flag commit, the deferred deletion and its retry, the read guard in
both phases with and without purgatory, the tiered path for a consumer read
of a local prefix that remote storage covers, a phase 2 read below the seal
that returns nothing, a storage error during the deletion, the promotion
decision, and a log directory move that spans the flag commit.
- `ReplicaFetcherThreadTest` checks that eviction deletes the prefix before
consolidation starts.
- `ReplicaAlterLogDirsThreadTest` checks that a delayed promotion keeps the
partition on the thread, and that the thread rolls the future log at the
seal before it appends.
- `make vale` doesn't report any alerts.
Assisted-by: Claude Opus 5.5
…KC-589]
Phase 2 deletes the local classic prefix and serves those offsets from remote
storage. The no-op remote storage manager reports the copy as complete and
doesn't keep the bytes. The switched topic is then unreadable, and the cluster
test's consumer receives an empty result.
The switcher cluster uses the same tiered storage stack as the consolidated
diskless tests. The test waits until every broker deletes its local prefix
before it consumes. `diskless.enable` commits before the prefix upload, so
without the wait the consumer can read the whole prefix locally in phase 1.
## Testing
- `InklessTopicTypeSwitcherClusterTest` passes for both alter-configs modes
and for the other four cases in the class.
- With the no-op remote storage manager, `switchClassicTopicToDiskless`
doesn't consume any record from the switched topic and fails.
- Vale doesn't report any alerts on the added comments.
Assisted-by: Grok 4.7
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
…fix state [KC-589]
The phase-based names describe only the classic-to-diskless switch. A later
diskless-to-classic switch would reuse the same checks with different phases.
The checks now say where the classic prefix lives:
- `isSwitchedWithLocalClassicPrefix` becomes `isClassicPrefixOnlyOnLocalDisks`
in `MetadataView`, `InklessMetadataView`, and `ReplicationControlManager`.
- `isSwitchedPhaseOne` becomes `isClassicPrefixWaitingForUpload`.
- `isBelowSealInPhaseTwo` becomes `isClassicPrefixOffsetInRemoteStorage`.
- The topic overload of `consolidationActiveFor` becomes
`consolidationEnabledFor`. It checks only the topic config. The partition
overload keeps its name and adds the phase check.
The read-path check drops its `isDisklessTopic` call. A committed seal and the
`classicPrefixInRemoteStorage` flag already imply a diskless topic, and
`isDisklessTopic` builds a `Properties` object for every fetched partition.
`ReplicaManagerTest` mocks `KRaftMetadataCache`, so its caches now stub
`currentImage()` with `MetadataImage.EMPTY`, the production default.
`ReplicaFetcherThread` calls `ReplicaManager.consolidationActiveFor` instead of
repeating the gate.
The broker logs two more events:
- When it marks a leader below the seal offline, the error says why. In phase
1 with consolidation and remote storage, it tells the operator to move
leadership to a replica that holds the prefix. Otherwise it says that remote
storage can't rebuild the prefix on this broker.
- When a read below the seal finds local records in phase 2, the broker logs
that it queued the deletion of the local classic prefix.
The comments and Scaladoc that the switch-phase gate adds say
"classic-to-diskless switch" instead of "switch".
- `ReplicaManagerInklessTest` adds two cases:
- A consolidating follower at the seal outside the ISR stays on the classic
fetcher in phase 1. The case fails if `isReadyForConsolidation` stops
checking `classicPrefixInRemoteStorage`.
- A follower that goes offline during truncation in phase 2 starts no
fetcher. The case fails without the `partitionsToStartFetching` filter.
- `ReplicaManagerTest`, `ReplicaManagerInklessTest`, `ReplicaFetcherThreadTest`,
`ConsolidationReconcilerTest`, `InklessMetadataViewTest`,
`InklessTopicMetadataTransformerTest`, `ReplicationControlManagerTest`,
`DisklessSwitchFlowTest`, and `DisklessSwitchInvariantsTest` pass.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
* feat(inkless:switch): gate consolidation, fetchers, and routing on the switch phase [KC-589]
A committed seal doesn't mean the classic prefix is safe to drop from local
disks. Until the controller commits `classicPrefixInRemoteStorage`, the local
logs of the ISR members are the only complete copy. A replica that starts
consolidating in phase 1 advances the cross-tier log start offset over a
prefix that remote storage doesn't hold yet. A client routed to a non-ISR
replica can read a truncated prefix.
The broker now treats phase 1 like a switched partition on a topic that
doesn't consolidate:
- The consolidation gate is per partition. `consolidationActiveFor(tp)` is
off in phase 1, and `isReadyForConsolidation` and the reconciler wait for
the flag.
- The classic fetcher evicts a consolidating partition at the seal without
waiting for the ISR only in phase 2. In phase 1 it applies the ISR rule of
a topic that doesn't consolidate. The flag commit bumps the leader epoch,
so the follower path re-runs and starts consolidation.
- The broker fences a leader whose local log ends below the seal. Nothing
rebuilds the prefix in phase 1, so only phase 2 keeps that leader online
for consolidation to rebuild the prefix from remote storage.
- The metadata transformer advertises only alive ISR members while the
switch is pending and in phase 1.
`InklessMetadataView` exposes the flag, along with `isSwitchedPhaseOne` and
`isSwitchedWithLocalClassicPrefix`.
- New and updated tests in `ReplicaManagerInklessTest`,
`ReplicaFetcherThreadTest`, `ConsolidationReconcilerTest`,
`InklessMetadataViewTest`, and `InklessTopicMetadataTransformerTest`.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* feat(inkless:switch): delete the local classic prefix when phase 2 starts [KC-589]
In the record that starts phase 2, the controller adds every live replica to
the ISR without checking its local copy of the classic prefix. A replica that
was below the seal, or that diverged, can then serve or consolidate from a
local prefix that nobody validated. From this record on, brokers trust only
the remote copy.
When a replica applies `classicPrefixInRemoteStorage`, it
10BC8
deletes every local
segment below the seal through `UnifiedLog.forceDeleteLocalSegmentsBelow`. If
the active segment contains the end of the prefix, the replica rolls it first.
The deletion skips the retention checks. It advances the local log start
offset and keeps the log start offset and the leader epoch cache, so reads
below the seal take the tiered path, and the leader can still validate
follower fetches. The classic fetcher runs the same idempotent step when it
evicts a partition at the seal. That covers a follower that fetched part of
the prefix after it applied the flag.
The leader requests the flag from its own view of remote storage. A
follower's remote log metadata can trail it. A tiered read of a segment that
the metadata doesn't list fails with `OFFSET_OUT_OF_RANGE` and resets the
consumer. So forced deletion removes only the segments that this
broker's highest offset in remote storage covers, and `ReplicaManager`
retries the rest once per second.
`readFromLog` never serves local records below the seal in phase 2. If this
broker's highest offset in remote storage covers the offset, a consumer read
takes the tiered path. Otherwise, and for followers, the broker answers
`NOT_LEADER_OR_FOLLOWER`, and the client retries. The check runs on the
records that the read returned, so one rule covers both fetch branches, a
fetch that waited in purgatory since phase 1, a read that races the
deletion, and a log directory move that swaps the log. A read that hits the
local prefix also queues the deletion again.
The tiered path matters most after an upgrade. On earlier releases, a
follower at the seal started consolidating in phase 1 and appended the first
diskless records to the segment that ends the prefix. Forced deletion can't
remove that segment, so without the tiered path, consumers routed to that
follower would get errors until local retention deletes it.
A consumer read below the seal never takes a diskless supplement. That read
returns nothing when the request's byte budget is already spent, and a
supplement from the local log end offset would skip the rest of the prefix.
If the deletion fails with a storage error, the broker reports the log
directory to the log directory failure handler instead of marking the
partition offline. The classic fetcher thread runs the deletion when it
evicts a partition, and `applyDelta` waits for idle fetcher threads to stop
while it holds `replicaStateChangeLock`. Taking that lock from the fetcher
thread could deadlock. The failure handler also stops the partition's
fetchers and notifies the controller.
A log directory move copies the prefix into the future log, and the flag
commit deletes it only from the current log. The alter-log-dirs thread rolls
the future log before it copies the first record at the seal. Without the
roll, one segment would contain both the end of the prefix and the start of the
diskless records, and forced deletion couldn't remove that segment. Before it
promotes, the thread asks `shouldDelayFutureReplicaPromotion`. That check
runs forced deletion on the caught-up future log with the current log's
remote storage coverage. Promotion waits while the future log still holds
prefix records that the current log no longer serves locally.
Known limitations:
- If a replica with a segment that spans the seal leads the partition, a
follower below the seal can't fetch that part of the prefix until local
retention removes the segment. A log directory move on such a replica
waits for the same deletion.
- If remote storage or `remote.log.copy.disable` keeps the broker from
copying, the broker keeps the prefix and logs a warning. Consumer reads
succeed only for offsets that this broker already sees in remote storage.
`UnifiedLog.java` and `ReplicaAlterLogDirsThread.scala` become interleaved
paths in `INKLESS_OWNERSHIP`.
## Testing
- `UnifiedLogTest` cases for the log start offset, the epoch cache, the
active-segment roll, the remote-copy precondition, and segments that
remote storage doesn't cover.
- `ReplicaManagerInklessTest` cases for the leader and follower deletion on
the flag commit, the deferred deletion and its retry, the read guard in
both phases with and without purgatory, the tiered path for a consumer read
of a local prefix that remote storage covers, a phase 2 read below the seal
that returns nothing, a storage error during the deletion, the promotion
decision, and a log directory move that spans the flag commit.
- `ReplicaFetcherThreadTest` checks that eviction deletes the prefix before
consolidation starts.
- `ReplicaAlterLogDirsThreadTest` checks that a delayed promotion keeps the
partition on the thread, and that the thread rolls the future log at the
seal before it appends.
- `make vale` doesn't report any alerts.
Assisted-by: Claude Opus 5.5
* test(inkless:switch): serve the switched prefix from tiered storage [KC-589]
Phase 2 deletes the local classic prefix and serves those offsets from remote
storage. The no-op remote storage manager reports the copy as complete and
doesn't keep the bytes. The switched topic is then unreadable, and the cluster
test's consumer receives an empty result.
The switcher cluster uses the same tiered storage stack as the consolidated
diskless tests. The test waits until every broker deletes its local prefix
before it consumes. `diskless.enable` commits before the prefix upload, so
without the wait the consumer can read the whole prefix locally in phase 1.
## Testing
- `InklessTopicTypeSwitcherClusterTest` passes for both alter-configs modes
and for the other four cases in the class.
- With the no-op remote storage manager, `switchClassicTopicToDiskless`
doesn't consume any record from the switched topic and fails.
- Vale doesn't report any alerts on the added comments.
Assisted-by: Grok 4.7
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* refactor(inkless:switch): name the phase checks after the classic prefix state [KC-589]
The phase-based names describe only the classic-to-diskless switch. A later
diskless-to-classic switch would reuse the same checks with different phases.
The checks now say where the classic prefix lives:
- `isSwitchedWithLocalClassicPrefix` becomes `isClassicPrefixOnlyOnLocalDisks`
in `MetadataView`, `InklessMetadataView`, and `ReplicationControlManager`.
- `isSwitchedPhaseOne` becomes `isClassicPrefixWaitingForUpload`.
- `isBelowSealInPhaseTwo` becomes `isClassicPrefixOffsetInRemoteStorage`.
- The topic overload of `consolidationActiveFor` becomes
`consolidationEnabledFor`. It checks only the topic config. The partition
overload keeps its name and adds the phase check.
The read-path check drops its `isDisklessTopic` call. A committed seal and the
`classicPrefixInRemoteStorage` flag already imply a diskless topic, and
`isDisklessTopic` builds a `Properties` object for every fetched partition.
`ReplicaManagerTest` mocks `KRaftMetadataCache`, so its caches now stub
`currentImage()` with `MetadataImage.EMPTY`, the production default.
`ReplicaFetcherThread` calls `ReplicaManager.consolidationActiveFor` instead of
repeating the gate.
The broker logs two more events:
- When it marks a leader below the seal offline, the error says why. In phase
1 with consolidation and remote storage, it tells the operator to move
leadership to a replica that holds the prefix. Otherwise it says that remote
storage can't rebuild the prefix on this broker.
- When a read below the seal finds local records in phase 2, the broker logs
that it queued the deletion of the local classic prefix.
The comments and Scaladoc that the switch-phase gate adds say
"classic-to-diskless switch" instead of "switch".
- `ReplicaManagerInklessTest` adds two cases:
- A consolidating follower at the seal outside the ISR stays on the classic
fetcher in phase 1. The case fails if `isReadyForConsolidation` stops
checking `classicPrefixInRemoteStorage`.
- A follower that goes offline during truncation in phase 2 starts no
fetcher. The case fails without the `partitionsToStartFetching` filter.
- `ReplicaManagerTest`, `ReplicaManagerInklessTest`, `ReplicaFetcherThreadTest`,
`ConsolidationReconcilerTest`, `InklessMetadataViewTest`,
`InklessTopicMetadataTransformerTest`, `ReplicationControlManagerTest`,
`DisklessSwitchFlowTest`, and `DisklessSwitchInvariantsTest` pass.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
---------
Co-authored-by: Cursor <cursoragent@cursor.com>
gqmelo
pushed a commit
that referenced
this pull request
Nov 5, 2026
* feat(inkless:switch): gate consolidation, fetchers, and routing on the switch phase [KC-589]
A committed seal doesn't mean the classic prefix is safe to drop from local
disks. Until the controller commits `classicPrefixInRemoteStorage`, the local
logs of the ISR members are the only complete copy. A replica that starts
consolidating in phase 1 advances the cross-tier log start offset over a
prefix that remote storage doesn't hold yet. A client routed to a non-ISR
replica can read a truncated prefix.
The broker now treats phase 1 like a switched partition on a topic that
doesn't consolidate:
- The consolidation gate is per partition. `consolidationActiveFor(tp)` is
off in phase 1, and `isReadyForConsolidation` and the reconciler wait for
the flag.
- The classic fetcher evicts a consolidating partition at the seal without
waiting for the ISR only in phase 2. In phase 1 it applies the ISR rule of
a topic that doesn't consolidate. The flag commit bumps the leader epoch,
so the follower path re-runs and starts consolidation.
- The broker fences a leader whose local log ends below the seal. Nothing
rebuilds the prefix in phase 1, so only phase 2 keeps that leader online
for consolidation to rebuild the prefix from remote storage.
- The metadata transformer advertises only alive ISR members while the
switch is pending and in phase 1.
`InklessMetadataView` exposes the flag, along with `isSwitchedPhaseOne` and
`isSwitchedWithLocalClassicPrefix`.
- New and updated tests in `ReplicaManagerInklessTest`,
`ReplicaFetcherThreadTest`, `ConsolidationReconcilerTest`,
`InklessMetadataViewTest`, and `InklessTopicMetadataTransformerTest`.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* feat(inkless:switch): delete the local classic prefix when phase 2 starts [KC-589]
In the record that starts phase 2, the controller adds every live replica to
the ISR without checking its local copy of the classic prefix. A replica that
was below the seal, or that diverged, can then serve or consolidate from a
local prefix that nobody validated. From this record on, brokers trust only
the remote copy.
When a replica applies `classicPrefixInRemoteStorage`, it deletes every local
segment below the seal through `UnifiedLog.forceDeleteLocalSegmentsBelow`. If
the active segment contains the end of the prefix, the replica rolls it first.
The deletion skips the retention checks. It advances the local log start
offset and keeps the log start offset and the leader epoch cache, so reads
below the seal take the tiered path, and the leader can still validate
follower fetches. The classic fetcher runs the same idempotent step when it
evicts a partition at the seal. That covers a follower that fetched part of
the prefix after it applied the flag.
The leader requests the flag from its own view of remote storage. A
follower's remote log metadata can trail it. A tiered read of a segment that
the metadata doesn't list fails with `OFFSET_OUT_OF_RANGE` and resets the
consumer. So forced deletion removes only the segments that this
broker's highest offset in remote storage covers, and `ReplicaManager`
retries the rest once per second.
`readFromLog` never serves local records below the seal in phase 2. If this
broker's highest offset in remote storage covers the offset, a consumer read
takes the tiered path. Otherwise, and for followers, the broker answers
`NOT_LEADER_OR_FOLLOWER`, and the client retries. The check runs on the
records that the read returned, so one rule covers both fetch branches, a
fetch that waited in purgatory since phase 1, a read that races the
deletion, and a log directory move that swaps the log. A read that hits the
local prefix also queues the deletion again.
The tiered path matters most after an upgrade. On earlier releases, a
follower at the seal started consolidating in phase 1 and appended the first
diskless records to the segment that ends the prefix. Forced deletion can't
remove that segment, so without the tiered path, consumers routed to that
follower would get errors until local retention deletes it.
A consumer read below the seal never takes a diskless supplement. That read
returns nothing when the request's byte budget is already spent, and a
supplement from the local log end offset would skip the rest of the prefix.
If the deletion fails with a storage error, the broker reports the log
directory to the log directory failure handler instead of marking the
partition offline. The classic fetcher thread runs the deletion when it
evicts a partition, and `applyDelta` waits for idle fetcher threads to stop
while it holds `replicaStateChangeLock`. Taking that lock from the fetcher
thread could deadlock. The failure handler also stops the partition's
fetchers and notifies the controller.
A log directory move copies the prefix into the future log, and the flag
commit deletes it only from the current log. The alter-log-dirs thread rolls
the future log before it copies the first record at the seal. Without the
roll, one segment would contain both the end of the prefix and the start of the
diskless records, and forced deletion couldn't remove that segment. Before it
promotes, the thread asks `shouldDelayFutureReplicaPromotion`. That check
runs forced deletion on the caught-up future log with the current log's
remote storage coverage. Promotion waits while the future log still holds
prefix records that the current log no longer serves locally.
Known limitations:
- If a replica with a segment that spans the seal leads the partition, a
follower below the seal can't fetch that part of the prefix until local
retention removes the segment. A log directory move on such a replica
waits for the same deletion.
- If remote storage or `remote.log.copy.disable` keeps the broker from
copying, the broker keeps the prefix and logs a warning. Consumer reads
succeed only for offsets that this broker already sees in remote storage.
`UnifiedLog.java` and `ReplicaAlterLogDirsThread.scala` become interleaved
paths in `INKLESS_OWNERSHIP`.
- `UnifiedLogTest` cases for the log start offset, the epoch cache, the
active-segment roll, the remote-copy precondition, and segments that
remote storage doesn't cover.
- `ReplicaManagerInklessTest` cases for the leader and follower deletion on
the flag commit, the deferred deletion and its retry, the read guard in
both phases with and without purgatory, the tiered path for a consumer read
of a local prefix that remote storage covers, a phase 2 read below the seal
that returns nothing, a storage error during the deletion, the promotion
decision, and a log directory move that spans the flag commit.
- `ReplicaFetcherThreadTest` checks that eviction deletes the prefix before
consolidation starts.
- `ReplicaAlterLogDirsThreadTest` checks that a delayed promotion keeps the
partition on the thread, and that the thread rolls the future log at the
seal before it appends.
- `make vale` doesn't report any alerts.
Assisted-by: Claude Opus 5.5
* test(inkless:switch): serve the switched prefix from tiered storage [KC-589]
Phase 2 deletes the local classic prefix and serves those offsets from remote
storage. The no-op remote storage manager reports the copy as complete and
doesn't keep the bytes. The switched topic is then unreadable, and the cluster
test's consumer receives an empty result.
The switcher cluster uses the same tiered storage stack as the consolidated
diskless tests. The test waits until every broker deletes its local prefix
before it consumes. `diskless.enable` commits before the prefix upload, so
without the wait the consumer can read the whole prefix locally in phase 1.
- `InklessTopicTypeSwitcherClusterTest` passes for both alter-configs modes
and for the other four cases in the class.
- With the no-op remote storage manager, `switchClassicTopicToDiskless`
doesn't consume any record from the switched topic and fails.
- Vale doesn't report any alerts on the added comments.
Assisted-by: Grok 4.7
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* refactor(inkless:switch): name the phase checks after the classic prefix state [KC-589]
The phase-based names describe only the classic-to-diskless switch. A later
diskless-to-classic switch would reuse the same checks with different phases.
The checks now say where the classic prefix lives:
- `isSwitchedWithLocalClassicPrefix` becomes `isClassicPrefixOnlyOnLocalDisks`
in `MetadataView`, `InklessMetadataView`, and `ReplicationControlManager`.
- `isSwitchedPhaseOne` becomes `isClassicPrefixWaitingForUpload`.
- `isBelowSealInPhaseTwo` becomes `isClassicPrefixOffsetInRemoteStorage`.
- The topic overload of `consolidationActiveFor` becomes
`consolidationEnabledFor`. It checks only the topic config. The partition
overload keeps its name and adds the phase check.
The read-path check drops its `isDisklessTopic` call. A committed seal and the
`classicPrefixInRemoteStorage` flag already imply a diskless topic, and
`isDisklessTopic` builds a `Properties` object for every fetched partition.
`ReplicaManagerTest` mocks `KRaftMetadataCache`, so its caches now stub
`currentImage()` with `MetadataImage.EMPTY`, the production default.
`ReplicaFetcherThread` calls `ReplicaManager.consolidationActiveFor` instead of
repeating the gate.
The broker logs two more events:
- When it marks a leader below the seal offline, the error says why. In phase
1 with consolidation and remote storage, it tells the operator to move
leadership to a replica that holds the prefix. Otherwise it says that remote
storage can't rebuild the prefix on this broker.
- When a read below the seal finds local records in phase 2, the broker logs
that it queued the deletion of the local classic prefix.
The comments and Scaladoc that the switch-phase gate adds say
"classic-to-diskless switch" instead of "switch".
- `ReplicaManagerInklessTest` adds two cases:
- A consolidating follower at the seal outside the ISR stays on the classic
fetcher in phase 1. The case fails if `isReadyForConsolidation` stops
checking `classicPrefixInRemoteStorage`.
- A follower that goes offline during truncation in phase 2 starts no
fetcher. The case fails without the `partitionsToStartFetching` filter.
- `ReplicaManagerTest`, `ReplicaManagerInklessTest`, `ReplicaFetcherThreadTest`,
`ConsolidationReconcilerTest`, `InklessMetadataViewTest`,
`InklessTopicMetadataTransformerTest`, `ReplicationControlManagerTest`,
`DisklessSwitchFlowTest`, and `DisklessSwitchInvariantsTest` pass.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
---------
Co-authored-by: Cursor <cursoragent@cursor.com>
On 4.2, `MetadataDelta` has no public constructor, only `MetadataDelta.Builder`. `HostedPartition.Online` and `HostedPartition.Offline` take no type parameter, `MemoryRecords` and `SimpleRecord` are in `org.apache.kafka.common.record`, and the Scala `BrokerReconfigurable` returns a Scala set. The tests added by #835, #840, #841, #850, and #852 used the main signatures.
The Java `UnifiedLogTest` on 4.2 is a stub, and the log tests still live in the Scala `kafka.log.UnifiedLogTest`. The cherry-pick of #840 ports its four `forceDeleteLocalSegmentsBelow` tests to the Scala suite instead of pulling the upstream Java tests in through the conflict.
Assisted-by: Claude Opus 5.5
gqmelo
pushed a commit
that referenced
this pull request
Nov 5, 2026
* feat(inkless:switch): gate consolidation, fetchers, and routing on the switch phase [KC-589]
A committed seal doesn't mean the classic prefix is safe to drop from local
disks. Until the controller commits `classicPrefixInRemoteStorage`, the local
logs of the ISR members are the only complete copy. A replica that starts
consolidating in phase 1 advances the cross-tier log start offset over a
prefix that remote storage doesn't hold yet. A client routed to a non-ISR
replica can read a truncated prefix.
The broker now treats phase 1 like a switched partition on a topic that
doesn't consolidate:
- The consolidation gate is per partition. `consolidationActiveFor(tp)` is
off in phase 1, and `isReadyForConsolidation` and the reconciler wait for
the flag.
- The classic fetcher evicts a consolidating partition at the seal without
waiting for the ISR only in phase 2. In phase 1 it applies the ISR rule of
a topic that doesn't consolidate. The flag commit bumps the leader epoch,
so the follower path re-runs and starts consolidation.
- The broker fences a leader whose local log ends below the seal. Nothing
rebuilds the prefix in phase 1, so only phase 2 keeps that leader online
for consolidation to rebuild the prefix from remote storage.
- The metadata transformer advertises only alive ISR members while the
switch is pending and in phase 1.
`InklessMetadataView` exposes the flag, along with `isSwitchedPhaseOne` and
`isSwitchedWithLocalClassicPrefix`.
- New and updated tests in `ReplicaManagerInklessTest`,
`ReplicaFetcherThreadTest`, `ConsolidationReconcilerTest`,
`InklessMetadataViewTest`, and `InklessTopicMetadataTransformerTest`.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* feat(inkless:switch): delete the local classic prefix when phase 2 starts [KC-589]
In the record that starts phase 2, the controller adds every live replica to
the ISR without checking its local copy of the classic prefix. A replica that
was below the seal, or that diverged, can then serve or consolidate from a
local prefix that nobody validated. From this record on, brokers trust only
the remote copy.
When a replica applies `classicPrefixInRemoteStorage`, it deletes every local
segment below the seal through `UnifiedLog.forceDeleteLocalSegmentsBelow`. If
the active segment contains the end of the prefix, the replica rolls it first.
The deletion skips the retention checks. It advances the local log start
offset and keeps the log start offset and the leader epoch cache, so reads
below the seal take the tiered path, and the leader can still validate
follower fetches. The classic fetcher runs the same idempotent step when it
evicts a partition at the seal. That covers a follower that fetched part of
the prefix after it applied the flag.
The leader requests the flag from its own view of remote storage. A
follower's remote log metadata can trail it. A tiered read of a segment that
the metadata doesn't list fails with `OFFSET_OUT_OF_RANGE` and resets the
consumer. So forced deletion removes only the segments that this
broker's highest offset in remote storage covers, and `ReplicaManager`
retries the rest once per second.
`readFromLog` never serves local records below the seal in phase 2. If this
broker's highest offset in remote storage covers the offset, a consumer read
takes the tiered path. Otherwise, and for followers, the broker answers
`NOT_LEADER_OR_FOLLOWER`, and the client retries. The check runs on the
records that the read returned, so one rule covers both fetch branches, a
fetch that waited in purgatory since phase 1, a read that races the
deletion, and a log directory move that swaps the log. A read that hits the
local prefix also queues the deletion again.
The tiered path matters most after an upgrade. On earlier releases, a
follower at the seal started consolidating in phase 1 and appended the first
diskless records to the segment that ends the prefix. Forced deletion can't
remove that segment, so without the tiered path, consumers routed to that
follower would get errors until local retention deletes it.
A consumer read below the seal never takes a diskless supplement. That read
returns nothing when the request's byte budget is already spent, and a
supplement from the local log end offset would skip the rest of the prefix.
If the deletion fails with a storage error, the broker reports the log
directory to the log directory failure handler instead of marking the
partition offline. The classic fetcher thread runs the deletion when it
evicts a partition, and `applyDelta` waits for idle fetcher threads to stop
while it holds `replicaStateChangeLock`. Taking that lock from the fetcher
thread could deadlock. The failure handler also stops the partition's
fetchers and notifies the controller.
A log directory move copies the prefix into the future log, and the flag
commit deletes it only from the current log. The alter-log-dirs thread rolls
the future log before it copies the first record at the seal. Without the
roll, one segment would contain both the end of the prefix and the start of the
diskless records, and forced deletion couldn't remove that segment. Before it
promotes, the thread asks `shouldDelayFutureReplicaPromotion`. That check
runs forced deletion on the caught-up future log with the current log's
remote storage coverage. Promotion waits while the future log still holds
prefix records that the current log no longer serves locally.
Known limitations:
- If a replica with a segment that spans the seal leads the partition, a
follower below the seal can't fetch that part of the prefix until local
retention removes the segment. A log directory move on such a replica
waits for the same deletion.
- If remote storage or `remote.log.copy.disable` keeps the broker from
copying, the broker keeps the prefix and logs a warning. Consumer reads
succeed only for offsets that this broker already sees in remote storage.
`UnifiedLog.java` and `ReplicaAlterLogDirsThread.scala` become interleaved
paths in `INKLESS_OWNERSHIP`.
- `UnifiedLogTest` cases for the log start offset, the epoch cache, the
active-segment roll, the remote-copy precondition, and segments that
remote storage doesn't cover.
- `ReplicaManagerInklessTest` cases for the leader and follower deletion on
the flag commit, the deferred deletion and its retry, the read guard in
both phases with and without purgatory, the tiered path for a consumer read
of a local prefix that remote storage covers, a phase 2 read below the seal
that returns nothing, a storage error during the deletion, the promotion
decision, and a log directory move that spans the flag commit.
- `ReplicaFetcherThreadTest` checks that eviction deletes the prefix before
consolidation starts.
- `ReplicaAlterLogDirsThreadTest` checks that a delayed promotion keeps the
partition on the thread, and that the thread rolls the future log at the
seal before it appends.
- `make vale` doesn't report any alerts.
Assisted-by: Claude Opus 5.5
* test(inkless:switch): serve the switched prefix from tiered storage [KC-589]
Phase 2 deletes the local classic prefix and serves those offsets from remote
storage. The no-op remote storage manager reports the copy as complete and
doesn't keep the bytes. The switched topic is then unreadable, and the cluster
test's consumer receives an empty result.
The switcher cluster uses the same tiered storage stack as the consolidated
diskless tests. The test waits until every broker deletes its local prefix
before it consumes. `diskless.enable` commits before the prefix upload, so
without the wait the consumer can read the whole prefix locally in phase 1.
- `InklessTopicTypeSwitcherClusterTest` passes for both alter-configs modes
and for the other four cases in the class.
- With the no-op remote storage manager, `switchClassicTopicToDiskless`
doesn't consume any record from the switched topic and fails.
- Vale doesn't report any alerts on the added comments.
Assisted-by: Grok 4.7
Assisted-by: Claude Opus 5.5
Co-authored-by: Cursor <cursoragent@cursor.com>
* refactor(inkless:switch): name the phase checks after the classic prefix state [KC-589]
The phase-based names describe only the classic-to-diskless switch. A later
diskless-to-classic switch would reuse the same checks with different phases.
The checks now say where the classic prefix lives:
- `isSwitchedWithLocalClassicPrefix` becomes `isClassicPrefixOnlyOnLocalDisks`
in `MetadataView`, `InklessMetadataView`, and `ReplicationControlManager`.
- `isSwitchedPhaseOne` becomes `isClassicPrefixWaitingForUpload`.
- `isBelowSealInPhaseTwo` becomes `isClassicPrefixOffsetInRemoteStorage`.
- The topic overload of `consolidationActiveFor` becomes
`consolidationEnabledFor`. It checks only the topic config. The partition
overload keeps its name and adds the phase check.
The read-path check drops its `isDisklessTopic` call. A committed seal and the
`classicPrefixInRemoteStorage` flag already imply a diskless topic, and
`isDisklessTopic` builds a `Properties` object for every fetched partition.
`ReplicaManagerTest` mocks `KRaftMetadataCache`, so its caches now stub
`currentImage()` with `MetadataImage.EMPTY`, the production default.
`ReplicaFetcherThread` calls `ReplicaManager.consolidationActiveFor` instead of
repeating the gate.
The broker logs two more events:
- When it marks a leader below the seal offline, the error says why. In phase
1 with consolidation and remote storage, it tells the operator to move
leadership to a replica that holds the prefix. Otherwise it says that remote
storage can't rebuild the prefix on this broker.
- When a read below the seal finds local records in phase 2, the broker logs
that it queued the deletion of the local classic prefix.
The comments and Scaladoc that the switch-phase gate adds say
"classic-to-diskless switch" instead of "switch".
- `ReplicaManagerInklessTest` adds two cases:
- A consolidating follower at the seal outside the ISR stays on the classic
fetcher in phase 1. The case fails if `isReadyForConsolidation` stops
checking `classicPrefixInRemoteStorage`.
- A follower that goes offline during truncation in phase 2 starts no
fetcher. The case fails without the `partitionsToStartFetching` filter.
- `ReplicaManagerTest`, `ReplicaManagerInklessTest`, `ReplicaFetcherThreadTest`,
`ConsolidationReconcilerTest`, `InklessMetadataViewTest`,
`InklessTopicMetadataTransformerTest`, `ReplicationControlManagerTest`,
`DisklessSwitchFlowTest`, and `DisklessSwitchInvariantsTest` pass.
- `make vale` reports no alerts.
Assisted-by: Claude Opus 5.5
---------
Co-authored-by: Cursor <cursoragent@cursor.com>
On 4.1, `RemoteStorageFetchInfo` carries a `TopicPartition` instead of a `TopicIdPartition`, so the remote fetch callback in `ReplicaManager` looks up the topic ID with `topicIdPartition`. `LogReadResult` takes no `Errors` argument, `RetriableRemoteStorageException` doesn't exist and the reclaim floor throws `RemoteStorageException`, and `remote.log.manager.thread.pool.size` sizes the follower pool. `HostedPartition.Online` and `HostedPartition.Offline` take no type parameter, `MemoryRecords` and `SimpleRecord` are in `org.apache.kafka.common.record`, and the Scala `BrokerReconfigurable` returns a Scala set. The changes added by #831, #834, #835, #840, #841, #847, #850, and #852 used the main signatures.
The Java `UnifiedLogTest` on 4.1 is a stub, and the log tests still live in the Scala `kafka.log.UnifiedLogTest`. The cherry-pick of #840 ports its four `forceDeleteLocalSegmentsBelow` tests to the Scala suite instead of pulling the upstream Java tests in through the conflict.
After #847, `kafka.server.ReplicaManager` on 4.1 is too big for SpotBugs to analyze, and `:core:spotbugsMain` fails with `SKIPPED_CLASS_TOO_BIG`. The class is about 400 lines longer on 4.1 than on 4.2 and 4.3, and SpotBugs 4.9.8 fails the same way. The exclude filter now skips that pattern for `ReplicaManager` only.
Assisted-by: Claude Opus 5.5
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
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.
A committed seal doesn't mean the classic prefix is safe to drop from local disks. Until the controller commits
classicPrefixInRemoteStorage, the local logs of the ISR members store the only complete copy. In phase 1, a replica that starts consolidating advances the cross-tier log start offset over a prefix that remote storage doesn't hold yet. A client routed to a replica outside the ISR can read a truncated prefix.In phase 2 the opposite holds. In the record that sets the flag, the controller adds every live replica to the ISR without checking its local copy. From then on, brokers trust only the remote copy.
Phase 1: treat the partition as not consolidating
consolidationActiveFor(tp)gates consolidation per partition and is off in phase 1.isReadyForConsolidationandConsolidationReconcilerwait for the flag.InklessMetadataViewexposes the flag,isSwitchedPhaseOne, andisSwitchedWithLocalClassicPrefix.Phase 2: delete the local classic prefix
UnifiedLog.forceDeleteLocalSegmentsBelow. If the active segment contains the end of the prefix, the replica rolls it first. The deletion skips the retention checks. It advances the local log start offset and keeps the log start offset and the leader epoch cache. Reads below the seal take the tiered path, and the leader can still validate follower fetches.ReplicaManagerretries the rest once per second.readFromLognever serves local records below the seal. If this broker's highest offset in remote storage covers the offset, a consumer read takes the tiered path. Otherwise, and for followers, the broker answersNOT_LEADER_OR_FOLLOWER, and the client retries. The check runs on the records that the read returned, so one rule covers both fetch branches, a fetch that waited in purgatory since phase 1, a read that races the deletion, and a log directory move that swaps the log. A read that hits the local prefix also queues the deletion again.shouldDelayFutureReplicaPromotionruns forced deletion on the caught-up future log and delays promotion while that log still holds prefix records.applyDeltawaits for idle fetcher threads to stop while it holdsreplicaStateChangeLock, so taking that lock from the fetcher thread could deadlock.UnifiedLog.javaandReplicaAlterLogDirsThread.scalabecome interleaved paths inINKLESS_OWNERSHIP.Known limitations
remote.log.copy.disableistrue, the broker keeps the prefix and logs a warning. Consumer reads succeed only for offsets that this broker already sees in remote storage.DisklessFetchOffsetRouterand the pending-state handling as they are. ISR shrink (KC-633),DeleteRecords, andRetentionEnforcerare out of scope.Testing
ReplicaManagerInklessTestcovers per-partition gating, the fence of a phase 1 leader below the seal, the phase 1 to phase 2 hand-off for leaders and followers, deletion on the flag commit and its retry, the read check in both phases with and without purgatory, the tiered path for a consumer read of a local prefix that remote storage covers, a phase 2 read below the seal that returns nothing, a storage error during the deletion, the promotion decision, and a log directory move that spans the flag commit.ReplicaFetcherThreadTestchecks that a consolidating follower at the seal waits for the ISR in phase 1, and that eviction deletes the prefix before consolidation starts.ConsolidationReconcilerTestcovers the phase 1 retry.UnifiedLogTestchecks that forced deletion keeps the log start offset and the epoch cache. Other cases cover the active-segment roll, the remote-copy precondition, and segments that remote storage doesn't cover.ReplicaAlterLogDirsThreadTestchecks that a delayed promotion keeps the partition on the thread, and that the thread rolls the future log at the seal before it appends.InklessMetadataViewTestandInklessTopicMetadataTransformerTestcover the phase predicates and ISR-only routing.InklessTopicTypeSwitcherClusterTestruns on the tiered storage stack and waits until every broker deletes its local prefix before it consumes. The consumer then reads the switched prefix from remote storage. With the no-op remote storage manager, the switch test fails.ReplicaManagerTest,DelayedFetchTest,DisklessSwitchFlowTest,InitDisklessLogManagerTest, andDisklessFetchOffsetRouterTestpass unchanged.make valedoesn't report any alerts.