private void populateTasksForMaps(final Map<TopicPartition, TaskId> taskForPartition,
                                      final Map<Subtopology, Set<TaskId>> tasksForTopicGroup,
                                      final Set<String> allSourceTopics,
                                      final Map<TaskId, Set<TopicPartition>> partitionsForTask,
                                      final Cluster fullMetadata) {
        // check if all partitions are assigned, and there are no duplicates of partitions in multiple tasks
        final Set<TopicPartition> allAssignedPartitions = new HashSet<>();
        for (final Map.Entry<TaskId, Set<TopicPartition>> entry : partitionsForTask.entrySet()) {
            final TaskId id = entry.getKey();
            final Set<TopicPartition> partitions = entry.getValue();

            for (final TopicPartition partition : partitions) {
                taskForPartition.put(partition, id);
                if (allAssignedPartitions.contains(partition)) {
                    log.warn("Partition {} is assigned to more than one tasks: {}", partition, partitionsForTask);
                }
            }
            allAssignedPartitions.addAll(partitions);

            tasksForTopicGroup.computeIfAbsent(new Subtopology(id.subtopology(), id.topologyName()), k -> new HashSet<>()).add(id);
        }

        checkAllPartitions(allSourceTopics, partitionsForTask, allAssignedPartitions, fullMetadata);
    }
Populates the taskForPartition and tasksForTopicGroup maps, and checks that partitions are assigned to exactly
one task.

@param taskForPartition a map from partition to the corresponding task. Populated here.
@param tasksForTopicGroup a map from the topicGroupId to the set of corresponding tasks. Populated here.
@param allSourceTopics a set of all source topics in the topology
@param partitionsForTask a map from task to the set of input partitions
@param fullMetadata the cluster metadata
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java