test(inkless:consolidation): retention.bytes cross-tier reclaim system test [KC-332]#708
Draft
viktorsomogyi wants to merge 4 commits into
Draft
Conversation
viktorsomogyi
force-pushed
the
svv/ts-unification-retention-bytes-test
branch
from
July 21, 2026 15:03
42ca364 to
fd4d0f0
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR expands Inkless cross-tier correctness coverage by adding new ducktape system tests and the supporting broker/test utilities needed to validate that earliest-offset semantics and remote reclamation behave consistently across brokers for consolidating diskless topics.
Changes:
- Add system tests for
retention.bytes-driven cross-tier reclaim and for cross-tierDeleteRecordsbehavior (including failover). - Extend
ConsolidationVerifierwith MinIO byte accounting, “stable” tiered-object counting, per-brokerEARLIESTprobing, and akafka-delete-records.shwrapper. - Adjust broker/runtime behavior so consolidating topics use the control-plane cross-tier earliest for
ListOffsets(EARLIEST), forDeleteRecordslow-watermark reporting, and as the RemoteLogManager reclaim floor; add an inter-broker forwarder for the leader-onlyDeleteRecordsleg.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/kafkatest/tests/inkless/consolidation_retention_bytes_across_tiers_test.py | New system test for size-based remote reclaim advancing cross-tier earliest. |
| tests/kafkatest/tests/inkless/consolidation_delete_records_across_tiers_test.py | New system test(s) for cross-tier DeleteRecords correctness and failover safety. |
| tests/kafkatest/services/inkless/consolidation_verifier.py | Adds MinIO byte counting, stable-count polling, per-broker earliest checks, and a DeleteRecords CLI helper. |
| storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java | Adds unit tests for consolidating reclaim-floor override and fail-safe behavior. |
| storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java | Introduces log-start override + consolidating predicate; uses them for reclaim floor and leader report. |
| core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala | Adds unit tests for cross-tier low-watermark reporting and cache/control-plane behavior. |
| core/src/test/scala/unit/kafka/server/DisklessFetchOffsetRouterTest.scala | Updates tests to assert consolidating EARLIEST routes to the control plane consistently. |
| core/src/test/scala/unit/kafka/server/DisklessDeleteRecordsForwarderTest.scala | New unit tests for the DeleteRecords forwarder routing/forwarding behavior. |
| core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala | Adds tests asserting consolidating whole-log start uses cross-tier earliest when available. |
| core/src/main/scala/kafka/server/ReplicaManager.scala | Reports/reads cross-tier earliest from control plane + cache; adjusts DeleteRecords completion semantics for consolidating topics. |
| core/src/main/scala/kafka/server/KafkaApis.scala | Fans out DeleteRecords to real leaders via the forwarder and merges responses. |
| core/src/main/scala/kafka/server/DisklessFetchOffsetRouter.scala | Routes consolidating ListOffsets(EARLIEST) to control plane instead of local classic log. |
| core/src/main/scala/kafka/server/DisklessDeleteRecordsForwarder.scala | New inter-broker forwarder for leader-only DeleteRecords local-log leg. |
| core/src/main/scala/kafka/server/BrokerServer.scala | Wires up the new forwarder and passes RLM override hooks into RemoteLogManager. |
| core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala | Uses cross-tier earliest (min with local) as whole-log start for consolidating partitions. |
| core/src/main/java/kafka/server/builders/KafkaApisBuilder.java | Updates builder call site for new KafkaApis constructor parameter. |
Comment on lines
+282
to
+291
| state = {"last": -1, "stable": 0, "value": 0} | ||
|
|
||
| def check(): | ||
| cur = self.tiered_object_count() | ||
| state["stable"] = state["stable"] + 1 if cur == state["last"] else 0 | ||
| state["last"] = cur | ||
| state["value"] = cur | ||
| self.logger.info("Tiered-storage object count: %d (stable for %d samples)" | ||
| % (cur, state["stable"])) | ||
| return state["stable"] >= settle_samples |
Comment on lines
+672
to
+683
| def check(): | ||
| per_broker = self.earliest_on_each_broker(topic, partition=partition) | ||
| values = set(per_broker.values()) | ||
| agreed = len(values) == 1 and all(v > 0 for v in values) | ||
| cur = next(iter(values)) if agreed else None | ||
| state["stable"] = state["stable"] + 1 if (agreed and cur == state["last"]) else 0 | ||
| state["last"] = cur | ||
| if agreed: | ||
| state["value"] = cur | ||
| self.logger.info("Per-broker earliest for %s-%d: %s (agreed=%s, stable for %d samples)" | ||
| % (topic, partition, per_broker, agreed, state["stable"])) | ||
| return state["stable"] >= settle_samples |
Comment on lines
+1804
to
+1809
| * Reads the write-through [[io.aiven.inkless.cache.CrossTierLogStartCache]] first and falls back to a | ||
| * control-plane `listOffsets(EARLIEST)` query, populating the cache on a hit. A stale cache entry can | ||
| * only be too low (the safe direction: it under-reclaims and over-serves, never the reverse). | ||
| */ | ||
| /** | ||
| * Whether `topicPartition` belongs to a consolidating diskless topic on this broker. |
Comment on lines
+1606
to
+1613
| val timeoutMs = deleteRecordsRequest.data.timeoutMs | ||
| // Split the diskless partitions whose local-log leg must run on another broker's real KRaft | ||
| // leader (the metadata transformer advertised a follower to the client). A forwarded request | ||
| // lands on its real leader, where routeByLeader resolves leader == self and handles it | ||
| // locally, so the chain terminates without an explicit loop guard. | ||
| val routed = disklessDeleteRecordsForwarder | ||
| .map(forwarder => (forwarder, forwarder.routeByLeader(authorizedForDeleteTopicOffsets.toMap))) | ||
|
|
…d report cross-tier low watermark [KC-332] For a switched consolidating diskless topic the metadata transformer advertises an AZ-selected replica (a follower) as the client-facing leader, so a DeleteRecords request looped on NOT_LEADER_OR_FOLLOWER and timed out. DisklessDeleteRecordsForwarder forwards the leader-only leg to the partition's real KRaft leader; KafkaApis fans the request out into local and forwarded legs and combines the results (KafkaApisBuilder / BrokerServer wire the forwarder). Completion no longer waits on the ISR low watermark for consolidating partitions (their follower log starts are frozen at the switch). Instead ReplicaManager.deleteRecords advances the control-plane cross-tier earliest and reports it as the low watermark, so DeleteRecords and a subsequent ListOffsets(EARLIEST) agree on every broker. Adds the shared crossTierEarliestOffset / isConsolidatingDisklessPartition accessors. Co-authored-by: Cursor <cursoragent@cursor.com>
…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. Co-authored-by: Cursor <cursoragent@cursor.com>
…332] Adds consolidation_delete_records_across_tiers_test.py, the end-to-end regression for Problem A and Problem B on a switched consolidating topic: after DeleteRecords the earliest offset advances off 0, is identical on every broker, settles exactly at the delete boundary, and the surviving classic prefix [delete_before, seal) reads back contiguous. Also covers leader-failover metadata-propagation timing and born-consolidated topics. Adds the verifier helpers these need: delete_records, per-broker earliest_on_each_broker / wait_for_consistent_earliest_across_brokers, create_consolidated_topic, _broker_bootstrap, offset_at bootstrap pinning, and wait_for_tiered_count_stable (a settled pre-reclaim tiered-count peak). Co-authored-by: Cursor <cursoragent@cursor.com>
…m test [KC-332] Adds consolidation_retention_bytes_across_tiers_test.py, exercising size-based (retention.bytes) reclaim of the cross-tier earliest offset on a consolidating diskless topic. Adds the verifier support it needs: tiered_object_bytes (with the _object_entries/object_keys split that carries object sizes, fixing the prefix-filter bug that always returned 0). Co-authored-by: Cursor <cursoragent@cursor.com>
viktorsomogyi
changed the base branch from
main
to
svv/ts-unification-delete-records-e2e
July 22, 2026 14:34
viktorsomogyi
force-pushed
the
svv/ts-unification-retention-bytes-test
branch
from
July 22, 2026 14:35
fd4d0f0 to
659293f
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-delete-records-e2e
branch
6 times, most recently
from
July 24, 2026 14:30
c617830 to
9bc2388
Compare
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.
Adds consolidation_retention_bytes_across_tiers_test.py, exercising size-based (retention.bytes) reclaim of the cross-tier earliest offset on a consolidating diskless topic. Adds the verifier helpers it needs: tiered_object_bytes (fixes the prefix-filter bug that always returned 0), wait_for_tiered_count_stable, and the _object_entries/object_keys split.