fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] by giuseppelillo · Pull Request #857 · aiven/inkless

giuseppelillo · 2026-11-05T12:02:02Z

In phase 2 of the classic-to-diskless switch, the controller adds every live replica to the ISR without checking its local copy of the classic prefix, and any replica can lead. A replica can hold records below the seal that the partition doesn't hold. It led an older leader epoch and wrote records that the next leader didn't have, or an unclean leader election before the switch discarded records below its high watermark. Forced deletion keeps the leader epoch cache of that replica, and the remote log manager uses it to compute the highest offset in remote storage and to pick remote segments. On that replica, a tiered read below the seal fails with OFFSET_OUT_OF_RANGE, and the consumer resets its offset. On an idle topic, the highest offset in remote storage also stays below the seal, so forced deletion never completes.

Example

Broker B leads epoch 1 and appends [80, 100), but no follower fetches those records before B goes offline. The controller elects A in epoch 2, and A appends its own [80, 100). A seals the partition at 100 and uploads the classic prefix while B stays offline.

offset                    0           80          100 = seal
                          |           |           |
A, C, and remote storage  [ epoch 0   | epoch 2   ]  diskless records from 100 on
B (offline)               [ epoch 0   | epoch 1   ]  B never replicated [80, 100)

A requests the phase 2 flag with lastClassicLeaderEpoch = 2, the epoch of its record at 99. In phase 2, B returns, joins the ISR, and can lead. Its log ends at the seal, so it looks complete. Without this change, B serves its epoch 1 records at [80, 100), which no other replica holds. The cache of B also maps that range to epoch 1, and no remote segment has that epoch.

With this change, B checks its log in applyDelta before it leads. Its epoch at 99 is 1, so the broker deletes the log and lets the leadership change create an empty one, which the existing recovery rebuilds from remote storage:

sequenceDiagram
    participant K as Controller
    participant B as ReplicaManager of B
    participant RLM as Remote log manager of B
    participant D as DisklessLeaderEndPoint
    participant R as Remote storage
    K-->>B: Delta: B leads epoch 5, lastClassicLeaderEpoch = 2
    B->>B: Stop the replica, move, and consolidation fetchers
    B->>B: Epoch at 99 is 1, so the log fails the check
    B->>RLM: Cancel the partition tasks, keep the metadata subscription
    B->>B: Delete the current and future logs
    B->>B: makeLeader creates an empty log in epoch 5
    B->>RLM: onLeadershipChange starts new tasks on the new log
    B->>D: Consolidation fetch at 0
    D-->>B: OFFSET_MOVED_TO_TIERED_STORAGE, earliest local offset 100
    B->>R: Segment at 99 in epoch 2, the recorded epoch
    R-->>B: Segment with epochs 0 and 2
    B->>B: Restore the epochs and producer state, start the log at 100
    B->>D: Consolidation fetch at 100

Recreate instead of truncate

The broker never truncates a log that it keeps in place. A truncation leaves state that later steps trust, and each piece needs its own guard. A recreated log starts without any of it. In this example, broker B led epoch 3, uploaded [50, 119], and still holds [100, 120) when an unclean leader election makes A the leader in epoch 4. A seals at 100 and records lastClassicLeaderEpoch = 2.

offset                     50             100 = seal           120
                           |              |                    |
log of B                   [ epoch 2      | epoch 3, discarded ]
remote progress of B                                         ^ 119
future log of B (moving)   [ epoch 2      | epoch 3, discarded ]
diskless records                          [ from 100 on ...

A truncation of B to the seal leaves this state behind:

State	After a truncation in place	After a recreate
Highest offset in remote storage	Stays 119, because it only grows. The WAL pruner then deletes diskless batches `[100, 120)` before consolidation uploads them.	`-1` until the remote log manager computes it from the rebuilt log.
Future log of a log directory move	Keeps `[100, 120)` in epoch 3 and replaces the current log once both end at 120.	Deleted with the current log.
Remote log tasks	Can compute 119 in epoch 3 before the truncation and apply it after consolidation grows the log back past it. The copy task keeps a copied offset that skips `[100, 120)`.	Canceled. A task that still runs updates only the deleted log.
Promotion of the future log	Can run between the check and the truncation.	Stopping the move fetcher waits for a running promotion.

The epoch of B at 99 is 2, but the check still rejects B because it holds classic records past the seal. If B applied the seal while it was online, it already truncated its log to 100. The check then rejects B because its remote progress 119 isn't below its log end offset 100. A remote log task that computed 119 before that truncation can't install it after the check, because the log drops an offset that a task computed before a truncation.

Remote segment that crosses the seal

Leader X appends [100, 120) in epoch 3 and uploads a segment [50, 119] with epochs 2 and 3. An unclean leader election then makes Y the leader in epoch 4. Y's log ends at 100. The partition loses [100, 120). Y seals at 100 before any new produce and records lastClassicLeaderEpoch = 2.

offset           50             100 = seal           120
                 |              |                    |
remote segment   [ epoch 2      | epoch 3, discarded ]
diskless log                    [ diskless records from 100 on ...
upstream rebuild                                     ^ local log starts at 120
phase 2 rebuild                 ^ local log starts at 100

The segment at 99 in epoch 2 is the right one to rebuild from, but it ends at 119. Without this change, TierStateMachine starts the local log at 120. The consolidating leader then never materializes the diskless records [100, 120), and the WAL prune deletes them. With this change, a phase 2 rebuild starts the local log at the seal with only the epoch 2 entry, and the leader consolidates the diskless records from 100 on.

Source with local classic records below the seal

In phase 2, a classic leader can still hold local classic segments below the seal, and so can the current log of a log directory move. Upstream TierStateMachine rebuilds from the segment that contains the record before the local log start of the source. Broker X leads epoch 1 and uploads a segment [0, 119] with its records [80, 120). An unclean leader election then makes A the leader in epoch 2. A appends its own records from 80 and seals at 100. It records lastClassicLeaderEpoch = 2. The local log of A starts at 50.

offset             0        50       80              100 = seal      120
                   |        |        |               |               |
committed prefix   [ epoch 0         | epoch 2       ]  diskless records from 100 on
local log of A              [ epoch 0| epoch 2       ]
segment of X       [ epoch 0         | epoch 1, discarded            ]
segment of A                [ epoch 0| epoch 2       ]

A follower that rebuilds from A looks up the record at 49 in epoch 0. The segment of X holds that record in the same epoch, so remote storage can return that segment:

Without this change, the follower restores epoch 1 for [80, 100) and starts its local log at 120, past the diskless records [100, 120).
With this change, the follower looks up the record at 99 in epoch 2, which only the segment of A holds. It restores epochs 0 and 2 and starts its local log at 100. It then hands off to consolidation.

Changes

Recorded epoch: the leader that requests the phase 2 flag sends the leader epoch of its record at seal - 1 in SetClassicPrefixInRemoteStorage. The controller stores it with the flag as tag 104 (lastClassicLeaderEpoch) and drops it when the switch is undone. The value is -1 if the leader's log doesn't contain a record below the seal. The controller rejects a value below -1 or higher than the current leader epoch, which no replica can match. DescribeTopicPartitions returns the epoch with the other switch fields, and kafka-topic-switch.sh state prints it.
Replica check: applyDelta checks each consolidating phase 2 replica that is new to the broker or gets a higher leader epoch, before the leader and follower transitions. The flag record bumps the leader epoch, which makes live replicas run the check when phase 2 starts. The broker first stops the replica, move, and consolidation fetchers of the partition. It keeps the log only if all of these are true:
- The epoch at min(seal, LEO) - 1 is the recorded epoch, or the log has no record below that offset. Each epoch has one leader, so a matching epoch proves that the records below the seal are the committed ones, at any high watermark.
- Records past the seal are consolidated diskless records (hasConsolidatedDisklessSuffixFromSeal).
- highestOffsetInRemoteStorage is below the log end offset.
- The future log, if any, ends at or below the current log end offset, in the same epoch as the current log at that offset.
Recreate: otherwise, the broker stops the replica with stopPartitions and deleteLocalLog = true, as for a deleted partition. That removes the partition, deletes the current and future logs, and cancels the remote log tasks. A new keepRemoteLogMetadata flag keeps the remote log metadata subscription and the topic ID of the partition, because the partition stays on the broker. The leader or follower transition in the same delta then creates an empty log, and the existing recovery rebuilds the log, the leader epoch cache, and the producer state from remote storage. UnifiedLog and the remote log manager need no change for this path.
Failed recreate: a storage error can stop the deletion before it reaches the future log, which would then restart the move on the new log. If the broker can't delete both logs or cancel the remote log tasks, it marks the replica offline with its topic ID, as after a storage error in a transition, so later deltas skip it. stopPartitions cancels the remote log tasks even if a deletion fails, because a copy task that keeps running also keeps the copy progress of the deleted log.
Remote progress fence: a kept log can still have truncated before the check, to the seal in phase 1 or to the new leader after an unclean leader election. The remote log tasks compute the highest offset in remote storage without the log lock. A follower task that read the discarded tail, or a copy task that uploaded it, could install its offset after the check accepted the log. UnifiedLog now counts truncations. A truncation counts itself and changes the log under the log lock. The follower and copy tasks read the count under that lock before they read the leader epoch cache or the segments. A task never pairs the new count with the state from before the truncation. updateHighestOffsetInRemoteStorage(offset, truncationCount) installs the offset under the log lock only if the count didn't change. After a truncation, the copy task also finds its copied offset again instead of resuming after the discarded tail.
Rebuild: in phase 2, TierStateMachine looks up the remote segment that holds seal - 1 in the recorded epoch, whatever the local log start of the source is, and starts the local log at the seal. Each epoch has one leader, so the records of that segment below the seal are the committed prefix, even if the segment also covers discarded records past the seal. The rebuild keeps the epochs that start below the seal and skips the epoch lookup on the source. It restores the producer snapshot only if the segment ends at seal - 1. Sealing aborts undecided transactions, so the empty producer state is correct at the seal. Partitions without a recorded epoch, including partitions in phase 1, keep the upstream rebuild. TierStateMachine.java is a new INTERLEAVED entry in INKLESS_OWNERSHIP.
First prune: for a switched topic, the endpoint uses the seal as the lowest WAL start. fetchEarliestLocalOffset applies the same bound. The control plane keeps the WAL start at the classic log start until the first prune moves it to the seal. Without the bound, a leader that lost its local log before that prune reads the first diskless batch at offset 0, and its local log hides the remote prefix.
Deleted prefix: if retention or DeleteRecords moved the cross-tier earliest offset to the diskless start or past it, the endpoint returns OFFSET_OUT_OF_RANGE before it asks remote storage. The reset then starts the local log at that offset. DeleteRecords at the seal runs only the local leg, so before the first prune the WAL start stays below the seal. Without this check, the fetcher waits for coverage of seal - 1 that never returns, or rebuilds a deleted prefix from a segment that remote storage still lists.
Retention: while a recreated leader rebuilds, its leader epoch cache lacks the older epochs. The remote log manager runs its expiration task as soon as the broker becomes the leader. That task deletes every remote segment whose epochs are all older than the earliest cached epoch, without an offset check. For a consolidating partition, that cleanup now deletes only the segments that end below the cross-tier log start. Without this guard, the task deletes the classic prefix that the rebuild reads. A leader that lost its local log had the same exposure.
Docs: CLASSIC_TO_DISKLESS_SWITCH.md describes the recorded epoch, the replica check, the recreate, and the phase 2 rebuild. DISKLESS_CONSOLIDATION.md describes the first-prune bound, the deleted-prefix reset, the retention guard, and why a replica with stale remote progress is recreated.

A leader uploads only segments below its last stable offset, so the partition had committed every record in remote storage when its leader uploaded it. Phase 1 blocks unclean leader election. An unclean leader election before the switch, through the cluster default or an ElectLeaders request, can still discard uploaded records, and the phase 2 rebuild handles that case.

inkless-release-0.50 has the SetClassicPrefixInRemoteStorage RPC and tag 103, but its brokers don't send the flag request. So no released partition is in phase 2 without the recorded epoch.

Limitations

If lastClassicLeaderEpoch is -1, brokers skip the check and use the upstream rebuild. Besides an empty prefix, that covers a flag that a controller or leader without this change set, as during a rolling upgrade. It also covers a metadata snapshot that such a controller wrote.
Only consolidating partitions run the check. A switched leader below the seal of a partition that doesn't consolidate goes offline, because nothing rebuilds its prefix from remote storage.
A replica whose log ends below the seal compares the epoch of its last record, so the broker can recreate it even when its records match. That costs only the rebuild.
A leader epoch change that Partition.makeLeader records at the log end offset is later replaced when consolidation appends batches in the older diskless epoch. The leader epoch cache then maps consolidated records to the classic epoch, so hasConsolidatedDisklessSuffixFromSeal fails on the next leadership change. main truncates that replica to the seal and consolidates again. With this change, the broker recreates it, which costs the same consolidation plus the rebuild of the epoch cache. The cache issue predates this change and needs its own fix, tracked in KC-727.
The leader relies on the TierStateMachine rebuild, like a leader that lost its local log. A gap in that rebuild affects both cases.
For a consolidating partition, a remote segment from an epoch older than the leader's cache that doesn't end below the cross-tier log start stays until retention moves that start past it. That includes a divergent segment from before the switch. Upstream treats a divergent segment inside the cached epochs the same way.
After an unclean leader election, a remote segment that crosses the seal still serves tiered reads below the seal. Consumers that read through it receive the discarded records past the seal in place of the diskless records at the same offsets. Upstream tiered storage has the same gap after an unclean leader election, and closing it needs a cap on remote reads in RemoteLogManager.

Assisted-by: Claude Opus 5.5

Copilot

🟡 Changes recommended

Recovery can restore discarded state beyond the seal and skip diskless records, while deleted-prefix handling can retry indefinitely after local-log loss.

2 open issues

What changed in this PR

This PR strengthens phase 2 of Inkless’s classic-to-diskless switch by recording the last classic record’s leader epoch and using it to detect divergent replicas and guide recovery.

Changes:

Carry the recorded epoch through the broker request, controller metadata, and broker metadata view.
Check replica prefixes, adjust consolidation recovery before the first prune, and protect remote segments during rebuilds.
Add regression tests and document the recovery behavior.

File	Description
`storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java`	Tests remote-prefix retention during rebuilds.
`storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java`	Protects retained segments from epoch-cache cleanup.
`metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java`	Tests epoch persistence, merging, and switch undo.
`metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerInklessTest.java`	Checks controller persistence of the recorded epoch.
`metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java`	Tests epoch tags in phase-2 records.
`metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java`	Stores and propagates the recorded epoch.
`metadata/src/main/java/org/apache/kafka/metadata/InitDisklessLogFields.java`	Adds encoding and decoding for tag `104`.
`metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java`	Passes the requested epoch to the change builder.
`metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java`	Includes the recorded epoch in phase-2 metadata.
`docs/inkless/DISKLESS_CONSOLIDATION.md`	Documents recovery bounds and retention protection.
`docs/inkless/CLASSIC_TO_DISKLESS_SWITCH.md`	Explains epoch recording and divergent-prefix recovery.
`core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala`	Tests replica checks and remote-segment selection.
`core/src/test/scala/unit/kafka/server/InitDisklessLogManagerTest.scala`	Checks epoch values in flag requests.
`core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala`	Tests epoch responses and first-prune behavior.
`core/src/test/scala/io/aiven/inkless/consolidation/ConsolidationFetcherThreadTest.scala`	Supplies recorded-epoch metadata in the fetcher test.
`core/src/main/scala/kafka/server/ReplicaManager.scala`	Checks and truncates divergent classic prefixes.
`core/src/main/scala/kafka/server/metadata/InklessMetadataView.scala`	Exposes the recorded epoch to brokers.
`core/src/main/scala/kafka/server/InitDisklessLogState.scala`	Captures the epoch for the phase-2 request.
`core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala`	Applies seal bounds and returns the recorded epoch.
`clients/src/main/resources/common/message/SetClassicPrefixInRemoteStorageRequest.json`	Adds the optional tagged epoch field.

_{Review effort: Balanced}

…re the first prune [KC-702] The control plane keeps a switched partition's WAL start at the classic log start until the leader's first prune moves it to the seal. That prune runs only once the leader's remote copy reaches `seal - 1`, which a leader with an empty local log never reaches. Its consolidation fetch at offset 0 then reads the first diskless batch from the WAL, and the local log hides the classic prefix that remote storage holds. `DisklessLeaderEndPoint` now uses the seal as the lowest diskless start for the tiered-storage redirect and for `fetchEarliestLocalOffset`. The tier-state machine then rebuilds the prefix and starts the local log at the seal, as it does after the first prune. If remote storage doesn't cover `seal - 1`, the endpoint returns `NOT_LEADER_OR_FOLLOWER` and sets `ConsolidationRemotePrefixUnknown` instead of consolidating below the seal. Retention or `DeleteRecords` can move the cross-tier earliest offset to the seal while the WAL start stays below it. `DeleteRecords` at the seal runs only the local leg. Without another check, the fetcher waits for coverage of `seal - 1` that never returns, or rebuilds a deleted prefix from a segment that remote storage still lists. If the cross-tier earliest offset is at the diskless start or past it, the endpoint now returns `OFFSET_OUT_OF_RANGE` before it asks remote storage. The out-of-range reset starts the local log at that offset. ## Testing `DisklessLeaderEndPointTest` covers the redirect, the retry without remote coverage, the reset when the cross-tier earliest offset is at the seal, a fetch at the seal, and the earliest local offset before the first prune. Assisted-by: Claude Opus 5.5

Copilot

🟡 Changes recommended

The log-directory repair leaves the move fetcher running, allowing a pending fetch response to promote a divergent future log and undo the repair.

1 open issue

Stop move fetcher and revalidate future log before promotion · New

2 resolved since last review

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

A critical truncation bug retains stale remote-storage progress, allowing the WAL pruner to delete diskless records before they are uploaded and making them unrecoverable past the seal.

1 open issue

Reset stale remote progress after full truncation · New

1 resolved since last review

Stop move fetcher and revalidate future log before promotion

_{Review effort: Balanced}

8000

Copilot

🟡 Changes recommended

Stale remote-progress updates after truncation and insufficient epoch validation during seal recovery can cause unuploaded records to be pruned and correct classic records to be skipped.

2 open issues

1 resolved since last review

Reset stale remote progress after full truncation

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

Critical races can leave divergent logs attached or promoted and allow remote copying to skip data while advancing progress and WAL pruning.

3 open issues

2 resolved since last review

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

A partial log-deletion failure can allow replica recreation while stale remote-copy tasks remain active, potentially skipping uploads and allowing WAL pruning over missing data.

1 open issue

Prevent replica recreation after partial log-deletion failure · New

3 resolved since last review

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

A critical race allows a surviving remote task to install stale progress after log truncation, letting the WAL pruner delete diskless records before consolidation uploads them.

1 open issue

Prevent stale remote offsets from pruning unuploaded WAL records · New

1 resolved since last review

Prevent replica recreation after partial log-deletion failure

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

The unlocked truncation-count getter can expose a new count with stale log state, allowing discarded consolidation progress to pass validation and trigger WAL pruning of records that have not been uploaded.

1 open issue

Prevent WAL pruning from using stale log state during truncation · New

1 resolved since last review

Prevent stale remote offsets from pruning unuploaded WAL records

_{Review effort: Balanced}

…mmitted classic prefix [KC-702] In phase 2 of the classic-to-diskless switch, the controller adds every live replica to the ISR without checking its local copy of the classic prefix, and any replica can lead. A replica can hold records below the seal that the partition doesn't hold. It led an older leader epoch and wrote records that the next leader didn't have, or an unclean leader election before the switch discarded records below its high watermark. Forced deletion keeps the leader epoch cache of that replica, and the remote log manager uses that cache to pick remote segments. On that replica, a tiered read below the seal fails with `OFFSET_OUT_OF_RANGE`. The leader that requests the phase 2 flag now sends the leader epoch of its record at `seal - 1`. The controller stores it with the flag as tag `104` (`lastClassicLeaderEpoch`). Each epoch has one leader, so a replica whose leader epoch cache has that epoch at `seal - 1` holds the committed prefix, at any high watermark. The controller rejects an epoch below `-1` or higher than the current leader epoch, which no replica can match. `DescribeTopicPartitions` returns the epoch with the other switch fields, and `kafka-topic-switch.sh state` prints it. A consolidating replica checks its local log when it first applies the partition and on every new leader epoch, after the broker stops the replica, consolidation, and log directory move fetchers of the partition. Stopping the move fetcher waits for a running promotion of the future log. The replica keeps its log only if its last record below the seal is in the recorded epoch, its records past the seal are consolidated diskless records, its highest offset in remote storage is below its log end offset, and its future log ends in the same epoch as the current log at that offset. Otherwise, `applyDelta` stops the replica as it stops a deleted partition. It deletes the current and future logs, and the remote log manager cancels the partition tasks but keeps the remote log metadata subscription. The leadership change in the same delta creates an empty log, and the existing recovery rebuilds it from remote storage. A storage error can stop the deletion before it reaches the future log, which would then restart the move on the new log. If the broker can't delete both logs or cancel the remote log tasks, it marks the replica offline with its topic ID, as after a storage error in a transition, so later deltas skip it. `stopPartitions` cancels the remote log tasks even if a deletion fails, because a copy task that keeps running also keeps the copy progress of the deleted log. The broker never truncates a log that it keeps in place. Truncating in place left state behind that later steps trusted: the future log of a log directory move, a highest offset in remote storage that only grows, and remote log tasks that compute offsets without the log lock. Each of them needed its own guard. A recreated log starts without that state, and the remote log tasks that still run update only the deleted log. A kept log can still have truncated before the check, to the seal in phase 1 or to the new leader after an unclean leader election. The remote log tasks compute the highest offset in remote storage without the log lock. A follower task that read the discarded tail, or a copy task that uploaded it, could install its offset after the check accepted the log. `UnifiedLog` now counts truncations. A truncation counts itself and changes the log under the log lock. The follower and copy tasks read the count under that lock before they read the leader epoch cache or the segments. A task never pairs the new count with the state from before the truncation. `updateHighestOffsetInRemoteStorage(offset, truncationCount)` installs the offset under the log lock only if the count didn't change. After a truncation, the copy task also finds its copied offset again instead of resuming after the discarded tail. Upstream `TierStateMachine` rebuilds from the remote segment that holds the record before the local log start of the source. In phase 2, a classic leader or the current log of a log directory move can still hold local classic segments below the seal. A segment from a discarded epoch can hold that record in the same epoch as the committed prefix and diverge after it, and it can also hold records past the seal. Restoring such a segment restores the wrong epochs and starts the local log at its end, past diskless records that the WAL prune then deletes. A phase 2 rebuild now looks up the segment that holds `seal - 1` in the recorded epoch, whatever the local log start of the source is, and starts the local log at the seal. Its records below the seal are the committed prefix, even if it also holds discarded records past the seal. The rebuild keeps the epochs that start below the seal and skips the epoch lookup on the source. It restores the producer snapshot only if the segment ends at `seal - 1`. Sealing aborts undecided transactions, so the empty producer state is correct at the seal. Partitions without a recorded epoch keep the upstream rebuild. ## Testing - `testSwitchedLeaderInPhaseTwoRecreatesReplicaWithoutCommittedHistory` covers a leader with a divergent epoch at `seal - 1`, and a leader whose future log ends past the current log or in another epoch. The broker deletes both logs, stops the remote log tasks before the leadership change, and starts consolidation at 0 on an empty log. A matching future log keeps the move running. - `testSwitchedLeaderInPhaseTwoDropsRemoteProgressPastTheSeal` runs the WAL pruner on leaders with a divergent epoch, a classic tail past the seal, or remote progress at their log end offset. The broker recreates them, and the pruner doesn't prune. A leader that holds the committed prefix keeps its log, and the pruner prunes at `seal - 1`. - `testSwitchedLeaderInPhaseTwoKeepsConsolidatedReplica` keeps a leader whose log holds the committed prefix and consolidated records past the seal, and resumes consolidation at its log end offset. - `testSwitchedConsolidatingFollowerInPhaseTwoRecreatesReplicaWithoutCommittedHistory` covers a follower at the seal with a matching and a divergent epoch. - `testOnlineSwitchedReplicaWithoutCommittedHistoryRecreatedOnNewLeaderEpoch` checks that an ISR change doesn't recreate an online replica, and that the next leader epoch does. - `testSwitchedReplicaStaysOfflineWhenItsDeletionFails` removes the directory of the current or the future log before the check. The broker cancels the remote log tasks and marks the replica offline. The next leader epoch doesn't recreate it, and no fetcher starts. Without the change, the broker recreates the replica. - `testUpdateHighestOffsetInRemoteStorageIgnoresOffsetComputedBeforeTruncation` checks that `truncateTo` and `truncateFullyAndStartAt` make the log drop an offset from before them, and that a `truncateTo` without effect doesn't. - `testTruncationCountWaitsForRunningTruncation` pauses `truncateTo` after it counts itself and before it truncates the leader epoch cache. A read of the count waits for the truncation to finish. If the read doesn't take the log lock, it returns the new count while the cache still holds the removed epoch, and the test fails. - `testFollowerTaskInstallsHighestRemoteOffsetWithTruncationCountFromBeforeTheLookup` and `testCopyTaskFindsCopiedOffsetAgainAfterTruncationDuringUpload` truncate the log during the lookup and during the upload. Both tasks install the offset with the count from before, and the copy task finds its copied offset again on the next run. Without the change, both tests fail. - `testTierStateMachineLooksUpLastClassicSegmentByRecordedEpoch` checks that the rebuild ignores a segment from a discarded epoch that covers `seal - 1`. - `testTierStateMachineStopsAtSealForSegmentCrossingIt` rebuilds from a segment that crosses the seal. The segment contains one epoch, or a discarded epoch past the seal. The rebuild starts the log at the seal, keeps only the epochs below it, and skips the producer snapshot. Without the change, the fetch starts at the segment end. - `testTierStateMachineLooksUpRecordedEpochForSourceWithLocalClassicPrefix` rebuilds from a source whose local log starts at 50, below seal 100. The segment that holds 49 in epoch 0 comes from discarded epoch 1, which epoch 2 replaced at 80. The rebuild uses the segment that holds 99 in epoch 2 instead, which ends at the seal or crosses it. Without the change, the rebuild restores epoch 1. - `PartitionRegistrationTest`, `PartitionChangeBuilderTest`, `ReplicationControlManagerInklessTest`, `InitDisklessLogManagerTest`, `InklessMetadataViewTest`, `DescribeTopicPartitionsRequestHandlerTest`, and `TopicSwitchCommandTest` cover the new field from the request to the broker metadata view, including the range check in the controller and the `kafka-topic-switch.sh state` output. Assisted-by: Claude Opus 5.5

… epoch cache rebuilds [KC-702] A phase 2 leader that the broker recreates caches only its own leader epoch until `TierStateMachine` restores the older epochs from remote storage. A leader that lost its local log is in the same state. The remote log manager runs its expiration task as soon as the broker becomes the leader. That task deletes every remote segment whose epochs are all older than the earliest cached epoch, and it doesn't check offsets, because upstream assumes that those segments end below the log start. For such a leader, the assumption doesn't hold, and the task deletes the classic prefix that the rebuild reads and consumers fetch. For a consolidating partition, this cleanup now deletes only the segments that end below the cross-tier log start, which is the authoritative log start of the partition. The other segments wait until retention moves that start past them, like a divergent segment inside the cached epochs does upstream. Classic topics keep the upstream behavior. ## Testing - `testEpochCacheCleanupKeepsConsolidatingSegmentsFromCrossTierStart` runs the expiration task for a leader that caches only epoch 5, with remote segments from epochs 0 and 1 on both sides of the cross-tier log start. For a consolidating partition, the task deletes only the segment below that start. Without the change, it deletes both. For a classic partition, it deletes both, as upstream does. - `testRecreatedLeaderKeepsRemotePrefixUntilRebuild` starts a real empty log in epoch 7, as a recreated leader does. Remote storage holds segments from classic epochs 0, 2, and 3 below seal 100 and from the diskless epoch 5, and the cross-tier log start is 40. Both before and after the rebuild, expiration deletes only the segment below 40. Without the guard, the first expiration deletes every remote segment. Assisted-by: Claude Opus 5.5

Copilot

🔵 Needs a closer look

The coupled replica-deletion, remote-retention, and recovery changes need storage-domain review and integration validation of concurrent remote tasks, failover, and log-directory moves to assess data-safety invariants.

No open issues 🎉

1 resolved since last review

Prevent WAL pruning from using stale log state during truncation

_{Review effort: Balanced}

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from a0816d2 to dc5f809 Compare November 5, 2026 12:07

giuseppelillo requested a balanced review from Copilot November 5, 2026 12:21

Copilot started reviewing on behalf of giuseppelillo November 5, 2026 12:22

View session

Copilot AI reviewed Nov 5, 2026

View reviewed changes

Comment thread core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala Resolved

Comment thread core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from dc5f809 to c6a11b5 Compare November 5, 2026 13:28

giuseppelillo requested a balanced review from Copilot November 5, 2026 13:42

Copilot started reviewing on behalf of giuseppelillo November 5, 2026 13:42

View session

Copilot AI reviewed Nov 5, 2026

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from c6a11b5 to 9f8deec Compare November 5, 2026 14:07

giuseppelillo requested a balanced review from Copilot November 5, 2026 14:10

Copilot started reviewing on behalf of giuseppelillo November 5, 2026 14:10

View session

Copilot AI reviewed Nov 5, 2026

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from 9f8deec to 4094c80 Compare November 5, 2026 15:41

giuseppelillo requested a balanced review from Copilot November 5, 2026 15:52

Copilot started reviewing on behalf of giuseppelillo November 5, 2026 15:52

View session

Copilot AI reviewed Nov 5, 2026

View reviewed changes

Comment thread core/src/main/java/kafka/server/TierStateMachine.java Resolved

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from 4094c80 to 10bc94c Compare November 6, 2026 08:55

giuseppelillo requested a balanced review from Copilot November 6, 2026 09:00

Copilot started reviewing on behalf of giuseppelillo November 6, 2026 09:00

View session

Copilot AI reviewed Nov 6, 2026

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from 10bc94c to b71e175 Compare November 6, 2026 09:51

giuseppelillo changed the title ~~fix(inkless:switch): check phase 2 replicas against the recorded last classic epoch [KC-702]~~ fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] Nov 6, 2026

giuseppelillo requested a balanced review from Copilot November 6, 2026 09:54

Copilot started reviewing on behalf of giuseppelillo November 6, 2026 09:54

View session

Copilot AI reviewed Nov 6, 2026

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from b71e175 to 6509237 Compare November 6, 2026 10:56

giuseppelillo requested a balanced review from Copilot November 6, 2026 11:01

Copilot started reviewing on behalf of giuseppelillo November 6, 2026 11:02

View session

Copilot AI reviewed Nov 6, 2026

View reviewed changes

Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Resolved

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from 6509237 to fdf1556 Compare November 6, 2026 11:44

giuseppelillo requested a balanced review from Copilot November 6, 2026 12:24

Copilot started reviewing on behalf of giuseppelillo November 6, 2026 12:25

View session

Copilot AI reviewed Nov 6, 2026

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java Resolved

giuseppelillo added 2 commits November 6, 2026 14:11

giuseppelillo force-pushed the giuseppelillo/kc-702-epoch-rework branch from fdf1556 to 037afb6 Compare November 6, 2026 13:12

giuseppelillo requested a balanced review from Copilot November 6, 2026 13:14

Copilot started reviewing on behalf of giuseppelillo November 6, 2026 13:15

View session

Copilot AI reviewed Nov 6, 2026

View reviewed changes

giuseppelillo marked this pull request as ready for review November 6, 2026 13:24

Navigation Menu

Repository navigation

fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702] - #857

fix(inkless:switch): recreate phase 2 replicas that don't hold the committed classic prefix [KC-702]#857
giuseppelillo wants to merge 3 commits into
mainfrom
giuseppelillo/kc-702-epoch-rework

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

Uh oh!

Example

Recreate instead of truncate

Remote segment that crosses the seal

Source with local classic records below the seal

Changes

Limitations

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Choose a reason for hiding this comment

🔵 Needs a closer look

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants