private void populatePartitionsByHostMaps(final Map<HostInfo, Set<TopicPartition>> partitionsByHost,
                                              final Map<HostInfo, Set<TopicPartition>> standbyPartitionsByHost,
                                              final Map<TaskId, Set<TopicPartition>> partitionsForTask,
                                              final Map<ProcessId, ClientMetadata> clientMetadataMap) {
        for (final Map.Entry<ProcessId, ClientMetadata> entry : clientMetadataMap.entrySet()) {
            final HostInfo hostInfo = entry.getValue().hostInfo;

            // if application server is configured, also include host state map
            if (hostInfo != null) {
                final Set<TopicPartition> topicPartitions = new HashSet<>();
                final Set<TopicPartition> standbyPartitions = new HashSet<>();
                final ClientState state = entry.getValue().state;

                for (final TaskId id : state.activeTasks()) {
                    topicPartitions.addAll(partitionsForTask.get(id));
                }

                for (final TaskId id : state.standbyTasks()) {
                    standbyPartitions.addAll(partitionsForTask.get(id));
                }

                partitionsByHost.put(hostInfo, topicPartitions);
                standbyPartitionsByHost.put(hostInfo, standbyPartitions);
            }
        }
    }
Populates the global partitionsByHost and standbyPartitionsByHost maps that are sent to each member

@param partitionsByHost a map from host to the set of partitions hosted there. Populated here.
@param standbyPartitionsByHost a map from host to the set of standby partitions hosted there. Populated here.
@param partitionsForTask a map from task to its set of assigned partitions
@param clientMetadataMap a map from client to its metadata and state
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java