static Entry<Set<TopicPartition>, Set<TopicPartitionReplica>> cancelAssignment(Admin adminClient,
                                                                                    String jsonString,
                                                                                    Boolean preserveThrottles,
                                                                                    Long timeoutMs,
                                                                                    Time time
    ) throws ExecutionException, InterruptedException, JsonProcessingException, TerseException {
        Entry<List<Entry<TopicPartition, List<Integer>>>, Map<TopicPartitionReplica, String>> t0 = parsePartitionReassignmentData(jsonString);

        List<Entry<TopicPartition, List<Integer>>> targetParts = t0.getKey();
        Map<TopicPartitionReplica, String> targetReplicas = t0.getValue();
        Set<TopicPartition> targetPartsSet = targetParts.stream().map(t -> t.getKey()).collect(Collectors.toSet());
        Set<TopicPartition> curReassigningParts = new HashSet<>();
        adminClient.listPartitionReassignments(targetPartsSet).reassignments().get().forEach((part, reassignment) -> {
            if (reassignment.addingReplicas().isEmpty() || !reassignment.removingReplicas().isEmpty())
                curReassigningParts.add(part);
        });
        if (!curReassigningParts.isEmpty()) {
            Map<TopicPartition, Throwable> errors = cancelPartitionReassignments(adminClient, curReassigningParts);
            if (!errors.isEmpty()) {
                throw new TerseException(String.format(
                    "Error cancelling partition reassignment%s for:%n%s",
                    errors.size() == 1 ? "" : "s",
                    errors.keySet().stream()
                        .sorted(ReassignPartitionsCommand::compareTopicPartitions)
                        .map(part -> part + ": " + errors.get(part).getMessage()).collect(Collectors.joining(System.lineSeparator())))
                );
            }
            System.out.printf("Successfully cancelled partition reassignment%s for: %s%n",
                curReassigningParts.size() == 1 ? "" : "s",
                curReassigningParts.stream().sorted(ReassignPartitionsCommand::compareTopicPartitions).map(Object::toString).collect(Collectors.joining(","))
            );
        } else {
            System.out.println("None of the specified partition reassignments are active.");
        }
        Map<TopicPartitionReplica, String> curMovingParts = new HashMap<>();
        findLogDirMoveStates(adminClient, targetReplicas).forEach((part, moveState) -> {
            if (moveState instanceof ActiveMoveState)
                curMovingParts.put(part, ((ActiveMoveState) moveState).currentLogDir);
        });
        if (curMovingParts.isEmpty()) {
            System.out.print("None of the specified partition moves are active.");
        } else {
            executeMoves(adminClient, curMovingParts, timeoutMs, time);
        }
        if (!preserveThrottles) {
            clearAllThrottles(adminClient, targetParts);
        }
        return new SimpleImmutableEntry<>(curReassigningParts, curMovingParts.keySet());
    }
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