private static Entry<List<Entry<TopicPartition, List<Integer>>>, Map<TopicPartitionReplica, String>> parsePartitionReassignmentData(
        int version, JsonValue jsonData
    ) throws JsonMappingException {
        switch (version) {
            case 1:
                List<Entry<TopicPartition, List<Integer>>> partitionAssignment = new ArrayList<>();
                Map<TopicPartitionReplica, String> replicaAssignment = new HashMap<>();

                Optional<JsonValue> partitionsSeq = jsonData.asJsonObject().get("partitions");
                if (partitionsSeq.isPresent()) {
                    Iterator<JsonValue> iter = partitionsSeq.get().asJsonArray().iterator();
                    while (iter.hasNext()) {
                        JsonObject partitionFields = iter.next().asJsonObject();
                        String topic = partitionFields.apply("topic").to(STRING);
                        int partition = partitionFields.apply("partition").to(INT);
                        List<Integer> newReplicas = partitionFields.apply("replicas").to(INT_LIST);
                        Optional<JsonValue> logDirsOpts = partitionFields.get("log_dirs");
                        List<String> newLogDirs;
                        if (logDirsOpts.isPresent())
                            newLogDirs = logDirsOpts.get().to(STRING_LIST);
                        else
                            newLogDirs = newReplicas.stream().map(r -> ANY_LOG_DIR).collect(Collectors.toList());
                        if (newReplicas.size() != newLogDirs.size())
                            throw new AdminCommandFailedException("Size of replicas list " + newReplicas + " is different from " +
                                "size of log dirs list " + newLogDirs + " for partition " + new TopicPartition(topic, partition));
                        partitionAssignment.add(new SimpleImmutableEntry<>(new TopicPartition(topic, partition), newReplicas));
                        for (int i = 0; i < newLogDirs.size(); i++) {
                            Integer replica = newReplicas.get(i);
                            String logDir = newLogDirs.get(i);

                            if (logDir.equals(ANY_LOG_DIR))
                                continue;

                            replicaAssignment.put(new TopicPartitionReplica(topic, partition, replica), logDir);
                        }
                    }
                }

                return new SimpleImmutableEntry<>(partitionAssignment, replicaAssignment);

            default:
                throw new AdminOperationException("Not supported version field value " + version);
        }
    }
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