ducky: tune partition_movement to be faster

duration of partition_movement_test.py falls down from
14 minutes 15.077 seconds to 12 minutes 39.137 seconds

tests/rptest/tests/end_to_end.py

@@ -28,11 +28,12 @@
 from rptest.services.redpanda_installer import InstallOptions
 from rptest.services.redpanda_installer import RedpandaInstaller
 from rptest.clients.default import DefaultClient
+from rptest.clients.rpk import RpkTool
+from rptest.services.admin import Admin
 from rptest.services.verifiable_consumer import VerifiableConsumer
 from rptest.services.verifiable_producer import VerifiableProducer, is_int_with_prefix
 from rptest.archival.s3_client import S3Client
 from rptest.clients.rpk import RpkTool
-from rptest.clients.rpk import RpkException
 
 TopicPartition = namedtuple('TopicPartition', ['topic', 'partition'])
 
@@ -134,9 +135,19 @@ def start_redpanda(self,
                                              RedpandaInstaller.HEAD)
             self.redpanda.restart_nodes(nodes_to_upgrade)
 
+        self._admin_client = Admin(self.redpanda)
         self._client = DefaultClient(self.redpanda)
         self._rpk_client = RpkTool(self.redpanda)
 
+    def await_controller(self, check=lambda node_id: True, hosts=None):
+        admin = self.admin_client()
+        return admin.await_stable_leader("controller",
+                                         partition=0,
+                                         namespace="redpanda",
+                                         timeout_s=30,
+                                         check=check,
+                                         hosts=hosts)
+
     def rpk_client(self):
         assert self._rpk_client is not None
         return self._rpk_client

tests/rptest/tests/partition_movement.py

@@ -106,6 +106,14 @@ def keep(p):
     def _wait_post_move(self, topic, partition, assignments, timeout_sec):
         admin = Admin(self.redpanda)
 
+        def derived_done():
+            info = self._get_current_partitions(admin, topic, partition)
+            self.logger.info(
+                f"derived assignments for {topic}-{partition}: {info}")
+            return self._equal_assignments(info, assignments)
+
+        wait_until(derived_done, timeout_sec=timeout_sec, backoff_sec=1)
+
         def status_done():
             results = []
             for n in self.redpanda._started:
@@ -121,17 +129,7 @@ def status_done():
         # wait until redpanda reports complete
         wait_until(status_done, timeout_sec=timeout_sec, backoff_sec=2)
 
-        def derived_done():
-            info = self._get_current_partitions(admin, topic, partition)
-            self.logger.info(
-                f"derived assignments for {topic}-{partition}: {info}")
-            return self._equal_assignments(info, assignments)
-
-        wait_until(derived_done, timeout_sec=timeout_sec, backoff_sec=2)
-
     def _do_move_and_verify(self, topic, partition, timeout_sec):
-        admin = Admin(self.redpanda)
-
         _, new_assignment = self._dispatch_random_partition_move(
             topic=topic, partition=partition)