private static Map<String, Map<Integer, PartitionMove>> calculateCurrentMoveMap(Map<TopicPartition, PartitionReassignment> currentReassignments) {
        Map<String, Map<Integer, PartitionMove>> moveMap = new HashMap<>();
        // Add the current reassignments to the move map.
        currentReassignments.forEach((part, reassignment) -> {
            List<Integer> allReplicas = reassignment.replicas();
            List<Integer> addingReplicas = reassignment.addingReplicas();

            // The addingReplicas is included in the replicas during reassignment
            Set<Integer> sources = new HashSet<>(allReplicas);
            addingReplicas.forEach(sources::remove);

            Set<Integer> destinations = new HashSet<>(addingReplicas);

            Map<Integer, PartitionMove> partMoves = moveMap.computeIfAbsent(part.topic(), k -> new HashMap<>());
            partMoves.put(part.partition(), new PartitionMove(sources, destinations));
        });
        return moveMap;
    }
Compute the in progress partition move from the current reassignments.
@param currentReassignments All replicas, adding replicas and removing replicas of target partitions
tools/src/main/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommand.java