test(inkless:consolidation): retention.bytes cross-tier reclaim system test [KC-332]

[viktorsomogyi]

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.

[Copilot]

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-tier DeleteRecords behavior (including failover).
• Extend ConsolidationVerifier with MinIO byte accounting, “stable” tiered-object counting, per-broker EARLIEST probing, and a kafka-delete-records.sh wrapper.
• Adjust broker/runtime behavior so consolidating topics use the control-plane cross-tier earliest for ListOffsets(EARLIEST), for DeleteRecords low-watermark reporting, and as the RemoteLogManager reclaim floor; add an inter-broker forwarder for the leader-only DeleteRecords leg.

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.