static Map<String, Map<Integer, PartitionMove>> calculateProposedMoveMap(Map<TopicPartition, PartitionReassignment> currentReassignments,
                                                                             Map<TopicPartition, List<Integer>> proposedParts,
                                                                             Map<TopicPartition, List<Integer>> currentParts) {
        Map<String, Map<Integer, PartitionMove>> moveMap = calculateCurrentMoveMap(currentReassignments);
        for (Entry<TopicPartition, List<Integer>> e : proposedParts.entrySet()) {
            TopicPartition part = e.getKey();
            List<Integer> replicas = e.getValue();
            Map<Integer, PartitionMove> partMoves = moveMap.computeIfAbsent(part.topic(), k -> new HashMap<>());

            // If there is a reassignment in progress, use the sources from moveMap, otherwise
            // use the sources from currentParts
            Set<Integer> sources = new HashSet<>();

            if (partMoves.containsKey(part.partition())) {
                PartitionMove move = partMoves.get(part.partition());
                sources.addAll(move.sources);
            } else if (currentParts.containsKey(part))
                sources.addAll(currentParts.get(part));
            else
                throw new RuntimeException("Trying to reassign a topic partition " + part + " with 0 replicas");

            Set<Integer> destinations = new HashSet<>(replicas);
            destinations.removeAll(sources);

            partMoves.put(part.partition(), new PartitionMove(sources, destinations));
        }
        return moveMap;
    }
Calculate the global map of all partitions that are moving.

@param currentReassignments    The currently active reassignments.
@param proposedParts           The proposed location of the partitions (destinations replicas only).
@param currentParts            The current location of the partitions that we are
                               proposing to move.
@return                        A map from topic name to partition map.
                               The partition map is keyed on partition index and contains
                               the movements for that partition.
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java