static Set<TopicPartitionReplica> alterReplicaLogDirs(Admin adminClient,
                                                          Map<TopicPartitionReplica, String> assignment) throws InterruptedException {
        Set<TopicPartitionReplica> results = new HashSet<>();
        Map<TopicPartitionReplica, KafkaFuture<Void>> values = adminClient.alterReplicaLogDirs(assignment).values();

        for (Entry<TopicPartitionReplica, KafkaFuture<Void>> e : values.entrySet()) {
            TopicPartitionReplica replica = e.getKey();
            KafkaFuture<Void> future = e.getValue();
            try {
                future.get();
                results.add(replica);
            } catch (ExecutionException t) {
                // Ignore ReplicaNotAvailableException.  It is OK if the replica is not
                // available at this moment.
                if (t.getCause() instanceof ReplicaNotAvailableException)
                    continue;
                throw new AdminCommandFailedException("Failed to alter dir for " + replica, t);
            }
        }
        return results;
    }
The entry point for the --cancel command.

@param adminClient           The AdminClient to use.
@param jsonString            The JSON string to use for the topics and partitions to cancel.
@param preserveThrottles     True if we should avoid changing topic or broker throttles.
@param timeoutMs             The maximum time in ms to wait for log directory
                             replica assignment to begin.
@param time                  The Time object to use.

@return                      A tuple of the partition reassignments that were cancelled,
                             and the replica movements that were cancelled.
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java