aiven · viktorsomogyi · Jul 30, 2026 · Jul 21, 2026
diff --git a/tests/kafkatest/services/inkless/consolidation_verifier.py b/tests/kafkatest/services/inkless/consolidation_verifier.py
@@ -217,15 +217,16 @@ def _ensure_mc_alias(self, node):
             "connectivity/credentials for the inkless system-test dependencies."
             % (self.MINIO_ALIAS, self.MINIO_ENDPOINT, rc))
 
-    def object_keys(self, prefix=""):
-        # Object keys under the given bucket prefix, via mc ls --recursive --json.
+    def _object_entries(self, prefix=""):
+        # (key, size) tuples under the given bucket prefix, via mc ls --recursive
+        # --json. mc emits one JSON object per line with a "size" field in bytes.
         node = self._storage_node()
         self._ensure_mc_alias(node)
         target = "%s/%s" % (self.MINIO_ALIAS, self.MINIO_BUCKET)
         if prefix:
             target += "/" + prefix
         cmd = "mc ls --recursive --json %s" % target
-        keys = []
+        entries = []
         for line in node.account.ssh_capture(cmd, allow_fail=False):
             line = line.strip()
             if not line:
@@ -244,8 +245,12 @@ def object_keys(self, prefix=""):
                 continue
             key = obj.get("key")
             if key:
-                keys.append(key)
-        return keys
+                entries.append((key, int(obj.get("size") or 0)))
+        return entries
+
+    def object_keys(self, prefix=""):
+        # Object keys under the given bucket prefix, via mc ls --recursive --json.
+        return [key for key, _ in self._object_entries(prefix)]
 
     def tiered_object_count(self):
         # Count under the tiered-storage prefix only (scanning the whole bucket
@@ -255,6 +260,17 @@ def tiered_object_count(self):
         # absolute `== N` flaps with test order.
         return len(self.object_keys(self.TIERED_PREFIX))
 
+    def tiered_object_bytes(self):
+        # Total bytes under the tiered-storage prefix. Like tiered_object_count,
+        # this is not topic-scoped and accumulates across tests in one
+        # `ducker-ak test` run, so use it as a baseline + delta, never absolute.
+        # A retention.bytes test uses it to pick a whole-log size limit relative
+        # to the data that actually reached remote, instead of guessing at the
+        # per-record on-disk size. Scoping the listing to the prefix already limits
+        # the entries to tiered storage (mc returns keys relative to the prefix, so
+        # no further key filtering is possible here anyway).
+        return sum(size for _, size in self._object_entries(self.TIERED_PREFIX))
+
     def wal_object_count(self):
         # Bucket-root count (outside tiered-storage/), i.e. diskless WAL files.
         # Same accumulation caveats as tiered_object_count: baseline + delta
@@ -733,7 +749,10 @@ def read_contiguous_from(self, topic, from_offset=0, max_messages=1, partition=0
         first_offset = -1
         num_read = 0
         try:
-            for line in node.account.ssh_capture(cmd, allow_fail=True):
+            # combine_stderr=False: ducktape merges stderr into stdout by default, so the
+            # consumer's log lines and its trailing "Processed a total of N messages" can
+            # land mid-line on a record and break the match, silently dropping that record.
+            for line in node.account.ssh_capture(cmd, allow_fail=True, combine_stderr=False):
                 line = line.decode("utf-8") if isinstance(line, bytes) else line
                 # We always pass print.offset=true, so each record line is exactly
                 # "Offset:<n>". Anchor on that (or a digits-only line for legacy
@@ -911,7 +930,10 @@ def read_records_with_values_from(self, topic, from_offset, max_messages, partit
                ))
         records = []
         try:
-            for line in node.account.ssh_capture(cmd, allow_fail=True):
+            # combine_stderr=False: ducktape merges stderr into stdout by default, so the
+            # consumer's log lines and its trailing "Processed a total of N messages" can
+            # land mid-line on a record and break the match, silently dropping that record.
+            for line in node.account.ssh_capture(cmd, allow_fail=True, combine_stderr=False):
                 line = line.decode("utf-8") if isinstance(line, bytes) else line
                 match = re.match(r"Offset:(\d+)\s+(-?\d+)\s*$", line.strip())
                 if match:
@@ -960,7 +982,10 @@ def read_values_with_old_client(self, topic, from_offset, max_messages, partitio
                ))
         values = []
         try:
-            for line in node.account.ssh_capture(cmd, allow_fail=True):
+            # combine_stderr=False: ducktape merges stderr into stdout by default, so the
+            # consumer's log lines and its trailing "Processed a total of N messages" can
+            # land mid-line on a record and break the match, silently dropping that record.
+            for line in node.account.ssh_capture(cmd, allow_fail=True, combine_stderr=False):
                 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.

diff --git a/tests/kafkatest/tests/inkless/consolidation_dependency_outage_test.py b/tests/kafkatest/tests/inkless/consolidation_dependency_outage_test.py
@@ -91,7 +91,12 @@ def test_dependency_outage_during_produce(self, metadata_quorum, dependency):
                         "diskless.enable": "true",
                         "remote.storage.enable": "true",
                         "min.insync.replicas": 2,
-                        "segment.bytes": 1048576,
+                        # segment.bytes must sit above max.message.bytes (default 1048588):
+                        # buildFetch reserves max.message.bytes of headroom for batch overshoot,
+                        # so a segment.bytes at or below it leaves a 1-byte fetch budget and
+                        # starves the consolidation fetcher. 2 MiB leaves ~1 MiB of fetch budget;
+                        # segment.ms still rolls small segments on time so tiering cadence holds.
+                        "segment.bytes": 2097152,
                         "segment.ms": 5000,
                         "local.retention.ms": 5000,
                     },

diff --git a/tests/kafkatest/tests/inkless/consolidation_pipeline_test.py b/tests/kafkatest/tests/inkless/consolidation_pipeline_test.py
@@ -75,8 +75,12 @@ def test_consolidation_pipeline(self, metadata_quorum):
                         "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.bytes must sit above max.message.bytes (default 1048588):
+                        # buildFetch reserves max.message.bytes of headroom for batch overshoot,
+                        # so a segment.bytes at or below it leaves a 1-byte fetch budget and
+                        # starves the consolidation fetcher. 2 MiB leaves ~1 MiB of fetch budget;
+                        # segment.ms still rolls small segments on time so tiering cadence holds.
+                        "segment.bytes": 2097152,
                         "segment.ms": 5000,
                     },
                 },

diff --git a/tests/kafkatest/tests/inkless/consolidation_read_from_remote_after_prune_test.py b/tests/kafkatest/tests/inkless/consolidation_read_from_remote_after_prune_test.py
@@ -75,8 +75,12 @@ def test_read_from_remote_after_prune(self, metadata_quorum):
                         "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.bytes must sit above max.message.bytes (default 1048588):
+                        # buildFetch reserves max.message.bytes of headroom for batch overshoot,
+                        # so a segment.bytes at or below it leaves a 1-byte fetch budget and
+                        # starves the consolidation fetcher. 2 MiB leaves ~1 MiB of fetch budget;
+                        # segment.ms still rolls small segments on time so tiering cadence holds.
+                        "segment.bytes": 2097152,
                         "segment.ms": 5000,
                         # Evict local segments soon after upload so data lives in
                         # remote before the wipe.

diff --git a/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py b/tests/kafkatest/tests/inkless/consolidation_retention_across_tiers_test.py
@@ -131,8 +131,12 @@ def _start_cluster(self):
                         "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.bytes must sit above max.message.bytes (default 1048588):
+                        # buildFetch reserves max.message.bytes of headroom for batch overshoot,
+                        # so a segment.bytes at or below it leaves a 1-byte fetch budget and
+                        # starves the consolidation fetcher. 2 MiB leaves ~1 MiB of fetch budget;
+                        # segment.ms still rolls small segments on time so tiering cadence holds.
+                        "segment.bytes": 2097152,
                         "segment.ms": 5000,
                         # Evict local segments soon after upload so the early prefix
                         # lives only in remote -- retention must then reclaim remote.