private List<StreamsMetadata> rebuildMetadataForNamedTopologies(final Map<HostInfo, Set<TopicPartition>> activePartitionHostMap,
                                                                    final Map<HostInfo, Set<TopicPartition>> standbyPartitionHostMap) {
        final List<StreamsMetadata> rebuiltMetadata = new ArrayList<>();
        Stream.concat(activePartitionHostMap.keySet().stream(), standbyPartitionHostMap.keySet().stream())
            .distinct()
            .sorted(Comparator.comparing(HostInfo::host).thenComparingInt(HostInfo::port))
            .forEach(hostInfo -> {
                for (final String topologyName : topologyMetadata.namedTopologiesView()) {
                    final Map<String, List<String>> storeToSourceTopics =
                        topologyMetadata.stateStoreNameToSourceTopicsForTopology(topologyName);

                    final Set<TopicPartition> activePartitionsOnHost = new HashSet<>();
                    final Set<String> activeStoresOnHost = new HashSet<>();
                    if (activePartitionHostMap.containsKey(hostInfo)) {
                        // filter out partitions for topics that are not connected to this topology
                        activePartitionsOnHost.addAll(
                            activePartitionHostMap.get(hostInfo).stream()
                                .filter(tp -> topologyMetadata.fullSourceTopicNamesForTopology(topologyName).contains(tp.topic()))
                                .collect(Collectors.toSet())
                        );
                        activeStoresOnHost.addAll(getStoresOnHost(storeToSourceTopics, activePartitionsOnHost));
                    }

                    // TODO KAFKA-13281: when we add support for global stores with named topologies we will
                    //  need to add the global stores to the activeStoresOnHost set

                    final Set<TopicPartition> standbyPartitionsOnHost = new HashSet<>();
                    final Set<String> standbyStoresOnHost = new HashSet<>();
                    if (standbyPartitionHostMap.containsKey(hostInfo)) {
                        standbyPartitionsOnHost.addAll(
                            standbyPartitionHostMap.get(hostInfo).stream()
                                .filter(tp -> topologyMetadata.fullSourceTopicNamesForTopology(topologyName).contains(tp.topic()))
                                .collect(Collectors.toSet()));
                        standbyStoresOnHost.addAll(getStoresOnHost(storeToSourceTopics, standbyPartitionsOnHost));
                    }

                    final StreamsMetadata metadata = new StreamsMetadataImpl(
                        hostInfo,
                        activeStoresOnHost,
                        activePartitionsOnHost,
                        standbyStoresOnHost,
                        standbyPartitionsOnHost,
                        topologyName
                    );

                    rebuiltMetadata.add(metadata);
                    if (hostInfo.equals(thisHost)) {
                        localMetadata.set(metadata);
                    }
                }

                // Construct metadata across all topologies on this host for the `localMetadata` field
                final Map<String, List<String>> storeToSourceTopics = topologyMetadata.stateStoreNameToSourceTopics();
                final Set<TopicPartition> localActivePartitions = activePartitionHostMap.get(thisHost);
                final Set<TopicPartition> localStandbyPartitions = standbyPartitionHostMap.get(thisHost);

                localMetadata.set(
                    new StreamsMetadataImpl(thisHost,
                                            getStoresOnHost(storeToSourceTopics, localActivePartitions),
                                            localActivePartitions,
                                            getStoresOnHost(storeToSourceTopics, localStandbyPartitions),
                                            localStandbyPartitions)
                );
            });
        return rebuiltMetadata;
    }
Respond to changes to the HostInfo -> TopicPartition mapping. Will rebuild the
metadata

@param activePartitionHostMap  the current mapping of {@link HostInfo} -> {@link TopicPartition}s for active partitions
@param standbyPartitionHostMap the current mapping of {@link HostInfo} -> {@link TopicPartition}s for standby partitions
@param topicPartitionInfo      the current mapping of {@link TopicPartition} -> {@Link PartitionInfo}
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java