public synchronized <K> KeyQueryMetadata getKeyQueryMetadataForKey(final String storeName,
                                                                       final K key,
                                                                       final StreamPartitioner<? super K, ?> partitioner) {
        Objects.requireNonNull(storeName, "storeName can't be null");
        Objects.requireNonNull(key, "key can't be null");
        Objects.requireNonNull(partitioner, "partitioner can't be null");
        if (topologyMetadata.hasNamedTopologies()) {
            throw new IllegalArgumentException("Cannot invoke the getKeyQueryMetadataForKey(storeName, key, partitioner)"
                                                   + "method when using named topologies, please use the overload that"
                                                   + "accepts a topologyName parameter to identify the correct store");
        }

        if (!isInitialized()) {
            return KeyQueryMetadata.NOT_AVAILABLE;
        }

        if (globalStores.contains(storeName)) {
            // global stores are on every node. if we don't have the host info
            // for this host then just pick the first metadata
            if (thisHost.equals(UNKNOWN_HOST)) {
                return new KeyQueryMetadata(allMetadata.get(0).hostInfo(), Collections.emptySet(), UNKNOWN_PARTITION);
            }
            return new KeyQueryMetadata(localMetadata.get().hostInfo(), Collections.emptySet(), UNKNOWN_PARTITION);
        }

        final SourceTopicsInfo sourceTopicsInfo = getSourceTopicsInfo(storeName);
        if (sourceTopicsInfo == null) {
            return null;
        }
        return getKeyQueryMetadataForKey(storeName, key, partitioner, sourceTopicsInfo);
    }
Find the {@link KeyQueryMetadata}s for a given storeName and key

Note: the key may not exist in the {@link StateStore},this method provides a way of finding which
{@link StreamsMetadata} it would exist on.

@param storeName   Name of the store
@param key         Key to use
@param partitioner partitioner to use to find correct partition for key
@param <K>         key type
@return The {@link KeyQueryMetadata} for the storeName and key or {@link KeyQueryMetadata#NOT_AVAILABLE}
if streams is (re-)initializing, or {@code null} if no matching metadata could be found.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java