private List<StreamsMetadata> rebuildMetadataForSingleTopology(final Map<HostInfo, Set<TopicPartition>> activePartitionHostMap,
                                                                   final Map<HostInfo, Set<TopicPartition>> standbyPartitionHostMap) {
        final List<StreamsMetadata> rebuiltMetadata = new ArrayList<>();
        final Map<String, List<String>> storeToSourceTopics = topologyMetadata.stateStoreNameToSourceTopics();
        Stream.concat(activePartitionHostMap.keySet().stream(), standbyPartitionHostMap.keySet().stream())
            .distinct()
            .sorted(Comparator.comparing(HostInfo::host).thenComparingInt(HostInfo::port))
            .forEach(hostInfo -> {
                final Set<TopicPartition> activePartitionsOnHost = new HashSet<>();
                final Set<String> activeStoresOnHost = new HashSet<>();
                if (activePartitionHostMap.containsKey(hostInfo)) {
                    activePartitionsOnHost.addAll(activePartitionHostMap.get(hostInfo));
                    activeStoresOnHost.addAll(getStoresOnHost(storeToSourceTopics, activePartitionsOnHost));
                }
                activeStoresOnHost.addAll(globalStores);

                final Set<TopicPartition> standbyPartitionsOnHost = new HashSet<>();
                final Set<String> standbyStoresOnHost = new HashSet<>();
                if (standbyPartitionHostMap.containsKey(hostInfo)) {
                    standbyPartitionsOnHost.addAll(standbyPartitionHostMap.get(hostInfo));
                    standbyStoresOnHost.addAll(getStoresOnHost(storeToSourceTopics, standbyPartitionsOnHost));
                }

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

                rebuiltMetadata.add(metadata);
                if (hostInfo.equals(thisHost)) {
                    localMetadata.set(metadata);
                }
            });
        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