fix(inkless:consolidation): stop over-reclaiming the classic prefix on rebuilt leaders [KC-332]#711
Open
viktorsomogyi wants to merge 9 commits into
Open
fix(inkless:consolidation): stop over-reclaiming the classic prefix on rebuilt leaders [KC-332]#711viktorsomogyi wants to merge 9 commits into
viktorsomogyi wants to merge 9 commits into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR fixes an Inkless consolidation safety issue where a freshly rebuilt consolidating leader could cause RemoteLogManager to over-reclaim the remote classic prefix by using a broker-local logStartOffset pinned at the classic-to-diskless seal. It introduces a control-plane driven log-start override for reclaim decisions and adjusts Diskless fetch behavior to use a broker-agnostic whole-log start.
Changes:
- Add a
logStartOffsetOverrideandisConsolidatingDisklessPartitionpredicate toRemoteLogManagerto drive a safe reclaim floor (and fail-safe when the override is unavailable). - Wire both lambdas from
BrokerServerviaReplicaManager(control-plane cross-tier earliest + consolidating predicate). - Update
DisklessLeaderEndPointto reportmin(crossTierEarliest, localLogStart)as the whole-log start, with new targeted tests.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java | Adds regression tests covering override-driven reclaim floor, fail-safe behavior when override is absent, and interactions with active retention. |
| storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java | Introduces override + consolidating predicate plumbing and uses them to compute a safe remote-reclaim floor (and leader log-start reporting). |
| core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala | Adds tests asserting whole-log start uses cross-tier earliest for consolidating leaders (and documents fallback behavior). |
| core/src/main/scala/kafka/server/BrokerServer.scala | Wires ReplicaManager-backed override and consolidating predicate into RemoteLogManager. |
| core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala | Uses min(crossTierEarliest, localLogStart) as whole-log start to avoid rejecting reads of surviving remote prefixes. |
viktorsomogyi
force-pushed
the
svv/ts-unification-delete-records-routing
branch
2 times, most recently
from
July 22, 2026 13:29
2549301 to
634299e
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
2 times, most recently
from
July 22, 2026 13:48
654af13 to
2336424
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-delete-records-routing
branch
from
July 22, 2026 15:13
634299e to
bbebd3d
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
from
July 22, 2026 15:14
2336424 to
ea588ff
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-delete-records-routing
branch
from
July 23, 2026 07:06
bbebd3d to
e4eac32
Compare
Base automatically changed from
svv/ts-unification-delete-records-routing
to
main
July 23, 2026 07:39
viktorsomogyi
changed the base branch from
main
to
svv/ts-unification-wal-fetcher
July 23, 2026 07:47
viktorsomogyi
changed the base branch from
svv/ts-unification-wal-fetcher
to
main
July 23, 2026 07:47
viktorsomogyi
marked this pull request as ready for review
July 23, 2026 07:47
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
from
July 23, 2026 07:53
ea588ff to
66ddbb7
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
from
July 23, 2026 08:53
52cf871 to
66ddbb7
Compare
Comment on lines
+925
to
+928
| OptionalLong override = logStartOffsetOverride.apply(topicIdPartition.topicPartition()); | ||
| long logStartOffset = override.isPresent() | ||
| ? override.getAsLong() | ||
| : findLogStartOffset(topicIdPartition, log); |
Contributor
There was a problem hiding this comment.
ReplicaManager.crossTierEarliestOffset returns ListOffsets(EARLIEST) = COALESCE(remote_log_start_offset, log_start_offset). If remote_log_start_offset is NULL due to an error in reporting the value, the cross tier earliest offset returned is the pruned WAL frontier, so there's the chance that all data from remote gets deleted in case this happens.
Example:
- Born-consolidated topic; records [0,1000) produced, [0,700) tiered to remote.
- The CrossTierLogStartReporter never lands a successful flush for this partition — crash between enqueue and the 1s flush, a persistent partition-specific advanceCrossTierLogStartOffset failure, or an enqueue guard drop during post-switch metadata propagation. remote_log_start_offset stays NULL.
- The 5-min WAL pruner advances log_start_offset to 700.
- A leader runs become-leader / expiration. crossTierEarliestOffset returns COALESCE(NULL, 700) = 700 → present.
- reclaimFloorLogStartOffset returns 700 → deletes remote [0,700), which was still within retention. And become-leader reports 700 → the reporter persists it via the forward-only advance_cross_tier_log_start_v1, so remote_log_start_offset is now permanently 700 and EARLIEST is permanently wrong.
Contributor
Author
There was a problem hiding this comment.
I think this is a good catch, let me fix it.
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
2 times, most recently
from
July 23, 2026 15:37
86e3b1a to
a6f0a03
Compare
Contributor
Author
|
@giuseppelillo addressed the bug you discovered, PTAL. |
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
2 times, most recently
from
July 23, 2026 15:47
bbca8b1 to
f6c0f80
Compare
viktorsomogyi
marked this pull request as draft
July 24, 2026 09:25
Contributor
Author
|
Moving this back to draft until I resolve the test failures. |
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
from
July 24, 2026 12:17
f6c0f80 to
e3f351d
Compare
…n rebuilt leaders [KC-332] A freshly-rebuilt consolidating leader had its local logStartOffset pinned at the classic-to-diskless seal, so the RemoteLogManager reclaimed the entire remote classic prefix [earliest, seal) regardless of DeleteRecords / retention. RemoteLogManager now takes a logStartOffsetOverride (the control-plane cross-tier earliest) as its reclaim floor, and DisklessLeaderEndPoint reports min(crossTierEarliest, localLogStart) as the whole-log start. When the override is unavailable (control plane down / metadata not yet propagated) the RLM fails safe for a consolidating partition. It uses the remote earliest and defers rather than falling back to the local seal, so a transient outage can never trigger irreversible over-deletion. A new isConsolidatingDisklessPartition predicate distinguishes this from classic topics, which keep the upstream local-log-start floor. BrokerServer wires both lambdas into the RLM, backed by the shared ReplicaManager.crossTierEarliestOffset / isConsolidatingDisklessPartition accessors added here (control-plane cross-tier earliest, write-through cache). Co-authored-by: Cursor <cursoragent@cursor.com>
…r [KC-332] The reclaim floor used ListOffsets(EARLIEST), which COALESCEs to the WAL prune frontier when remote_log_start_offset is NULL. That frontier can run ahead of the true remote start, so using it as the floor would over-reclaim live remote segments and lock the wrong value in via the forward-only advance. Add a dedicated crossTierRemoteLogStartOffset that reads remote_log_start_offset directly, null-aware. When it is NULL the RLM fails safe to the true remote earliest instead. crossTierEarliestOffset keeps the COALESCE fallback for the read path.
…floor [KC-332] Update the reclaim-floor comments to reference the raw remote log start instead of the COALESCE'd cross-tier earliest, and explain why ListOffsets(EARLIEST) is not used. Rename the local to match.
…KC-332] advanceCrossTierEarliestForDeleteRecords caught Throwable, which would swallow Errors (OutOfMemoryError, StackOverflowError) and fail the control-plane report silently. Narrow to Exception so Errors propagate.
…hared-state branch [KC-332] Add a test for the sharedState == null branch of crossTierRemoteLogStartOffset, mirroring the existing isConsolidatingDisklessPartition test. Verifies the accessor returns empty without touching the control plane.
…C-332] Wire GetCrossTierLogStartJob through the timed JobUtils.run and add a PostgresControlPlaneMetrics entry for it. Regenerate metrics.rst.
…st comments [KC-332] The Problem A/B names only exist in local design notes, so reviewers have no context for them. Replace them with self-describing wording (cross-broker EARLIEST consistency, seal-based over-reclaim) and keep the comments' intent. Co-authored-by: Cursor <cursoragent@cursor.com>
…eport against over-reclaim [KC-332] Cover the consolidating data-loss paths directly. When the cross-tier remote start override is empty, the become-leader report resolves to the remote earliest (findLogStartOffset), never the local seal and never skipped, so remote_log_start_offset heals instead of leaving EARLIEST to COALESCE to the pruned WAL frontier. When the override is present it is reported verbatim. findLogStartOffset returns the remote earliest even though the rebuilt leader's local start is pinned at the seal, and a combined bootstrap-plus-reclaim guard asserts both the report and the reclaim floor use that earliest so the classic prefix is not deleted. Co-authored-by: Cursor <cursoragent@cursor.com>
viktorsomogyi
force-pushed
the
svv/ts-unification-reclaim-floor
branch
from
July 24, 2026 13:07
e3f351d to
5079dcd
Compare
viktorsomogyi
marked this pull request as ready for review
July 24, 2026 13:07
Contributor
Author
|
@giuseppelillo the tests should pass now, it's ready for review again. |
…ched topic [KC-332] Add an integration test that deletes records below the seal on a two-partition consolidated topic and asserts the earliest settles at exactly the requested per-partition boundary, never over-reclaiming the surviving classic prefix down to the seal. Also covers the forward-only control-plane advance (a repeat delete below the current start is a no-op) and asserts remote_log_start_offset bootstraps to 0 in the control plane rather than staying NULL. Co-authored-by: Cursor <cursoragent@cursor.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Builds on the two merged consolidation fixes (#709 ListOffsets(EARLIEST) routing and #710 DeleteRecords forwarding). Those made the control plane the authoritative cross-tier earliest for reads and the DeleteRecords write path. This PR carries that same authority into remote-retention reclaim and the become-leader log-start report, so
a consolidating topic never deletes remote data that is still within the delete/retention boundary, and never reports a stale earliest.
Background
After a classic topic switches to diskless, the classic prefix
[0, seal)lives only in the remote tier. A broker that rebuilds the partition resumes its local log at the seal, so the whole-logUnifiedLog.logStartOffsetis pinned at the seal by the tier-state rebuild and only ever increments. Two things then read that stale seal instead of thetrue cross-tier earliest
Xthat DeleteRecords/retention left in the control plane:[X, seal)and reported the seal as the earliest on becoming leader;[X, seal)was rejected as out of range and the rebuild restarted at the seal. The reclaim leg is irreversible, so this is a data-loss path, not just a stale read.What changed
DisklessLeaderEndPoint): a consolidating partition now advertisesmin(crossTierEarliest, localLog.logStartOffset)as the whole-log start instead of the raw local seal, so a fetch in[X, seal)rebuilds from the remote tier and the rebuild restarts the log atX.RemoteLogManager, wired inBrokerServer): both now consult a control-plane override for consolidating partitions. On an empty override they fail safe to the remote earliest (findLogStartOffset), never the seal; classic/non-inkless topics keep the upstream broker-local behavior.ReplicaManager.crossTierRemoteLogStartOffset+ newControlPlane.getCrossTierLogStart): the reclaim floor reads the rawremote_log_start_offset, null-aware, rather thanListOffsets(EARLIEST).EARLIESTreturnsCOALESCE(remote_log_start_offset, log_start_offset), which falls back to the pruned WAL frontier when the remote start is unreported. Using that as the floor would delete still-live remote segments and lock the wrong value in through the forward-only advance. Returning empty here makes the RLM fail safe to the true remote earliest instead.crossTierEarliestOffsetkeeps the COALESCE fallback for reads, where the worst case is a transient, self-healing over-rejection.GetCrossTierLogStartJob(in-memory + Postgres) reads the raw column directly, with aGetCrossTierLogStarttimer inPostgresControlPlaneMetricsand ametrics.rstentry.Exceptionrather thanThrowable, soErrors propagate instead of silently failing open.Testing
RemoteLogManagerTest: reclaim floor honored over the seal; fail-safe on empty override for consolidating vs. upstream behavior for classic; retention x DeleteRecords composition; and hardening tests proving the become-leader report resolves to the remote earliest (never the seal, never skipped), thatfindLogStartOffsetreturns the remote earliest despite a seal-pinned local start, and a combined bootstrap-plus-reclaim guard.ReplicaManagerInklessTest/DisklessLeaderEndPointTest: the accessors, the consolidating predicate, and the whole-log-startmin()behavior.AbstractControlPlaneTest:getCrossTierLogStartis null-aware (returns empty for an unreported remote start) unlikeEARLIEST.DeleteRecordsAcrossTiersTest), which asserts the earliest settles at exactly the delete boundary on every broker with retention unset.Compatibility / risk
No behavior change for classic or non-inkless partitions: the override defaults to empty and the predicate to false, so the reclaim floor stays the broker-local log start. The control-plane advance is forward-only, and a stale cross-tier cache entry can only be too low, which under-reclaims and over-serves rather than losing data.