private ApplicationState buildApplicationState(final TopologyMetadata topologyMetadata,
                                                   final Map<ProcessId, ClientMetadata> clientMetadataMap,
                                                   final Map<Subtopology, TopicsInfo> topicGroups,
                                                   final Cluster cluster) {
        final Map<Subtopology, Set<String>> sourceTopicsByGroup = new HashMap<>();
        final Map<Subtopology, Set<String>> changelogTopicsByGroup = new HashMap<>();
        for (final Map.Entry<Subtopology, TopicsInfo> entry : topicGroups.entrySet()) {
            final Set<String> sourceTopics = entry.getValue().sourceTopics;
            final Set<String> changelogTopics = entry.getValue().changelogTopics();
            sourceTopicsByGroup.put(entry.getKey(), sourceTopics);
            changelogTopicsByGroup.put(entry.getKey(), changelogTopics);
        }

        final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask = new HashMap<>();
        final Map<TaskId, Set<TopicPartition>> sourcePartitionsForTask =
            partitionGrouper.partitionGroups(sourceTopicsByGroup, changelogTopicsByGroup,
                changelogPartitionsForTask, cluster);

        if (!sourcePartitionsForTask.keySet().equals(changelogPartitionsForTask.keySet())) {
            log.error("Partition grouper returned {} tasks for source topics but {} tasks for changelog topics",
                sourcePartitionsForTask.size(), changelogPartitionsForTask.size());
            throw new TaskAssignmentException("Partition grouper returned conflicting information about the "
                                              + "tasks for source topics vs changelog topics.");
        }

        final Set<DefaultTaskTopicPartition> topicsRequiringRackInfo = new HashSet<>();
        final AtomicBoolean rackInformationFetched = new AtomicBoolean(false);
        final Runnable fetchRackInformation = () -> {
            if (!rackInformationFetched.get()) {
                RackUtils.annotateTopicPartitionsWithRackInfo(cluster, internalTopicManager, topicsRequiringRackInfo);
                rackInformationFetched.set(true);
            }
        };

        final Map<TaskId, Set<TaskTopicPartition>> topicPartitionsForTask = new HashMap<>();
        final Set<TaskId> logicalTaskIds = unmodifiableSet(sourcePartitionsForTask.keySet());
        logicalTaskIds.forEach(taskId -> {
            final Set<TaskTopicPartition> topicPartitions = new HashSet<>();

            for (final TopicPartition topicPartition : sourcePartitionsForTask.get(taskId)) {
                final boolean isSource = true;
                final boolean isChangelog = changelogPartitionsForTask.get(taskId).contains(topicPartition);
                final DefaultTaskTopicPartition racklessTopicPartition = new DefaultTaskTopicPartition(
                    topicPartition, isSource, isChangelog, fetchRackInformation);
                topicsRequiringRackInfo.add(racklessTopicPartition);
                topicPartitions.add(racklessTopicPartition);
            }

            for (final TopicPartition topicPartition : changelogPartitionsForTask.get(taskId)) {
                final boolean isSource = sourcePartitionsForTask.get(taskId).contains(topicPartition);
                final boolean isChangelog = true;
                final DefaultTaskTopicPartition racklessTopicPartition = new DefaultTaskTopicPartition(
                    topicPartition, isSource, isChangelog, fetchRackInformation);
                topicsRequiringRackInfo.add(racklessTopicPartition);
                topicPartitions.add(racklessTopicPartition);
            }

            topicPartitionsForTask.put(taskId, topicPartitions);
        });

        final Map<TaskId, TaskInfo> logicalTasks = logicalTaskIds.stream().collect(Collectors.toMap(
            Function.identity(),
            taskId -> {
                final Set<String> stateStoreNames = topologyMetadata
                    .stateStoreNamesForSubtopology(taskId.topologyName(), taskId.subtopology());
                final Set<TaskTopicPartition> topicPartitions = topicPartitionsForTask.get(taskId);
                return new DefaultTaskInfo(
                    taskId,
                    !stateStoreNames.isEmpty(),
                    stateStoreNames,
                    topicPartitions
                );
            }
        ));

        return new DefaultApplicationState(
            assignmentConfigs,
            logicalTasks,
            clientMetadataMap
        );
    }
@param clientMetadataMap the map of process id to client metadata used to build an immutable
                         {@code ApplicationState}
@return The {@code ApplicationState} needed by the TaskAssigner to compute new task
        assignments.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java