private static void executeMoves(Admin adminClient,
                                     Map<TopicPartitionReplica, String> proposedReplicas,
                                     Long timeoutMs,
                                     Time time
    ) throws InterruptedException, TerseException {
        long startTimeMs = time.milliseconds();
        Map<TopicPartitionReplica, String> pendingReplicas = new HashMap<>(proposedReplicas);
        boolean done = false;
        do {
            Set<TopicPartitionReplica> completed = alterReplicaLogDirs(adminClient, pendingReplicas);
            if (!completed.isEmpty()) {
                completed.stream().sorted(ReassignPartitionsCommand::compareTopicPartitionReplicas).forEach(replica -> {
                    System.out.printf("Successfully started moving log directory to %s for replica %s-%s with broker %s %n",
                        pendingReplicas.get(replica), replica.topic(), replica.partition(), replica.brokerId());
                });
            }
            completed.forEach(pendingReplicas::remove);
            if (pendingReplicas.isEmpty()) {
                done = true;
            } else if (time.milliseconds() >= startTimeMs + timeoutMs) {
                throw new TerseException(String.format(
                    "Timed out before log directory move%s could be started for: %s",
                        pendingReplicas.size() == 1 ? "" : "s",
                        pendingReplicas.keySet().stream()
                            .sorted(ReassignPartitionsCommand::compareTopicPartitionReplicas)
                            .map(Object::toString)
                            .collect(Collectors.joining(","))));
            } else {
                // If a replica has been moved to a new host and we also specified a particular
                // log directory, we will have to keep retrying the alterReplicaLogDirs
                // call.  It can't take effect until the replica is moved to that host.
                time.sleep(100);
            }
        } while (!done);
    }
Execute some partition log directory movements.

@param adminClient                 The AdminClient to use.
@param proposedReplicas            A map from TopicPartitionReplicas to the
                                   directories to move them to.
@param timeoutMs                   The maximum time in ms to wait for log directory
                                   replica assignment to begin.
@param time                        The Time object to use.
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java