Delegate wait_for_workers to cluster instances only when implemented

distributed/client.py

@@ -1454,10 +1454,11 @@ def wait_for_workers(self, n_workers: int, timeout: float | None = None) -> None
                 f"`n_workers` must be a positive integer. Instead got {n_workers}."
             )
 
-        if self.cluster is None:
-            return self.sync(self._wait_for_workers, n_workers, timeout=timeout)
+        if self.cluster and hasattr(self.cluster, "wait_for_workers"):
+            return self.cluster.wait_for_workers(n_workers, timeout)
+
+        return self.sync(self._wait_for_workers, n_workers, timeout=timeout)
 
-        return self.cluster.wait_for_workers(n_workers, timeout)
 
     def _heartbeat(self):
         # Don't send heartbeat if scheduler comm or cluster are already closed