public static void executeAssignment(Admin adminClient,
                                  Boolean additional,
                                  String reassignmentJson,
                                  Long interBrokerThrottle,
                                  Long logDirThrottle,
                                  Long timeoutMs,
                                  Time time
    ) throws ExecutionException, InterruptedException, JsonProcessingException, TerseException {
        Entry<Map<TopicPartition, List<Integer>>, Map<TopicPartitionReplica, String>> t0 = parseExecuteAssignmentArgs(reassignmentJson);

        Map<TopicPartition, List<Integer>> proposedParts = t0.getKey();
        Map<TopicPartitionReplica, String> proposedReplicas = t0.getValue();
        Map<TopicPartition, PartitionReassignment> currentReassignments = adminClient.
            listPartitionReassignments().reassignments().get();
        // If there is an existing assignment, check for --additional before proceeding.
        // This helps avoid surprising users.
        if (!additional && !currentReassignments.isEmpty()) {
            throw new TerseException(CANNOT_EXECUTE_BECAUSE_OF_EXISTING_MESSAGE);
        }
        Set<Integer> brokers = new HashSet<>();
        proposedParts.values().forEach(brokers::addAll);

        verifyBrokerIds(adminClient, brokers);
        Map<TopicPartition, List<Integer>> currentParts = getReplicaAssignmentForPartitions(adminClient, proposedParts.keySet());
        System.out.println(currentPartitionReplicaAssignmentToString(proposedParts, currentParts));

        if (interBrokerThrottle >= 0 || logDirThrottle >= 0) {
            System.out.println(YOU_MUST_RUN_VERIFY_PERIODICALLY_MESSAGE);

            if (interBrokerThrottle >= 0) {
                Map<String, Map<Integer, PartitionMove>> moveMap = calculateProposedMoveMap(currentReassignments, proposedParts, currentParts);
                modifyReassignmentThrottle(adminClient, moveMap, interBrokerThrottle);
            }

            if (logDirThrottle >= 0) {
                Set<Integer> movingBrokers = calculateMovingBrokers(proposedReplicas.keySet());
                modifyLogDirThrottle(adminClient, movingBrokers, logDirThrottle);
            }
        }

        // Execute the partition reassignments.
        Map<TopicPartition, Throwable> errors = alterPartitionReassignments(adminClient, proposedParts);
        if (!errors.isEmpty()) {
            throw new TerseException(
                String.format("Error reassigning partition(s):%n%s",
                    errors.keySet().stream()
                        .sorted(ReassignPartitionsCommand::compareTopicPartitions)
                        .map(part -> part + ": " + errors.get(part).getMessage())
                        .collect(Collectors.joining(System.lineSeparator()))));
        }
        System.out.printf("Successfully started partition reassignment%s for %s%n",
            proposedParts.size() == 1 ? "" : "s",
            proposedParts.keySet().stream()
                .sorted(ReassignPartitionsCommand::compareTopicPartitions)
                .map(Objects::toString)
                .collect(Collectors.joining(",")));
        if (!proposedReplicas.isEmpty()) {
            executeMoves(adminClient, proposedReplicas, timeoutMs, time);
        }
    }
The entry point for the --execute and --execute-additional commands.

@param adminClient                 The AdminClient to use.
@param additional                  Whether --additional was passed.
@param reassignmentJson            The JSON string to use for the topics to reassign.
@param interBrokerThrottle         The inter-broker throttle to use, or a negative
                                   number to skip using a throttle.
@param logDirThrottle              The replica log directory throttle to use, or a
                                   negative number to skip using a throttle.
@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