ducky: pin to the latest version

tests/rptest/services/kafka_cli_consumer.py

@@ -51,6 +51,7 @@ def script(self):
 
     def _worker(self, _, node):
         self._done = False
+        self.logger.debug("%s: starting worker thread" % self.who_am_i(node))
         self._stopping.clear()
         try:
 
@@ -80,6 +81,8 @@ def _worker(self, _, node):
                     f"[{self._instance_name}] consumed: '{line}'")
                 self._messages.append(line)
         except:
+            self.logger.exception("%s: something is wrong" %
+                                  self.who_am_i(node))
             if self._stopping.is_set():
                 # Expect a non-zero exit code when killing during teardown
                 pass
@@ -95,15 +98,23 @@ def wait_for_messages(self, messages, timeout=30):
 
     def wait_for_started(self, timeout=10):
         def all_started():
-            return all([
-                len(node.account.java_pids("ConsoleConsumer")) == 1
-                for node in self.nodes
-            ])
+            for node in self.nodes:
+                pids = node.account.java_pids("ConsoleConsumer")
+                self.logger.debug(
+                    "%s: ConsoleConsumer pids: %s" %
+                    (self.who_am_i(node), ",".join(map(str, pids))))
+                if len(pids) != 1:
+                    return False
+            return True
 
         wait_until(all_started, timeout, backoff_sec=1)
 
     def stop_node(self, node):
         self._stopping.set()
+        self.logger.exception("%s: stop requested" % self.who_am_i(node))
+        pids = node.account.java_pids("ConsoleConsumer")
+        self.logger.debug("%s: ConsoleConsumer pids: %s" %
+                          (self.who_am_i(node), ",".join(map(str, pids))))
         node.account.kill_process("java", clean_shutdown=True)
 
         try:

tests/rptest/tests/partition_balancer_test.py

@@ -12,7 +12,8 @@
 
 from rptest.services.cluster import cluster
 from rptest.services.admin import Admin
-from rptest.util import wait_until, wait_until_result
+from ducktape.utils.util import wait_until
+from rptest.util import wait_until_result
 from rptest.clients.default import DefaultClient
 from rptest.services.redpanda import RedpandaService, CHAOS_LOG_ALLOW_LIST, MetricsEndpoint
 from rptest.services.failure_injector import FailureInjector, FailureSpec

tests/rptest/tests/partition_movement_upgrade_test.py

@@ -19,8 +19,8 @@
 from rptest.services.redpanda import RESTART_LOG_ALLOW_LIST
 from rptest.services.redpanda_installer import RedpandaInstaller, wait_for_num_versions
 
-from rptest.util import wait_until
 from ducktape.mark import ok_to_fail
+from ducktape.utils.util import wait_until
 
 
 class PartitionMovementUpgradeTest(PreallocNodesTest, PartitionMovementMixin):

tests/rptest/tests/raft_availability_test.py

@@ -16,7 +16,7 @@
 import requests
 
 from ducktape.mark import parametrize
-from rptest.util import wait_until
+from ducktape.utils.util import wait_until
 
 from rptest.clients.kafka_cat import KafkaCat
 from rptest.clients.rpk import RpkTool, RpkException

tests/rptest/util.py

@@ -14,8 +14,7 @@
 from rptest.clients.kafka_cli_tools import KafkaCliTools
 from rptest.services.storage import Segment
 
-from ducktape.errors import TimeoutError
-import time
+from ducktape.utils.util import wait_until
 
 
 class Scale:
@@ -50,31 +49,6 @@ def release(self):
         return self._scale == Scale.RELEASE
 
 
-def wait_until(condition,
-               timeout_sec,
-               backoff_sec=.1,
-               err_msg="",
-               retry_on_exc=False):
-    start = time.time()
-    stop = start + timeout_sec
-    last_exception = None
-    while time.time() < stop:
-        try:
-            if condition():
-                return
-            else:
-                last_exception = None
-        except BaseException as e:
-            last_exception = e
-            if not retry_on_exc:
-                raise e
-        time.sleep(backoff_sec)
-
-    # it is safe to call Exception from None - will be just treated as a normal exception
-    raise TimeoutError(
-        err_msg() if callable(err_msg) else err_msg) from last_exception
-
-
 def wait_until_result(condition, *args, **kwargs):
     """
     a near drop-in replacement for ducktape's wait_util except that when

tests/setup.py

@@ -13,7 +13,7 @@
     package_data={'': ['*.md']},
     include_package_data=True,
     install_requires=[
-        'ducktape@git+https://github.com/redpanda-data/ducktape.git@7ba83f32d5f265aad955a31096dcc0c97d96c0ce',
+        'ducktape@git+https://github.com/redpanda-data/ducktape.git@e1e19d4b1cb8f8476367f413b5a3e9be08492887',
         'prometheus-client==0.9.0', 'pyyaml==5.3.1', 'kafka-python==2.0.2',
         'crc32c==2.2', 'confluent-kafka==1.7.0', 'zstandard==0.15.2',
         'xxhash==2.0.2', 'protobuf==3.19.3', 'fastavro==1.4.9',