test(inkless:consolidation): retention.bytes cross-tier reclaim system test [KC-332] - #708
Draft
viktorsomogyi wants to merge 2 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))) | ||
|
|
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
10 times, most recently
from
July 29, 2026 07:56
0878b80 to
3f7ea8d
Compare
viktorsomogyi
force-pushed
the
svv/ts-unification-retention-bytes-test
branch
from
July 30, 2026 13:03
659293f to
e0de0f9
Compare
…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
force-pushed
the
svv/ts-unification-retention-bytes-test
branch
from
July 30, 2026 14:21
e0de0f9 to
a5d0f8b
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.