public static String formatAsReassignmentJson(Map<TopicPartition, List<Integer>> partitionsToBeReassigned,
                                                   Map<TopicPartitionReplica, String> replicaLogDirAssignment) throws JsonProcessingException {
        List<Map<String, Object>> partitions = new ArrayList<>();
        partitionsToBeReassigned.keySet().stream().sorted(ReassignPartitionsCommand::compareTopicPartitions).forEach(tp -> {
            List<Integer> replicas = partitionsToBeReassigned.get(tp);
            Map<String, Object> data = new LinkedHashMap<>();

            data.put("topic", tp.topic());
            data.put("partition", tp.partition());
            data.put("replicas", replicas);
            data.put("log_dirs", replicas.stream()
                .map(r -> replicaLogDirAssignment.getOrDefault(new TopicPartitionReplica(tp.topic(), tp.partition(), r), ANY_LOG_DIR))
                .collect(Collectors.toList()));

            partitions.add(data);
        });

        Map<String, Object> results = new LinkedHashMap<>();

        results.put("version", 1);
        results.put("partitions", partitions);

        return Json.encodeAsString(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