From 3f7ea8dbfa62f6202b39fdd085ceaad7715a66a2 Mon Sep 17 00:00:00 2001 From: Viktor Somogyi-Vass Date: Tue, 21 Jul 2026 16:09:46 +0200 Subject: [PATCH] test(inkless:consolidation): DeleteRecords across tiers e2e test [KC-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 --- .../inkless/consolidation_verifier.py | 275 ++++++- .../consolidation_cross_tier_earliest_test.py | 169 ++++ ...dation_delete_records_across_tiers_test.py | 741 ++++++++++++++++++ ...nsolidation_retention_across_tiers_test.py | 114 +++ 4 files changed, 1291 insertions(+), 8 deletions(-) create mode 100644 tests/kafkatest/tests/inkless/consolidation_cross_tier_earliest_test.py create mode 100644 tests/kafkatest/tests/inkless/consolidation_delete_records_across_tiers_test.py diff --git a/tests/kafkatest/services/inkless/consolidation_verifier.py b/tests/kafkatest/services/inkless/consolidation_verifier.py index ed33119c674..8af4a2b09a0 100644 --- a/tests/kafkatest/services/inkless/consolidation_verifier.py +++ b/tests/kafkatest/services/inkless/consolidation_verifier.py @@ -32,6 +32,7 @@ from ducktape.utils.util import wait_until from kafkatest.services.verifiable_producer import VerifiableProducer +from kafkatest.version import V_2_2_2 class ConsolidationVerifier(object): @@ -78,6 +79,13 @@ class ConsolidationVerifier(object): # name a test uses to fault them (see block_dependency / heal_dependency). DEP_ALIASES = {"object_storage": "storage", "control_plane": "postgres"} + # Oldest consumer the ducker image ships (tests/docker/Dockerfile installs 2.2.2). + # Its Fetch predates KIP-392 / Fetch v11, so it carries no clientMetadata -- the + # isOlderConsumer branch the follower cross-tier-earliest read guard protects. A + # modern (>= 2.3) client always sends clientMetadata and takes a different branch, + # so it cannot exercise that guard. + OLD_CONSUMER_VERSION = V_2_2_2 + def __init__(self, kafka): self.kafka = kafka self.logger = kafka.logger @@ -253,6 +261,31 @@ def wal_object_count(self): # only, never absolute. return len([k for k in self.object_keys() if not k.startswith(self.TIERED_PREFIX)]) + def wait_for_tiered_count_stable(self, settle_samples=3, backoff_sec=5, timeout_sec=300): + """Wait until the tiered-storage object count stops growing (held steady across + ``settle_samples`` consecutive samples) and return that stable count. + + Consolidation/RLM keeps copying closed segments to remote after the WAL has + drained locally, so a count snapshot taken right after ``wait_for_tiered_offset_at_least`` + can still be climbing. Tests that later assert the count *dropped* after a + reclaim need a settled pre-reclaim peak, otherwise late tiering can mask the + deletion. Under a long retention nothing removes tiered objects, so the count + only grows and then plateaus.""" + 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 + + wait_until(check, timeout_sec=timeout_sec, backoff_sec=backoff_sec, + err_msg="Tiered-storage object count did not stabilize within %ds." % timeout_sec) + return state["value"] + # --------------------- Control plane --------------------- def _psql(self, sql): @@ -293,6 +326,37 @@ def min_log_start_offset(self, topic): "SELECT coalesce(min(log_start_offset), 0) FROM logs WHERE topic_name = %s" % self._sql_literal(topic)) + def remote_log_start_offset(self, topic, partition=0): + """Raw ``logs.remote_log_start_offset`` for the partition, or ``None`` when it is + still NULL (never reported). Distinct from ``min_log_start_offset`` (the WAL prune + frontier): ``ListOffsets(EARLIEST)`` is + ``COALESCE(remote_log_start_offset, log_start_offset)``, so a NULL here makes + EARLIEST fall back to the pruned frontier and skip the still-live remote prefix.""" + out = self._psql( + "SELECT remote_log_start_offset FROM logs WHERE topic_name = %s AND partition = %d" + % (self._sql_literal(topic), partition)) + first = out.splitlines()[0].strip() if out else "" + return None if first == "" else int(first) + + def wait_for_remote_log_start_bootstrapped(self, topic, expected=0, partition=0, timeout_sec=180): + """Wait until ``remote_log_start_offset`` is reported (non-NULL) and equals + ``expected``, and return it. This is the become-leader bootstrap that keeps + EARLIEST pinned to the true remote earliest once the WAL prune frontier has run + ahead; a value stuck at NULL is the data-loss regression it guards against.""" + state = {"value": "unset"} + + def check(): + state["value"] = self.remote_log_start_offset(topic, partition) + self.logger.info("Topic %s-%d remote_log_start_offset: %s (want %d)" + % (topic, partition, state["value"], expected)) + return state["value"] is not None and state["value"] == expected + + wait_until(check, timeout_sec=timeout_sec, backoff_sec=2, + err_msg=("remote_log_start_offset for %s-%d was not bootstrapped to %d " + "(still %s); a NULL leaves EARLIEST COALESCE'd to the pruned WAL " + "frontier" % (topic, partition, expected, state["value"]))) + return state["value"] + # --------------------- Tooling --------------------- def verify_tooling(self): @@ -329,6 +393,28 @@ def create_classic_topic(self, topic, num_partitions, replication_factor, }, }) + def create_consolidated_topic(self, topic, num_partitions, replication_factor, + min_isr=2, segment_bytes=2 * 1024 * 1024, segment_ms=5000): + """Create a *born-consolidated* diskless topic: diskless + remote storage are + enabled at creation (no classic phase, no switch, so no seal). This is accepted + because the cluster runs with ``diskless.allow.from.classic.enable=true``, which + also opens diskless+remote at create time. ``segment.bytes`` defaults to 2 MiB + (> the 1 MiB batch ceiling so a consolidation append does not trip + ``RecordBatchTooLargeException``) and ``segment.ms`` to 5s, so a produced stream + rolls closed, tierable segments.""" + self.kafka.create_topic({ + "topic": topic, + "partitions": num_partitions, + "replication-factor": replication_factor, + "configs": { + "min.insync.replicas": min_isr, + "segment.bytes": segment_bytes, + "segment.ms": segment_ms, + "diskless.enable": "true", + "remote.storage.enable": "true", + }, + }) + def produce(self, topic, num_records, label="", timeout_sec=180, throughput=-1): """Produce ``num_records`` to ``topic`` with a one-shot ``VerifiableProducer`` and return the acked count, asserting there were no worker errors. The @@ -445,14 +531,26 @@ def partition_has_leader(self, topic, partition=0): self.logger.debug("Leader lookup for %s-%d not ready yet: %s" % (topic, partition, e)) return False - def offset_at(self, topic, time_spec, partition=0): + def _broker_bootstrap(self, node): + """``hostname:port`` for a single broker's client listener, so a query can be + pinned to one broker instead of the whole ``bootstrap_servers()`` list.""" + protocol = self.kafka.security_protocol + port = self.kafka.port_mappings[protocol].port_number + return "%s:%s" % (node.account.hostname, port) + + def offset_at(self, topic, time_spec, partition=0, bootstrap_server=None): """Offset for the partition at a ``kafka-get-offsets.sh --time`` spec (-1 latest, -2 earliest, -4 earliest-local, -5 latest-tiered). Returns -1 if - not parseable yet.""" + not parseable yet. + + ``bootstrap_server`` pins the query to a single broker (see + ``_broker_bootstrap``); by default the whole cluster is used. Pinning lets a + test detect a per-broker divergence in the answer (e.g. ListOffsets(EARLIEST) + served inconsistently across brokers).""" node = self.kafka.nodes[0] cmd = "%s --bootstrap-server %s --topic %s --partitions %d --time %s" % ( self.kafka.path.script("kafka-get-offsets.sh", node), - self.kafka.bootstrap_servers(), + bootstrap_server or self.kafka.bootstrap_servers(), topic, partition, time_spec, @@ -566,6 +664,53 @@ def check(): % (topic, partition, timeout_sec))) return state["value"] + def earliest_on_each_broker(self, topic, partition=0): + """Query ListOffsets(EARLIEST) once against each broker individually and return + a ``{hostname: offset}`` map. Before the ``DisklessFetchOffsetRouter`` fix a + follower served ListOffsets(EARLIEST) from its frozen-at-switch local classic + log while the leader served the advanced value, so the answer diverged per + broker. After the fix every broker routes to the control plane and returns the + same cross-tier earliest.""" + result = {} + for node in self.kafka.nodes: + result[node.account.hostname] = self.offset_at( + topic, time_spec=-2, partition=partition, + bootstrap_server=self._broker_bootstrap(node)) + return result + + def wait_for_consistent_earliest_across_brokers(self, topic, partition=0, + settle_samples=3, backoff_sec=5, + timeout_sec=240): + """Wait until ListOffsets(EARLIEST) is (a) > 0, (b) identical on every broker, + and (c) held steady across ``settle_samples`` consecutive rounds; return that + agreed value. + + This is the end-to-end cross-broker consistency assertion: the earliest must be one + broker-agnostic value (the control-plane cross-tier earliest), not a per-broker + answer that depends on which replica the metadata transformer routed the client + to. A single positive value on every broker, stable over time, means the router + no longer serves it from a replica's local classic log.""" + state = {"last": None, "stable": 0, "value": 0} + + 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 + + wait_until(check, timeout_sec=timeout_sec, backoff_sec=backoff_sec, + err_msg=("ListOffsets(EARLIEST) for %s-%d did not converge to a single " + "positive value across all brokers within %ds (per-broker " + "earliest divergence)" % (topic, partition, timeout_sec))) + return state["value"] + def read_contiguous_from(self, topic, from_offset=0, max_messages=1, partition=0, timeout_ms=120000): """Fetch up to ``max_messages`` records from ``from_offset`` and return @@ -613,6 +758,62 @@ def first_served_offset(self, topic, partition=0, from_offset=0, timeout_ms=1200 timeout_ms=timeout_ms) return first + def delete_records(self, topic, before_offset, partition=0): + """Run ``kafka-delete-records.sh`` to delete every record before + ``before_offset`` on ``topic``-``partition`` and return the resulting + ``low_watermark`` (the new log start) for that partition. + + ``DeleteRecords`` takes a *beforeOffset*: the given offset becomes the new + log start, so records ``[0, before_offset)`` are removed. On a consolidating + topic the broker splits the request across tiers -- local ``UnifiedLog`` (which + drives ``RemoteLogManager`` deletion of the remote segments below the new start) + and, past the local log end, the diskless WAL in the control plane. The offset + JSON file is written on the broker node and passed with ``--offset-json-file``. + + ``DeleteRecordsCommand`` prints the per-partition outcome on stdout as either + ``partition: \\terror: `` or ``partition: \\tlow_watermark: ``. + Success/failure is decided purely from that output, NOT from the process exit + code: older ``kafka-delete-records.sh`` exits 0 even when a partition failed, so + ``ssh_capture`` runs with ``allow_fail=True`` and we treat an ``error`` line for + the target partition (or a missing ``low_watermark``) as failure. This keeps the + test independent of whether the exit-code fix is present.""" + node = self.kafka.nodes[0] + offset_json = ('{"partitions":[{"topic":"%s","partition":%d,"offset":%d}],"version":1}' + % (topic, partition, before_offset)) + json_path = "/tmp/delete-records-%s-%d.json" % (topic, partition) + node.account.create_file(json_path, offset_json) + cmd = "%s --bootstrap-server %s --offset-json-file %s" % ( + self.kafka.path.script("kafka-delete-records.sh", node), + self.kafka.bootstrap_servers(), + json_path, + ) + self.logger.info("Deleting records before offset %d on %s-%d: %s" + % (before_offset, topic, partition, cmd)) + # allow_fail=True: don't rely on the exit code (older CLIs exit 0 on per-partition + # failure); the per-partition result is parsed from stdout below instead. + output = "" + for line in node.account.ssh_capture(cmd, allow_fail=True): + output += line.decode("utf-8") if isinstance(line, bytes) else line + self.logger.info("kafka-delete-records.sh output:\n%s" % output) + + target = "%s-%d" % (topic, partition) + low_watermark = None + for line in output.splitlines(): + line = line.strip() + err = re.search(r"partition:\s*(\S+)\s+error:\s*(.+)$", line) + if err and err.group(1) == target: + raise AssertionError( + "DeleteRecords before offset %d on %s failed: %s" + % (before_offset, target, err.group(2))) + lw = re.search(r"partition:\s*(\S+)\s+low_watermark:\s*(-?\d+)$", line) + if lw and lw.group(1) == target: + low_watermark = int(lw.group(2)) + assert low_watermark is not None, ( + "DeleteRecords before offset %d on %s returned no low_watermark; output was:\n%s" + % (before_offset, target, output)) + self.logger.info("DeleteRecords on %s returned low_watermark=%d" % (target, low_watermark)) + return low_watermark + # --------------------- Dependency outage (iptables) --------------------- # # The object store ("storage") and control plane ("postgres") are standalone @@ -677,19 +878,26 @@ def _drop_rule(node, ip): "sudo iptables -D OUTPUT -d %s -j DROP; done" % (ip, ip), allow_fail=True) def read_records_with_values_from(self, topic, from_offset, max_messages, partition=0, - timeout_ms=120000): - """Fetch up to ``max_messages`` records from ``from_offset`` and return them as a - list of ``(offset, value)`` tuples in returned order. + timeout_ms=120000, consumer_properties=None): + """Fetch up to ``max_messages`` records from ``from_offset`` with the current + (modern) console consumer and return them as a list of ``(offset, value)`` tuples + in returned order. The VerifiableProducer default (``is_int``) writes each record's per-producer sequence number as the value, so callers can assert payload content -- not just offsets -- and detect gaps, duplicates, or reordering across the classic->diskless boundary. Each output line is ``Offset:\\t``: DefaultMessageFormatter prints the offset prefix, then key.separator '\\t', then - the value.""" + the value. + + ``consumer_properties`` is an optional ``{k: v}`` of ``--consumer-property`` + settings (e.g. ``{"auto.offset.reset": "earliest"}`` so an out-of-range seek + resets to the earliest readable offset instead of the log end).""" node = self.kafka.nodes[0] + props = "".join(" --consumer-property %s=%s" % (k, v) + for k, v in (consumer_properties or {}).items()) cmd = ("%s --bootstrap-server %s --topic %s --partition %d --offset %d " - "--max-messages %d --timeout-ms %d " + "--max-messages %d --timeout-ms %d%s " "--property print.offset=true --property print.key=false " "--property print.value=true" % ( self.kafka.path.script("kafka-console-consumer.sh", node), @@ -699,6 +907,7 @@ def read_records_with_values_from(self, topic, from_offset, max_messages, partit from_offset, max_messages, timeout_ms, + props, )) records = [] try: @@ -711,3 +920,53 @@ def read_records_with_values_from(self, topic, from_offset, max_messages, partit self.logger.warn("Failed to read records from %s-%d at offset %d: %s" % (topic, partition, from_offset, str(e))) return records + + def read_values_with_old_client(self, topic, from_offset, max_messages, partition=0, + timeout_ms=120000, bootstrap_server=None): + """Run the pre-KIP-392 console consumer (Kafka 2.2, Fetch < v11) so the read carries no + ``clientMetadata`` -- the ``isOlderConsumer`` fetch path the follower + cross-tier-earliest read guard protects. Returns the list of record *values* read, in + order. + + Values, not offsets: the 2.2 ``DefaultMessageFormatter`` predates KIP-431, so it has no + ``print.offset``/``print.partition`` -- only the value can be printed (value prints by + default). That is enough here because ``VerifiableProducer``'s default (``is_int``) writes + each record's per-producer sequence as the value, and the classic producer wrote + ``[0, seal)`` so in the surviving classic prefix ``value == offset``. Callers therefore + treat the returned values as offsets (only valid below the seal). + + Assign mode (``--partition``/``--offset``) is deliberate: no consumer group or coordinator + is involved, so the exchange is only Metadata + ListOffsets + Fetch, all + backward-compatible with a modern broker. ``auto.offset.reset=earliest`` makes an + out-of-range seek (below the cross-tier earliest) reset to the earliest readable offset + rather than jump to the log end, so the test can tell "reset to the delete boundary" apart + from "served a logically-deleted record". + + ``bootstrap_server`` pins the initial metadata lookup to one broker; the metadata + transformer still redirects the fetch to its hash-selected replica regardless.""" + node = self.kafka.nodes[0] + script = self.kafka.path.script("kafka-console-consumer.sh", self.OLD_CONSUMER_VERSION) + cmd = ("%s --bootstrap-server %s --topic %s --partition %d --offset %d " + "--max-messages %d --timeout-ms %d " + "--consumer-property auto.offset.reset=earliest " + "--property print.key=false --property print.value=true" % ( + script, + bootstrap_server or self.kafka.bootstrap_servers(), + topic, + partition, + from_offset, + max_messages, + timeout_ms, + )) + values = [] + try: + for line in node.account.ssh_capture(cmd, allow_fail=True): + line = line.decode("utf-8") if isinstance(line, bytes) else line + stripped = line.strip() + # 2.2 prints just the value per line (no offset prefix); records are integers. + if re.match(r"-?\d+$", stripped): + values.append(int(stripped)) + except Exception as e: # noqa: BLE001 - best effort; tool may time out + self.logger.warn("Old-client read from %s-%d at offset %d failed: %s" + % (topic, partition, from_offset, str(e))) + return values diff --git a/tests/kafkatest/tests/inkless/consolidation_cross_tier_earliest_test.py b/tests/kafkatest/tests/inkless/consolidation_cross_tier_earliest_test.py new file mode 100644 index 00000000000..42463c78840 --- /dev/null +++ b/tests/kafkatest/tests/inkless/consolidation_cross_tier_earliest_test.py @@ -0,0 +1,169 @@ +# Inkless +# Copyright (C) 2024 - 2026 Aiven OY +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +import uuid + +from ducktape.mark import matrix +from ducktape.mark.resource import cluster +from ducktape.tests.test import Test +from ducktape.utils.util import wait_until + +from kafkatest.services.kafka import KafkaService, quorum +from kafkatest.services.inkless.consolidation_verifier import ConsolidationVerifier + + +class CrossTierEarliestBootstrapTest(Test): + """A consolidating topic must keep ``ListOffsets(EARLIEST)`` pinned to the true + cross-tier earliest after the WAL prune frontier has advanced well past it. + + EARLIEST is served from the control plane as + ``COALESCE(remote_log_start_offset, log_start_offset)``. ``log_start_offset`` is the + diskless WAL prune frontier and climbs as batches tier and are pruned; + ``remote_log_start_offset`` is the true lowest offset still readable from the remote + tier and is nullable. If it were left NULL, EARLIEST would COALESCE to the pruned + frontier and a consumer with ``auto.offset.reset=earliest`` would silently skip the + still-live remote prefix, i.e. lose data. The become-leader report bootstraps + ``remote_log_start_offset`` to the true remote earliest (0 for a freshly-tiered + born-consolidated topic) so this cannot happen. + + This proves the invariant end to end: produce, tier, and prune the WAL well past 0, + then assert the control plane holds ``remote_log_start_offset = 0`` (never NULL), + EARLIEST is 0 on every broker despite the advanced frontier, and offset 0 is still + served from the remote tier with the correct content. + + The genuine reporter/become-leader *fault* path (``remote_log_start_offset`` stuck + NULL while the WAL is pruned) is left to unit coverage + (``AbstractControlPlaneTest.getCrossTierLogStart*``, the ``RemoteLogManager`` + become-leader tests): on this code both the become-leader report and the reporter + re-populate the value, and blackholing the control plane to suppress them also stalls + the WAL pruner and the EARLIEST read itself, so the NULL-and-pruned state is not + observable from a system test. The positive assertion below is the regression guard + for the exact bootstrap the fix relies on.""" + + # Unique per run: the Postgres/MinIO containers persist across runs, so a stale + # topic name would let old rows/objects skew the control-plane and mc queries. + TOPIC_PREFIX = "cross-tier-earliest-bootstrap" + NUM_PARTITIONS = 1 + REPLICATION_FACTOR = 3 + # Enough to roll many closed 1 MiB segments so consolidation tiers a remote prefix and + # the WAL pruner advances log_start_offset well past 0 (the frontier this test needs + # high for the "EARLIEST stays 0" assertion to be non-vacuous). + NUM_RECORDS = 300000 + # How many records from offset 0 to spot-check for readability/content. + SPOT_CHECK = 20000 + + def __init__(self, test_context): + super(CrossTierEarliestBootstrapTest, self).__init__(test_context=test_context) + self.num_brokers = 3 + self.TOPIC = "%s-%s" % (self.TOPIC_PREFIX, uuid.uuid4().hex[:8]) + + def _start_cluster(self): + self.kafka = KafkaService( + self.test_context, + num_nodes=self.num_brokers, + zk=None, + controller_num_nodes_override=1, + consolidation=True, + # Run the WAL pruner / file cleaner / remote-log task fast so the frontier + # advances within the test window (not the default minutes). + server_prop_overrides=[ + ["inkless.consolidation.cleanup.interval.ms", "5000"], + ["inkless.file.cleaner.interval.ms", "5000"], + ["inkless.file.cleaner.retention.period.ms", "6000"], + ["inkless.consume.batch.coordinate.cache.ttl.ms", "2000"], + ["remote.log.manager.task.interval.ms", "5000"], + ["log.retention.check.interval.ms", "5000"], + ], + topics={ + self.TOPIC: { + "partitions": self.NUM_PARTITIONS, + "replication-factor": self.REPLICATION_FACTOR, + "configs": { + "diskless.enable": "true", + "remote.storage.enable": "true", + "min.insync.replicas": 2, + # Roll segments by size/time so they close and get tiered. + "segment.bytes": 1048576, + "segment.ms": 5000, + # Evict local copies soon after upload so the early prefix lives + # only in remote and the read from 0 is a genuine cross-tier read. + "local.retention.ms": 5000, + }, + }, + }, + ) + self.kafka.start() + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_cross_tier_earliest_survives_wal_prune(self, metadata_quorum): + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + verifier.start_jmx() + baseline_tiered = verifier.tiered_object_count() + + acked = verifier.produce(self.TOPIC, self.NUM_RECORDS, "bootstrap") + self.logger.info("Produced born-consolidated stream: acked=%d" % acked) + + wait_until(lambda: verifier.tiered_object_count() > baseline_tiered, + timeout_sec=240, backoff_sec=2, + err_msg="Consolidation never tiered a remote prefix.") + + # Drive the WAL prune frontier well past 0 so "EARLIEST stays at 0" is a real + # signal, not vacuously 0 because the frontier is also 0. + wait_until(lambda: verifier.min_log_start_offset(self.TOPIC) > 0, + timeout_sec=240, backoff_sec=2, + err_msg="WAL was never pruned in the control plane; frontier stayed at 0.") + frontier = verifier.min_log_start_offset(self.TOPIC) + assert frontier > 0, "WAL prune frontier did not advance past 0" + self.logger.info("WAL prune frontier advanced to log_start_offset=%d" % frontier) + + # 1) The control plane holds remote_log_start_offset = 0 (bootstrapped, never NULL). + remote_start = verifier.wait_for_remote_log_start_bootstrapped(self.TOPIC, expected=0) + self.logger.info("Control plane remote_log_start_offset bootstrapped to %d (frontier=%d)" + % (remote_start, frontier)) + + # 2) EARLIEST is 0 on every broker even though the frontier is at %d: EARLIEST is + # COALESCE(remote_log_start_offset=0, log_start_offset=frontier) = 0. A NULL + # remote start would make it jump to the frontier and skip the remote prefix. + per_broker = verifier.earliest_on_each_broker(self.TOPIC) + assert set(per_broker.values()) == {0}, ( + "EARLIEST is not 0 on every broker despite remote_log_start_offset=0 and a WAL frontier " + "at %d: %s (EARLIEST fell back to the pruned frontier, so the remote prefix [0, %d) " + "would be silently skipped)" % (frontier, per_broker, frontier)) + self.logger.info("EARLIEST agreed at 0 across all brokers with the frontier at %d" % frontier) + + # 3) Offset 0 is actually served from the remote tier, contiguous with correct content + # (VerifiableProducer writes value == offset for a single born-consolidated run). + first_served = verifier.first_served_offset(self.TOPIC, from_offset=0) + assert first_served == 0, ( + "a fetch from offset 0 returned offset %d; the tiered prefix below the WAL frontier %d " + "was not served" % (first_served, frontier)) + spot = min(acked, self.SPOT_CHECK) + records = verifier.read_records_with_values_from( + self.TOPIC, from_offset=0, max_messages=spot, timeout_ms=240000) + assert len(records) >= spot, ( + "bounded read from offset 0 returned only %d of %d records; the remote prefix below the " + "WAL frontier %d was not fully served" % (len(records), spot, frontier)) + for i, (offset, value) in enumerate(records[:spot]): + assert offset == i, ( + "non-contiguous read at position %d: offset=%d, expected %d (gap/dupe/reorder)" + % (i, offset, i)) + assert value == offset, ( + "content mismatch at offset %d: value=%d, expected %d" % (offset, value, offset)) + self.logger.info("Remote prefix from 0 read back contiguous (%d records) below frontier %d" + % (spot, frontier)) diff --git a/tests/kafkatest/tests/inkless/consolidation_delete_records_across_tiers_test.py b/tests/kafkatest/tests/inkless/consolidation_delete_records_across_tiers_test.py new file mode 100644 index 00000000000..e0f7b73152c --- /dev/null +++ b/tests/kafkatest/tests/inkless/consolidation_delete_records_across_tiers_test.py @@ -0,0 +1,741 @@ +# Inkless +# Copyright (C) 2024 - 2026 Aiven OY +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +import time +import uuid + +from ducktape.mark import matrix +from ducktape.mark.resource import cluster +from ducktape.tests.test import Test +from ducktape.utils.util import wait_until + +from kafkatest.services.kafka import KafkaService, quorum +from kafkatest.services.inkless.consolidation_verifier import ConsolidationVerifier + + +class DeleteRecordsAcrossTiersTest(Test): + """``DeleteRecords`` on a *switched* (hybrid) consolidating topic must advance the + topic's earliest readable offset to *exactly* the requested boundary, in a way every + broker agrees on, and must not reclaim any records the caller did not ask to delete. + This guards two distinct behaviors: cross-broker EARLIEST consistency and no + over-reclaim of the classic prefix. + + Cross-broker EARLIEST consistency. ``ListOffsets(EARLIEST)`` for a + consolidating partition used to be served from the broker-local classic + ``UnifiedLog.logStartOffset`` while it was still below the seal. With managed + replicas enabled (``diskless.managed.rf.enable=true``, which consolidation + requires) ``InklessTopicMetadataTransformer`` advertises a *hash-selected replica* + as the partition leader for locality-aware reads -- deterministically the same + replica for every client, and generally a *follower*, not the real KRaft leader + that holds the writable log. That follower's local classic log start is frozen at + the switch, so a ``DeleteRecords`` (which advances only the real leader's log start + and the control plane) left the client-visible earliest stuck at ``0`` forever. The + fix routes ``ListOffsets(EARLIEST)`` for consolidating topics to the control plane + on *every* broker (``COALESCE(remote_log_start_offset, log_start_offset)`` from + ``list_offsets_v1``), which the classic leader keeps current, so the earliest + advances on whichever replica the transformer routed the client to, and is one + broker-agnostic value. + + No over-reclaim of the classic prefix. With retention unset, a + ``DeleteRecords`` before ``delete_before < S`` must remove only ``[0, delete_before)`` + and leave ``[delete_before, S)`` readable, so the earliest settles at *exactly* + ``delete_before``. The consolidation cleanup used to reclaim the whole classic prefix + ``[0, S)`` down to the seal instead. On a leader that rebuilt its log from the remote + tier, the broker-local whole-log start (``UnifiedLog.logStartOffset``, which is a + per-replica field and is not the local-segment start ``localLogStartOffset``) sits at + ``S`` and only ever increments, so the RLM remote-retention reclaim floor and the + become-leader report both read ``S`` and treated it as the cross-tier earliest. The fix + drives both the consolidating partition's + whole-log start (``DisklessLeaderEndPoint``) and the RLM reclaim floor / + become-leader report (``RemoteLogManager``) from the broker-agnostic control-plane + cross-tier earliest via ``ReplicaManager.crossTierEarliestOffset``. + + The scenario: produce a classic prefix, switch to a consolidating diskless topic + (sealing at ``S``), produce + consolidate a diskless tail, and let the classic + prefix tier to remote with its local copies evicted -- so the prefix ``[0, S)`` + lives in the remote tier. ``retention.ms``/``retention.bytes`` are held unset + throughout so the only thing that can move the earliest offset is the + ``DeleteRecords`` request. A ``DeleteRecords`` before ``delete_before = S // 2`` + (inside that remote prefix) must then: + + - return the requested boundary as the partition's new log start, + - advance the client-visible earliest to *exactly* ``delete_before``, identical on + every broker (served from the control plane, not a replica's local log), and + - leave the surviving prefix ``[delete_before, S)`` readable and contiguous with the + originally produced content. + + Leader routing for the write side of ``DeleteRecords``: the op is leader-routed and + its local leg requires the real KRaft leader, so the broker the transformer points + the admin client at forwards the affected partitions to their real leader + (``DisklessDeleteRecordsForwarder``); the leader runs both the local-log and the + diskless (control-plane) legs authoritatively. + """ + + # Unique per run: the Postgres/MinIO containers persist across runs, so a stale + # topic name would let old rows/objects skew the control-plane and mc queries. + TOPIC_PREFIX = "delete-records-across-tiers" + NUM_PARTITIONS = 1 + REPLICATION_FACTOR = 3 + # Classic prefix. segment.bytes is floored at 1 MiB and records are ~13 B, so a + # closed classic segment holds ~80k records. The delete boundary (seal // 2) must + # sit strictly inside the sealed classic region so the reclaimed range is + # remote-backed and the classic log start stays below the seal. Sizing it this way + # also clears whole ~1 MiB segments below the boundary while leaving several whole + # surviving segments in [delete_before, seal) -- so a correct implementation drops + # only the reclaimed ones and keeps the rest (the over-reclaim guard). + NUM_CLASSIC_RECORDS = 400000 + # Diskless tail: ~13 B/record => ~4 MiB, exceeding segment.bytes (2 MiB) so the + # tail rolls an inactive segment. Only rolled segments tier, so this is what lifts + # highestOffsetInRemoteStorage past the seal and lets the WAL pruner advance. + NUM_DISKLESS_RECORDS = 300000 + # How many surviving-prefix records to spot-check for readability/content. + SPOT_CHECK = 20000 + + def __init__(self, test_context): + super(DeleteRecordsAcrossTiersTest, self).__init__(test_context=test_context) + self.num_brokers = 3 + self.TOPIC = "%s-%s" % (self.TOPIC_PREFIX, uuid.uuid4().hex[:8]) + + def _start_cluster(self): + self.kafka = KafkaService( + self.test_context, + num_nodes=self.num_brokers, + zk=None, + controller_num_nodes_override=1, + consolidation=True, + # log.diskless.enable=false: keep the cluster default classic (else new + # topics are born diskless and trip the diskless/remote validator). + # diskless.allow.from.classic.enable=true: open the classic->diskless + # switch bridge (validated on every node). + # The rest run the WAL pruner / file cleaner / remote-log task fast. + server_prop_overrides=[ + ["log.diskless.enable", "false"], + ["diskless.allow.from.classic.enable", "true"], + ["inkless.consolidation.cleanup.interval.ms", "5000"], + ["inkless.file.cleaner.interval.ms", "5000"], + ["inkless.file.cleaner.retention.period.ms", "6000"], + ["inkless.consume.batch.coordinate.cache.ttl.ms", "2000"], + ["remote.log.manager.task.interval.ms", "5000"], + ["log.retention.check.interval.ms", "5000"], + ], + # Switch-completion gauges live on the broker; the controller never + # exposes them, so scraping it would hang start_jmx_tool's --wait. + jmx_object_names=list(ConsolidationVerifier.SWITCH_JMX_OBJECT_NAMES), + jmx_attributes=list(ConsolidationVerifier.SWITCH_JMX_ATTRIBUTES), + ) + # server_prop_overrides only reach the brokers, but the controller also + # validates the switch bridge: mirror it onto the controller and disable + # JMX there (it has no switch gauges). + if getattr(self.kafka, "isolated_controller_quorum", None): + ctrl = self.kafka.isolated_controller_quorum + ctrl.server_prop_overrides = list(ctrl.server_prop_overrides) + [ + ["log.diskless.enable", "false"], + ["diskless.allow.from.classic.enable", "true"], + ] + ctrl.jmx_object_names = None + ctrl.jmx_attributes = [] + self.kafka.start() + + def _switch_and_drain(self, verifier, baseline_tiered): + """Produce a classic prefix, switch to consolidating diskless, tier the classic + prefix to remote (with local copies evicted), then produce + consolidate a + diskless tail and wait until the remote tier covers through the seal. Returns + ``(seal_offset, total_acked)``.""" + # --- 1) Classic phase: produce a classic prefix on a plain classic topic --- + verifier.create_classic_topic(self.TOPIC, self.NUM_PARTITIONS, self.REPLICATION_FACTOR) + classic_acked = verifier.produce(self.TOPIC, self.NUM_CLASSIC_RECORDS, "classic") + seal_offset = classic_acked + self.logger.info("Produced classic prefix: acked=%d (seal offset will be %d)" + % (classic_acked, seal_offset)) + + # --- 2) Switch the classic topic straight to consolidating diskless --- + # + # One combined alter: the leader seals the classic log at `seal_offset` and + # starts the consolidation fetcher; short local.retention.ms lets the tiered + # classic prefix be deleted locally. segment.bytes = 2 MiB (> the 1 MiB batch + # ceiling so a consolidation append does not trip RecordBatchTooLargeException, + # and < the diskless tail so the tail rolls an inactive, tierable segment). + # retention.ms/.bytes are left at their (long/unlimited) defaults so only the + # DeleteRecords below can move the earliest offset. + self.logger.info("Switching topic %s to consolidating diskless (combined alter)" % self.TOPIC) + self.kafka.alter_topic_configs(self.TOPIC, { + "diskless.enable": "true", + "remote.storage.enable": "true", + "local.retention.ms": "5000", + "segment.bytes": str(2 * 1024 * 1024), + }) + verifier.wait_for_switch_complete(self.TOPIC, self.NUM_PARTITIONS) + self.logger.info("Classic-to-consolidated switch completed for %s" % self.TOPIC) + + # --- 3) Tier the classic prefix and evict the local copies --- + wait_until(lambda: verifier.tiered_object_count() > baseline_tiered, + timeout_sec=180, backoff_sec=2, + err_msg="Classic tiered-storage object count did not grow above baseline " + "after enabling remote storage on the switched topic.") + classic_remote_watermark = verifier.wait_for_local_log_truncation(self.TOPIC) + assert classic_remote_watermark > 0, \ + "Classic prefix never tiered: earliest-local stayed at 0 after enabling remote storage." + self.logger.info("Classic prefix tiered; earliest-local watermark=%d" % classic_remote_watermark) + + # --- 4) Diskless phase: produce the tail and consolidate it to remote --- + pre_diskless_tiered = verifier.tiered_object_count() + diskless_acked = verifier.produce(self.TOPIC, self.NUM_DISKLESS_RECORDS, "diskless") + self.logger.info("Produced diskless tail: acked=%d" % diskless_acked) + + wait_until(lambda: verifier.tiered_object_count() > pre_diskless_tiered, + timeout_sec=180, backoff_sec=2, + err_msg="Diskless consolidation did not grow the tiered-storage object count.") + + # The whole classic prefix [0, seal) -- including the boundary tail -- is durable + # in remote once the latest-tiered offset reaches the seal, so the delete boundary + # we pick inside the prefix is remote-backed. + remote_watermark = verifier.wait_for_tiered_offset_at_least(self.TOPIC, seal_offset) + self.logger.info("Remote tier covers through the seal: latest-tiered=%d, seal=%d" + % (remote_watermark, seal_offset)) + + total_acked = classic_acked + diskless_acked + return seal_offset, total_acked + + def _rolling_restart(self, verifier): + """Clean-restart every broker one at a time, waiting for the partition to regain a + leader and full ISR before moving to the next. A clean restart reloads each broker's + on-disk log and its checkpointed whole-log start, so this does not force a + rebuild-from-remote (see ``test_delete_records_survives_leader_rebuild_from_remote`` + for that). What it does exercise is leadership moving across the cluster: each + surviving replica takes over as partition leader and RLM leader, re-runs the + become-leader report, and drives a reclaim cycle. That is the leadership-change / + metadata-propagation-timing path the reclaim floor and the report must survive + without over-reclaiming.""" + for node in list(self.kafka.nodes): + self.logger.info("Rolling restart: bouncing %s" % node.account.hostname) + self.kafka.restart_node(node, clean_shutdown=True) + wait_until(lambda: verifier.partition_has_leader(self.TOPIC), + timeout_sec=120, backoff_sec=2, + err_msg="partition %s-0 had no leader after restarting %s" + % (self.TOPIC, node.account.hostname)) + wait_until(lambda: len(self.kafka.isr_idx_list(self.TOPIC, 0)) >= self.REPLICATION_FACTOR, + timeout_sec=180, backoff_sec=3, + err_msg="ISR for %s-0 did not recover to %d after restarting %s" + % (self.TOPIC, self.REPLICATION_FACTOR, node.account.hostname)) + self.logger.info("Broker %s back in ISR for %s-0" % (node.account.hostname, self.TOPIC)) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_survives_leader_failover(self, metadata_quorum): + """After a cross-tier ``DeleteRecords`` on a switched consolidating topic, moving + leadership across the cluster must NOT over-reclaim the surviving classic prefix. + + When leadership moves, the new leader is also the partition's RLM leader: it runs the + become-leader report and the next reclaim cycle. If either used the broker-local + whole-log start as the remote-retention reclaim floor instead of the control-plane + cross-tier earliest (with the fail-safe remote-earliest floor when that is momentarily + unavailable), it could irreversibly delete ``[delete_before, S)`` from remote. This is + the leadership-change / metadata-propagation-timing path behind the RLM + ``isConsolidatingDisklessPartition`` fail-safe. + + This uses a clean rolling restart, which reloads each broker's on-disk log rather than + rebuilding it from remote, so it targets the leadership-change path specifically; the + from-remote rebuild that re-seeds the seal into the whole-log start is covered by + ``test_delete_records_survives_leader_rebuild_from_remote``. It performs the same + cross-tier delete as ``test_delete_records_across_tiers``, rolls every broker so + leadership visits each replica, then re-asserts the earliest still settles at *exactly* + the delete boundary and the surviving prefix ``[delete_before, S)`` is still readable + and contiguous.""" + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + baseline_tiered = verifier.tiered_object_count() + + seal_offset, total_acked = self._switch_and_drain(verifier, baseline_tiered) + + delete_before = seal_offset // 2 + assert delete_before > 0, "seal too small to pick a delete boundary: %d" % seal_offset + + low_watermark = verifier.delete_records(self.TOPIC, before_offset=delete_before) + self.logger.info("Requested DeleteRecords before offset %d (seal=%d); low_watermark=%d" + % (delete_before, seal_offset, low_watermark)) + assert low_watermark == delete_before, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (low_watermark, delete_before)) + + # Baseline (pre-failover) invariants, identical to the base test. + pre_failover_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert pre_failover_earliest == delete_before, ( + "pre-failover earliest %d did not settle at the requested boundary %d (seal=%d)" + % (pre_failover_earliest, delete_before, seal_offset)) + pre_failover_tiered = verifier.wait_for_tiered_count_stable() + self.logger.info("Pre-failover earliest agreed at %d; settled tiered count=%d" + % (pre_failover_earliest, pre_failover_tiered)) + + # Force fresh leadership across the whole cluster. + self._rolling_restart(verifier) + + # Post-failover: a broker that over-reclaimed using its local seal would push the + # earliest to S and make [delete_before, S) unreadable. Assert it stayed put. + post_failover_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert post_failover_earliest == delete_before, ( + "after leader failover the earliest settled at %d, not the delete boundary %d " + "(seal=%d); a new leader over-reclaimed [delete_before, seal) using its broker-local " + "whole-log start as the reclaim floor instead of the cross-tier earliest / fail-safe" + % (post_failover_earliest, delete_before, seal_offset)) + self.logger.info("Post-failover earliest still agreed at exactly %d" % post_failover_earliest) + + # The surviving prefix is still served from its exact boundary... + first_served = verifier.first_served_offset(self.TOPIC, from_offset=delete_before) + assert first_served == delete_before, ( + "after failover a fetch from the delete boundary %d returned offset %d (seal=%d); the " + "surviving classic prefix was reclaimed" % (delete_before, first_served, seal_offset)) + + # ...and reads back contiguous with the correct content. + spot = min(seal_offset - delete_before, self.SPOT_CHECK) + records = verifier.read_records_with_values_from( + self.TOPIC, from_offset=delete_before, max_messages=spot, timeout_ms=240000) + assert len(records) >= spot, ( + "after failover a bounded read from the surviving boundary %d returned only %d of %d " + "records; the surviving classic prefix was reclaimed" % (delete_before, len(records), spot)) + for i, (offset, value) in enumerate(records[:spot]): + expected_offset = delete_before + i + assert offset == expected_offset, ( + "non-contiguous read at position %d after failover: offset=%d, expected %d" + % (i, offset, expected_offset)) + assert value == offset, ( + "content mismatch at offset %d after failover: value=%d, expected %d" + % (offset, value, offset)) + self.logger.info("Surviving classic prefix from %d intact and readable after failover (%d records)" + % (delete_before, spot)) + + def _rebuild_all_from_remote(self, verifier): + """Force every replica to rebuild the partition from the remote tier. Stop all + brokers cleanly, delete only the partition's local dirs on each (keeping the + broker's KRaft identity and internal topics), then restart. With no local copy + left, whichever broker is elected leader must reconstruct the surviving prefix from + remote through the tier-state machine (``buildRemoteLogAuxState``), which is the + path that seeds the seal into the broker-local whole-log start on a switched topic. + A clean rolling restart cannot reach this: it reloads the on-disk log instead.""" + self.logger.info("Stopping all brokers to wipe local copies of %s" % self.TOPIC) + for node in self.kafka.nodes: + self.kafka.stop_node(node, clean_shutdown=True, timeout_sec=120) + for node in self.kafka.nodes: + verifier.wipe_topic_local_logs(node, self.TOPIC) + self.logger.info("Restarting all brokers with empty local logs for %s" % self.TOPIC) + for node in self.kafka.nodes: + self.kafka.start_node(node, timeout_sec=120) + wait_until(lambda: verifier.partition_has_leader(self.TOPIC), + timeout_sec=120, backoff_sec=2, + err_msg="partition %s-0 had no leader after the wipe+restart" % self.TOPIC) + wait_until(lambda: len(self.kafka.isr_idx_list(self.TOPIC, 0)) >= self.REPLICATION_FACTOR, + timeout_sec=240, backoff_sec=3, + err_msg="ISR for %s-0 did not recover to %d after the wipe+restart" + % (self.TOPIC, self.REPLICATION_FACTOR)) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_survives_leader_rebuild_from_remote(self, metadata_quorum): + """After a cross-tier ``DeleteRecords``, a leader that rebuilds its log from the + remote tier must NOT over-reclaim the surviving classic prefix. + + This is the strong form of the over-reclaim guard. Unlike a clean restart (which + reloads the on-disk log), wiping every replica's local partition dirs forces the + elected leader to reconstruct ``[delete_before, S)`` from remote via the tier-state + machine. That rebuild seeds the classic-to-diskless seal ``S`` into the broker-local + whole-log start (``UnifiedLog.logStartOffset``), which is exactly the value the RLM + reclaim floor and the become-leader report must NOT adopt. If they read that seal + instead of the control-plane cross-tier earliest (persisted at ``delete_before`` by + the delete), the rebuilt leader would report ``S`` and irreversibly delete the + surviving remote prefix ``[delete_before, S)``. The control-plane value is present + throughout, so the guard here is that a rebuilt leader prefers it over its own + seal-valued local start; the override-absent fail-safe (control plane momentarily + unavailable, RLM falling back to ``findLogStartOffset``) is unit-covered. + + Delete inside the sealed classic prefix, wait for the earliest to settle at the + boundary, wipe and rebuild the whole cluster from remote, then re-assert the earliest + is still *exactly* ``delete_before`` on every broker and the survivors are readable + and contiguous. A short idle after the rebuild lets the rebuilt leader run RLM + reclaim cycles, so a seal-based over-reclaim would surface as the earliest creeping up + toward ``S`` or the survivors going unreadable.""" + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + baseline_tiered = verifier.tiered_object_count() + + seal_offset, total_acked = self._switch_and_drain(verifier, baseline_tiered) + + delete_before = seal_offset // 2 + assert delete_before > 0, "seal too small to pick a delete boundary: %d" % seal_offset + + low_watermark = verifier.delete_records(self.TOPIC, before_offset=delete_before) + self.logger.info("Requested DeleteRecords before offset %d (seal=%d); low_watermark=%d" + % (delete_before, seal_offset, low_watermark)) + assert low_watermark == delete_before, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (low_watermark, delete_before)) + + pre_rebuild_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert pre_rebuild_earliest == delete_before, ( + "pre-rebuild earliest %d did not settle at the requested boundary %d (seal=%d)" + % (pre_rebuild_earliest, delete_before, seal_offset)) + # Settle the remote reclaim of [0, delete_before) so the surviving remote prefix is + # exactly [delete_before, seal) before the wipe forces a rebuild from it. + verifier.wait_for_tiered_count_stable() + self.logger.info("Pre-rebuild earliest agreed at %d; reclaim settled" % pre_rebuild_earliest) + + # Force the rebuild-from-remote path across the whole cluster. + self._rebuild_all_from_remote(verifier) + + # A rebuilt leader that adopted its seal-valued whole-log start would push the + # earliest to S and drop [delete_before, S). Assert it stayed at the boundary. + post_rebuild_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert post_rebuild_earliest == delete_before, ( + "after a rebuild-from-remote the earliest settled at %d, not the delete boundary %d " + "(seal=%d); the rebuilt leader adopted its seal-valued whole-log start as the " + "reclaim floor / become-leader report instead of the cross-tier earliest" + % (post_rebuild_earliest, delete_before, seal_offset)) + self.logger.info("Post-rebuild earliest still agreed at exactly %d" % post_rebuild_earliest) + + # Let the rebuilt leader run a few RLM reclaim cycles (remote.log.manager.task.interval.ms + # is 5s); a seal-based over-reclaim would show up as creep past the boundary or + # unreadable survivors. + self.logger.info("Idling to let the rebuilt leader run RLM reclaim cycles") + time.sleep(15) + steady_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert steady_earliest == delete_before, ( + "after the rebuild the earliest crept to %d (delete boundary %d, seal=%d); an RLM " + "reclaim cycle on the rebuilt leader over-reclaimed the surviving prefix" + % (steady_earliest, delete_before, seal_offset)) + + # The rebuilt survivors are served from their exact boundary, contiguous, correct. + first_served = verifier.first_served_offset(self.TOPIC, from_offset=delete_before) + assert first_served == delete_before, ( + "after the rebuild a fetch from the delete boundary %d returned offset %d (seal=%d); " + "the surviving classic prefix was reclaimed" % (delete_before, first_served, seal_offset)) + spot = min(seal_offset - delete_before, self.SPOT_CHECK) + records = verifier.read_records_with_values_from( + self.TOPIC, from_offset=delete_before, max_messages=spot, timeout_ms=240000) + assert len(records) >= spot, ( + "after the rebuild a bounded read from the surviving boundary %d returned only %d of " + "%d records; the surviving classic prefix was reclaimed" + % (delete_before, len(records), spot)) + for i, (offset, value) in enumerate(records[:spot]): + expected_offset = delete_before + i + assert offset == expected_offset, ( + "non-contiguous read at position %d after rebuild: offset=%d, expected %d" + % (i, offset, expected_offset)) + assert value == offset, ( + "content mismatch at offset %d after rebuild: value=%d, expected %d" + % (offset, value, offset)) + self.logger.info("Surviving classic prefix from %d intact and readable after rebuild (%d records)" + % (delete_before, spot)) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_across_tiers_born_consolidated(self, metadata_quorum): + """The cross-tier ``DeleteRecords`` guarantees must also hold for a + *born-consolidated* topic (diskless + remote enabled at creation, no classic + prefix, no seal), not just a switched one. + + Born-consolidated has no seal-pinned classic prefix, so it was never subject to + over-reclaim; the point here is to prove a single broker-agnostic earliest served + from the control plane and a correct cross-tier reclaim hold for this topology, + which historically had a separate ``DeleteRecords`` limitation. Produce a diskless + stream, let consolidation tier a remote prefix, + delete inside that prefix, and assert the earliest advances to *exactly* the + boundary on every broker and the survivor is readable.""" + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + + # Born consolidated: diskless + remote at creation, no switch. + verifier.create_consolidated_topic(self.TOPIC, self.NUM_PARTITIONS, self.REPLICATION_FACTOR) + acked = verifier.produce(self.TOPIC, self.NUM_DISKLESS_RECORDS, "born-consolidated") + self.logger.info("Produced born-consolidated stream: acked=%d" % acked) + + # Wait until consolidation has tiered a remote prefix, so the delete boundary we + # pick is remote-backed (a genuine cross-tier delete, not a WAL-only one). + tier_target = acked // 2 + remote_watermark = verifier.wait_for_tiered_offset_at_least(self.TOPIC, tier_target) + self.logger.info("Remote tier covers through %d (target %d)" % (remote_watermark, tier_target)) + + delete_before = tier_target // 2 # strictly inside the tiered/remote prefix + assert delete_before > 0, "not enough tiered data to pick a delete boundary (acked=%d)" % acked + + # Pre-delete: earliest is one broker-agnostic value below the boundary. + pre_earliest = verifier.earliest_on_each_broker(self.TOPIC) + pre_values = set(pre_earliest.values()) + assert len(pre_values) == 1, ( + "pre-delete earliest diverged across brokers: %s (ListOffsets(EARLIEST) not served " + "from the control plane for a born-consolidated topic)" % pre_earliest) + pre_value = next(iter(pre_values)) + assert 0 <= pre_value < delete_before, ( + "pre-delete earliest %d is not below the delete boundary %d; cannot exercise a " + "cross-tier delete" % (pre_value, delete_before)) + + low_watermark = verifier.delete_records(self.TOPIC, before_offset=delete_before) + self.logger.info("Requested DeleteRecords before offset %d (acked=%d); low_watermark=%d" + % (delete_before, acked, low_watermark)) + assert low_watermark == delete_before, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (low_watermark, delete_before)) + + agreed_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert agreed_earliest == delete_before, ( + "post-delete earliest %d did not settle at exactly the requested boundary %d for the " + "born-consolidated topic" % (agreed_earliest, delete_before)) + self.logger.info("Post-delete earliest agreed across all brokers at exactly %d" % agreed_earliest) + + first_served = verifier.first_served_offset(self.TOPIC, from_offset=delete_before) + assert first_served == delete_before, ( + "a fetch from the delete boundary %d returned offset %d; the surviving prefix was " + "reclaimed instead of served" % (delete_before, first_served)) + + spot = min(acked - delete_before, self.SPOT_CHECK) + records = verifier.read_records_with_values_from( + self.TOPIC, from_offset=delete_before, max_messages=spot, timeout_ms=240000) + assert len(records) >= spot, ( + "bounded read from the surviving boundary %d returned only %d of %d records; the " + "surviving prefix was reclaimed" % (delete_before, len(records), spot)) + for i, (offset, value) in enumerate(records[:spot]): + expected_offset = delete_before + i + assert offset == expected_offset, ( + "non-contiguous read at position %d: offset=%d, expected %d" % (i, offset, expected_offset)) + assert value == offset, ( + "content mismatch at offset %d: value=%d, expected %d" % (offset, value, offset)) + self.logger.info("Born-consolidated surviving prefix from %d read back contiguous (%d records)" + % (delete_before, spot)) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_follower_read_guard(self, metadata_quorum): + """A consumer that the metadata transformer routes to a *follower* must not be served + records below the cross-tier earliest after a ``DeleteRecords``, whether it is an old + (pre-KIP-392) or a modern client. + + With managed replicas the transformer advertises a hash-selected replica -- usually a + follower, not the real KRaft leader -- as the partition leader, and the follower then + serves the read from its local classic log (an old consumer via the + ``isOlderConsumer && switched`` relaxation, a modern consumer because ``allowReplica`` + is true for it once routed to a non-leader). A follower's broker-local whole-log start + (``UnifiedLog.logStartOffset``) does not move on a ``DeleteRecords`` (that advances only + the real leader's log start and the control plane), so a consumer resuming from a + position inside the now-deleted range + ``[0, delete_before)`` could read those logically-deleted records straight off the + follower. The fix consults the control-plane cross-tier earliest on the follower read + path and returns ``OFFSET_OUT_OF_RANGE`` for anything below it, so the consumer resets + (``auto.offset.reset=earliest``) to the delete boundary. + + Both client generations are exercised: the 2.2 console consumer the ducker image ships + (Fetch < v11, empty ``clientMetadata``) and the current one (Fetch v11+). Each asserts + a seek below the boundary never yields an offset ``< delete_before`` (guard rejects it; + the consumer resets to ``delete_before``), and the old client additionally checks a read + at the boundary is still served (the guard does not over-reject valid data).""" + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + baseline_tiered = verifier.tiered_object_count() + + seal_offset, total_acked = self._switch_and_drain(verifier, baseline_tiered) + + delete_before = seal_offset // 2 + assert delete_before > 0, "seal too small to pick a delete boundary: %d" % seal_offset + + low_watermark = verifier.delete_records(self.TOPIC, before_offset=delete_before) + self.logger.info("Requested DeleteRecords before offset %d (seal=%d); low_watermark=%d" + % (delete_before, seal_offset, low_watermark)) + assert low_watermark == delete_before, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (low_watermark, delete_before)) + + agreed_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert agreed_earliest == delete_before, ( + "post-delete earliest %d did not settle at exactly the requested boundary %d (seal=%d)" + % (agreed_earliest, delete_before, seal_offset)) + self.logger.info("Earliest settled at %d across all brokers" % agreed_earliest) + + spot = min(seal_offset - delete_before, self.SPOT_CHECK) + + # 1) Old consumer resuming from *inside* the deleted range must not be served any + # record below the cross-tier earliest. With the guard it gets OFFSET_OUT_OF_RANGE + # and resets (auto.offset.reset=earliest) to the boundary; without it the follower + # would hand back the logically-deleted prefix from its frozen local log. + # + # The 2.2 client cannot print offsets (no print.offset before KIP-431), so we read the + # record values instead; in the classic prefix [0, seal) the producer wrote value == offset, + # so a returned value is the record's offset. + probe = delete_before // 2 + below = verifier.read_values_with_old_client( + self.TOPIC, from_offset=probe, max_messages=spot, timeout_ms=240000) + stale = [v for v in below if v < delete_before] + assert not stale, ( + "a pre-KIP-392 consumer seeking to %d (inside the deleted range [0, %d)) was served " + "%d record(s) below the cross-tier earliest, e.g. %s; the follower served " + "logically-deleted data instead of rejecting the read" + % (probe, delete_before, len(stale), stale[:5])) + if below: + assert below[0] == delete_before, ( + "old consumer reset to offset %d, not the delete boundary %d" + % (below[0], delete_before)) + for i, value in enumerate(below): + expected = delete_before + i + assert value == expected, ( + "non-contiguous/mismatched old-client read at position %d: value=%d, " + "expected %d (value == offset in the classic prefix)" % (i, value, expected)) + self.logger.info("Old consumer reset to the boundary %d and read %d contiguous records" + % (delete_before, len(below))) + else: + self.logger.info("Old consumer returned nothing for a below-earliest seek; the key " + "invariant (no record below %d) still holds" % delete_before) + + # 1b) A modern consumer (Fetch v11+, sends clientMetadata) is routed to the same + # follower and served from its local log, so the broadened guard must reject its + # below-earliest reads too. Reset to earliest so an out-of-range seek lands on the + # boundary rather than the log end. + modern_below = verifier.read_records_with_values_from( + self.TOPIC, from_offset=probe, max_messages=spot, timeout_ms=240000, + consumer_properties={"auto.offset.reset": "earliest"}) + modern_stale = [(o, v) for (o, v) in modern_below if o < delete_before] + assert not modern_stale, ( + "a modern consumer seeking to %d (inside the deleted range [0, %d)) was served %d " + "record(s) below the cross-tier earliest, e.g. %s; the follower served " + "logically-deleted data to a modern client" % (probe, delete_before, len(modern_stale), modern_stale[:5])) + if modern_below: + assert modern_below[0][0] == delete_before, ( + "modern consumer reset to offset %d, not the delete boundary %d" + % (modern_below[0][0], delete_before)) + self.logger.info("Modern consumer reset to the boundary %d and read %d records" + % (delete_before, len(modern_below))) + else: + self.logger.info("Modern consumer returned nothing for a below-earliest seek; the key " + "invariant (no record below %d) still holds" % delete_before) + + # 2) The guard must not over-reject: an old consumer reading from exactly the + # surviving boundary is served that offset with the originally produced content + # (value == offset in the classic prefix). + at = verifier.read_values_with_old_client( + self.TOPIC, from_offset=delete_before, max_messages=spot, timeout_ms=240000) + assert at and at[0] == delete_before, ( + "a pre-KIP-392 consumer reading from the surviving boundary %d got %s; the guard " + "over-rejected a valid at-earliest read" % (delete_before, at[:1])) + for i, value in enumerate(at): + expected = delete_before + i + assert value == expected, ( + "non-contiguous/mismatched old-client read at position %d: value=%d, " + "expected %d (value == offset in the classic prefix)" % (i, value, expected)) + self.logger.info("Old consumer read the surviving prefix from exactly %d (%d records)" + % (delete_before, len(at))) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_across_tiers(self, metadata_quorum): + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + baseline_tiered = verifier.tiered_object_count() + + seal_offset, total_acked = self._switch_and_drain(verifier, baseline_tiered) + + # Precondition: every broker agrees the earliest is still below the delete + # boundary (the classic remote prefix is intact). We assert *agreement*, not a + # specific value: the control-plane cross-tier earliest is the same on every + # broker, and it must be well below the boundary so the DeleteRecords is a + # genuine cross-tier delete of remote data. + delete_before = seal_offset // 2 + assert delete_before > 0, "seal too small to pick a delete boundary: %d" % seal_offset + + pre_earliest = verifier.earliest_on_each_broker(self.TOPIC) + pre_values = set(pre_earliest.values()) + assert len(pre_values) == 1, ( + "pre-delete earliest diverged across brokers: %s (ListOffsets(EARLIEST) is not " + "served from the broker-agnostic control plane)" % pre_earliest) + pre_value = next(iter(pre_values)) + assert 0 <= pre_value < delete_before, ( + "pre-delete earliest %d is not below the delete boundary %d; cannot exercise a " + "cross-tier delete of the remote prefix" % (pre_value, delete_before)) + self.logger.info("Pre-delete earliest agreed across brokers at %d (boundary %d)" + % (pre_value, delete_before)) + + # Settle the pre-delete peak: the diskless tail keeps tiering after it drains + # locally, so a snapshot here can still be climbing. Capturing a settled peak + # keeps later diagnostics meaningful. + tiered_peak = verifier.wait_for_tiered_count_stable() + self.logger.info("Settled pre-delete tiered-storage object count: %d" % tiered_peak) + + # Delete the first half of the classic prefix. The boundary is strictly inside + # the sealed classic region [0, seal), which is tiered to remote, so the + # reclaimed range [0, delete_before) is remote-backed: this exercises the + # local-log -> RemoteLogManager cross-tier delete on the real leader (reached + # via DisklessDeleteRecordsForwarder). + # + # delete_records parses the per-partition result and raises on a partition error + # (kafka-delete-records.sh reports those on stdout yet still exits 0). The returned + # low_watermark is the partition's new log start as the broker computed it. + low_watermark = verifier.delete_records(self.TOPIC, before_offset=delete_before) + self.logger.info("Requested DeleteRecords before offset %d (seal=%d, total=%d); low_watermark=%d" + % (delete_before, seal_offset, total_acked, low_watermark)) + assert low_watermark == delete_before, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d " + "(the broker did not advance the log start to the delete offset)" + % (low_watermark, delete_before)) + + # Combined assertion: the client-visible earliest advances from below the boundary + # to a single, stable value that is the SAME on every broker (served from the + # control plane, not a hash-selected follower's frozen local log) and settles at + # EXACTLY the delete boundary (only [0, delete_before) is reclaimed -- with retention + # unset nothing else may move the earliest, and the consolidation cleanup must not + # over-reclaim the classic prefix down to the seal). Before the fixes this stayed at + # 0 (per-broker divergence) or ran ahead to the seal (over-reclaim). + agreed_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + self.logger.info("Post-delete earliest agreed across all brokers at a stable %d" + % agreed_earliest) + assert agreed_earliest == delete_before, ( + "post-delete earliest %d did not settle at exactly the requested boundary %d " + "(seal=%d); the cleanup over-reclaimed the classic prefix past the delete boundary " + "even though retention is unset" % (agreed_earliest, delete_before, seal_offset)) + + # The advertised floor is backed by real, readable data: a fetch from the boundary + # serves that exact offset (not a skip-ahead to the seal). + first_served = verifier.first_served_offset(self.TOPIC, from_offset=delete_before) + assert first_served == delete_before, ( + "a fetch from the delete boundary %d returned offset %d (seal=%d); the surviving " + "classic prefix was reclaimed instead of served" + % (delete_before, first_served, seal_offset)) + self.logger.info("Fetch from the delete boundary %d served that offset" % delete_before) + + # The surviving prefix [delete_before, seal) comes back contiguous with the correct + # content. VerifiableProducer writes each record's per-producer sequence as its value; + # the classic producer wrote [0, seal) so value == offset below the seal. + spot = min(seal_offset - delete_before, self.SPOT_CHECK) + records = verifier.read_records_with_values_from( + self.TOPIC, from_offset=delete_before, max_messages=spot, timeout_ms=240000) + assert len(records) >= spot, ( + "bounded read from the surviving boundary %d returned only %d of %d records; the " + "surviving classic prefix was reclaimed" % (delete_before, len(records), spot)) + for i, (offset, value) in enumerate(records[:spot]): + expected_offset = delete_before + i + assert offset == expected_offset, ( + "non-contiguous read at position %d: offset=%d, expected %d (gap/dupe/reorder)" + % (i, offset, expected_offset)) + assert value == offset, ( + "content mismatch at offset %d: value=%d, expected %d; the surviving classic " + "record does not match what was produced" % (offset, value, offset)) + self.logger.info("Surviving classic prefix from %d read back contiguous with correct content (%d records)" + % (delete_before, spot)) diff --git a/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py b/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py index 80bdd6a4deb..ed6a4108ede 100644 --- a/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py +++ b/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py @@ -291,3 +291,117 @@ def test_retention_reclaims_across_tiers(self, metadata_quorum): # 4) End-to-end: the full surviving tail [earliest, end) comes back. self._read_back_tail(verifier, from_offset=earliest, expected=expected, phase1_end=acked1) + + @cluster(num_nodes=6) + @matrix(metadata_quorum=[quorum.isolated_kraft]) + def test_delete_records_composed_with_active_retention(self, metadata_quorum): + """``DeleteRecords`` and active time-based retention must compose across tiers: + the earliest settles at ``max(delete_boundary, retention_floor)`` and neither + over-reclaims the survivors. + + Reuses this test's two age-separated cohorts so the retention floor is + deterministic (the aged phase-1 cohort expires to the boundary ``acked1``; the + fresh phase-2 cohort survives), then folds a ``DeleteRecords`` into the same run + twice: + + - Below the retention floor (``delete < acked1``): retention dominates. The + smaller delete must not pin the earliest low; the forward-only floor advances to + the retention boundary once the aged cohort expires. + - Above the retention floor, inside the fresh survivors (``delete > acked1``): the + delete dominates. Retention (restored long, so phase 2 never ages) must not + block it, and the survivors above the delete stay intact and contiguous. + + A born-consolidated topic is used deliberately. The switched-topic seal + over-reclaim is already covered by ``DeleteRecordsAcrossTiersTest`` (including the + leader-failover variant), and a switched topic cannot place a *time-based* + retention floor strictly inside the classic prefix deterministically: the whole + pre-switch prefix ages as one cohort, so the floor can only land at 0 or at the + seal. What is new here is the retention x ``DeleteRecords`` composition end to end; + it is unit-covered + (``testConsolidatingOverrideFloorHonoredWithActive{Size,Time}Retention``) but + never exercised across real tiers.""" + self._start_cluster() + verifier = ConsolidationVerifier(self.kafka) + verifier.verify_tooling() + baseline_tiered = verifier.tiered_object_count() + + verifier.start_jmx() + + acked1 = verifier.produce(self.TOPIC, self.PHASE1_RECORDS, label="phase1", + throughput=self.PHASE1_THROUGHPUT, + timeout_sec=self.PHASE1_SPAN_SEC + 120) + self.logger.info("Phase 1 (aged cohort) produced and acked %d records" % acked1) + + tiered_peak = self._drain_pipeline(verifier, baseline_tiered, acked1) + + self.logger.info("Idling %ds so the whole phase-1 cohort ages past retention.ms=%d" + % (self.AGE_AFTER_TIER_SEC, self.RETENTION_MS)) + time.sleep(self.AGE_AFTER_TIER_SEC) + + acked2 = verifier.produce(self.TOPIC, self.PHASE2_RECORDS, label="phase2") + total_acked = acked1 + acked2 + self.logger.info("Phase 2 (fresh survivors) produced and acked %d records (total=%d)" + % (acked2, total_acked)) + + # --- Sub-case A: delete BELOW the retention boundary; retention dominates. --- + delete_below = acked1 // 2 + assert delete_below > 0, "phase-1 cohort too small to pick a delete boundary: %d" % acked1 + lw_below = verifier.delete_records(self.TOPIC, before_offset=delete_below) + assert lw_below == delete_below, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (lw_below, delete_below)) + self.logger.info("Deleted below the retention floor at %d (phase-1 boundary %d)" + % (delete_below, acked1)) + + # Lower retention so the aged phase-1 cohort expires. The earliest must advance + # past the (smaller) delete to the cohort boundary, not stay pinned at the delete. + verifier.kafka.alter_topic_config(self.TOPIC, "retention.ms", str(self.RETENTION_MS)) + wait_until(lambda: verifier.offset_at(self.TOPIC, time_spec=-2) > delete_below, + timeout_sec=300, backoff_sec=5, + err_msg=("retention did not advance the earliest above the delete boundary %d; " + "the aged phase-1 cohort was not reclaimed" % delete_below)) + earliest_after_retention = verifier.wait_for_earliest_stable(self.TOPIC, timeout_sec=300) + # Freeze the floor before the next phase so the survivors cannot age into the window. + verifier.kafka.alter_topic_config(self.TOPIC, "retention.ms", str(self.INITIAL_RETENTION_MS)) + assert delete_below < earliest_after_retention <= acked1, ( + "with a delete at %d below the retention boundary, earliest settled at %d; expected " + "retention to dominate and advance it into (%d, %d]" + % (delete_below, earliest_after_retention, delete_below, acked1)) + self.logger.info("Retention dominated the smaller delete: earliest advanced to %d (boundary %d)" + % (earliest_after_retention, acked1)) + + # The aged phase-1 remote segments were physically reclaimed (no storage leak). + wait_until(lambda: verifier.tiered_object_count() < tiered_peak, + timeout_sec=240, backoff_sec=5, + err_msg=("tiered-storage object count did not drop below the pre-reclaim peak " + "of %d; the aged phase-1 remote segments were not deleted" % tiered_peak)) + + # --- Sub-case B: delete ABOVE the retention boundary, inside the fresh survivors; + # the delete dominates and the survivors above it stay intact. --- + delete_above = earliest_after_retention + max((total_acked - earliest_after_retention) // 2, 1) + assert earliest_after_retention < delete_above < total_acked, ( + "could not place a delete boundary inside the survivor cohort (earliest=%d, total=%d)" + % (earliest_after_retention, total_acked)) + lw_above = verifier.delete_records(self.TOPIC, before_offset=delete_above) + assert lw_above == delete_above, ( + "DeleteRecords returned low_watermark=%d; expected the requested boundary %d" + % (lw_above, delete_above)) + + agreed_earliest = verifier.wait_for_consistent_earliest_across_brokers( + self.TOPIC, timeout_sec=300) + assert agreed_earliest == delete_above, ( + "post-delete earliest %d did not settle at exactly the delete boundary %d (retention " + "floor was %d); the delete must dominate above the retention floor" + % (agreed_earliest, delete_above, earliest_after_retention)) + self.logger.info("Delete dominated above the retention floor: earliest at exactly %d" + % agreed_earliest) + + # Survivors [delete_above, total) are readable and contiguous with correct content. + first_served = verifier.first_served_offset(self.TOPIC, from_offset=delete_above) + assert first_served == delete_above, ( + "a fetch from the delete boundary %d returned offset %d; the survivors were reclaimed" + % (delete_above, first_served)) + expected = total_acked - delete_above + self._read_back_tail(verifier, from_offset=delete_above, expected=expected, phase1_end=acked1) + self.logger.info("Survivor tail [%d, %d) intact after the retention x delete composition" + % (delete_above, total_acked))