From 0811c845d71d30b57b955c534a322673ffeb105b Mon Sep 17 00:00:00 2001 From: Andrew Wong Date: Thu, 2 Jul 2026 09:30:54 -0700 Subject: [PATCH] [CORE-16405] cl/test: uncap target consumer reads for compacted syncs The target consumer carried a fixed read-count cap (max_msgs=msg_count) while the source consumer did not. Under compaction, workload completion is gated on per-partition offset parity (max_offsets_match), which a bounded read cannot reach: rebalance re-reads spend the budget, and a partition whose leadership churns has its tail produced and replicated late, so the consumer stops below max_offsets_produced and the workload never finishes reading to the HWM despite replication finishing. This commit drops the cap on the compacted path so the target consumer tails to parity like the source consumer; the non-compacted path keeps its count-based cap unchanged. To avoid this in the future, we also now reject max_msgs with use_compaction up front so the incompatibility fails fast instead of stalling for the progress timeout. This attempts to fix ShadowLinkingRandomOpsTest.test_node_operations, which is quite flaky. --- tests/rptest/tests/cluster_linking_test_base.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/rptest/tests/cluster_linking_test_base.py b/tests/rptest/tests/cluster_linking_test_base.py index 74ddcbc1e1491..d0246da868662 100644 --- a/tests/rptest/tests/cluster_linking_test_base.py +++ b/tests/rptest/tests/cluster_linking_test_base.py @@ -232,6 +232,13 @@ def __init__( self.consumer_properties: dict[str, Any] = ( consumer_properties if consumer_properties else {} ) + # When using compaction, the completion criteria examines per-partition + # offsets, which may be at odds with having a max_msgs set. + assert not (self.use_compaction and "max_msgs" in self.consumer_properties), ( + "max_msgs is incompatible with use_compaction: completion requires " + "per-partition offset parity, which a bounded read may never reach. " + "Let the consumer tail (continuous) instead." + ) self.timeout_sec = timeout_sec self.validate_number_of_messages_on_target = ( validate_number_of_messages_on_target @@ -275,12 +282,14 @@ def start(self): ) self.source_consumer.start(clean=False) + # NOTE: when using compaction, the completion criteria examines + # per-partition offsets, which is at odds with having a max_msgs. self.target_consumer = KgoVerifierConsumerGroupConsumer( context=self.test_context, redpanda=self.target_cluster.service, topic=self.topic, msg_size=self.msg_size, - max_msgs=self.msg_count, + max_msgs=None if self.use_compaction else self.msg_count, readers=readers, use_transactions=self.use_transactions, group_name=f"target-cg-{self._instance_id}",