rptest: stop() in nodes_decommissioning_test

Author: WillemKauf

tests/rptest/tests/nodes_decommissioning_test.py

@@ -308,17 +308,19 @@ def verify(self):
             f"verifying workload: topic: {self._topic}, with [rate_limit: {self.producer_throughput}, message size: {self.msg_size}, message count: {self.msg_count}]"
         )
         self.producer.wait()
+        self.producer.stop()
 
         # Await the consumer that is reading only the subset of data that
         # was written before it started.
         self.consumer.wait()
-
+        self.consumer.stop()
         assert self.consumer.consumer_status.validator.invalid_reads == 0, f"Invalid reads in topic: {self._topic}, invalid reads count: {self.consumer.consumer_status.validator.invalid_reads}"
         del self.consumer
 
         # Start a new consumer to read all data written
         self.start_consumer()
         self.consumer.wait()
+        self.consumer.stop()
 
         assert self.consumer.consumer_status.validator.invalid_reads == 0, f"Invalid reads in topic: {self._topic}, invalid reads count: {self.consumer.consumer_status.validator.invalid_reads}"
 
@@ -490,7 +492,6 @@ def calculate_total_learners_gap() -> int | None:
 
         self.logger.info(f"decommissioning node: {to_decommission_id}", )
         self._decommission(to_decommission_id)
-        self.producer.wait()
 
         def learner_gap_reported(decommissioned_node_id: int):
             total_gap = calculate_total_learners_gap()
@@ -1337,6 +1338,7 @@ def hmb(bs):
             return [convert(b) for b in bs]
 
         producer.wait(timeout_sec=300)
+        producer.stop()
         produce_duration = time.time() - produce_start_time
         self.logger.info(
             f"Produced {hmb([data_size])} in {produce_duration} seconds")