private <K> KeyQueryMetadata getKeyQueryMetadataForKey(final String storeName,
                                                           final K key,
                                                           final StreamPartitioner<? super K, ?> partitioner,
                                                           final SourceTopicsInfo sourceTopicsInfo,
                                                           final String topologyName) {
        Objects.requireNonNull(topologyName, "topology name must not be null");
        final Integer partition = getPartition.apply(partitioner.partitions(sourceTopicsInfo.topicWithMostPartitions, key, null, sourceTopicsInfo.maxPartitions));
        final Set<TopicPartition> matchingPartitions = new HashSet<>();
        for (final String sourceTopic : sourceTopicsInfo.sourceTopics) {
            matchingPartitions.add(new TopicPartition(sourceTopic, partition));
        }

        HostInfo activeHost = UNKNOWN_HOST;
        final Set<HostInfo> standbyHosts = new HashSet<>();
        for (final StreamsMetadata streamsMetadata : allMetadata) {
            final String metadataTopologyName = ((StreamsMetadataImpl) streamsMetadata).topologyName();
            if (metadataTopologyName != null && metadataTopologyName.equals(topologyName)) {

                final Set<String> activeStateStoreNames = streamsMetadata.stateStoreNames();
                final Set<TopicPartition> topicPartitions = new HashSet<>(streamsMetadata.topicPartitions());
                final Set<String> standbyStateStoreNames = streamsMetadata.standbyStateStoreNames();
                final Set<TopicPartition> standbyTopicPartitions = new HashSet<>(streamsMetadata.standbyTopicPartitions());

                topicPartitions.retainAll(matchingPartitions);
                if (activeStateStoreNames.contains(storeName) && !topicPartitions.isEmpty()) {
                    activeHost = streamsMetadata.hostInfo();
                }

                standbyTopicPartitions.retainAll(matchingPartitions);
                if (standbyStateStoreNames.contains(storeName) && !standbyTopicPartitions.isEmpty()) {
                    standbyHosts.add(streamsMetadata.hostInfo());
                }
            }
        }

        return new KeyQueryMetadata(activeHost, standbyHosts, partition);
    }
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