tests: made atomic consume-transform-produce test more robust

The exactly once delivery semantic tests was using a stop condition
where a size of the finished processing partition set was compared with
the size of high watermark list. Sometimes the high watermark list may
not contain all the partitions as they are not reported when there is no
leader. This lead to test stopping to early and desitination topic
missing some messages.  Changed the test to use the topic partition
count in the stop condition instead.

Fixes: #20315

Signed-off-by: Michał Maślanka <[email protected]>

tests/rptest/transactions/tx_atomic_produce_consume_test.py

@@ -35,6 +35,7 @@ class ExactlyOnceVerifier():
     def __init__(self,
                  redpanda,
                  src_topic: str,
+                 partition_count: int,
                  dst_topic: str,
                  transform_func,
                  logger,
@@ -44,6 +45,7 @@ def __init__(self,
                  commit_every: int = 50,
                  timeout_sec: int = 60):
         self._src_topic = src_topic
+        self._partition_count = partition_count
         self._dst_topic = dst_topic
         self._transform_func = transform_func
         self._logger = logger
@@ -185,7 +187,7 @@ def reached_end():
                 with self._lock:
                     self._finished_partitions |= end_for
                     consumers = self._consumer_cnt
-                    if len(self._finished_partitions) == len(high_watermarks):
+                    if len(self._finished_partitions) == self._partition_count:
                         return True
 
                 return len(end_for) == len(assignments) and consumers > 1
@@ -359,6 +361,7 @@ def simple_transform(k, v):
 
         transformer = ExactlyOnceVerifier(self.redpanda,
                                           topic.name,
+                                          topic.partition_count,
                                           dst_topic.name,
                                           simple_transform,
                                           self.logger,