@SuppressWarnings("unchecked")
    private static <R> QueryResult<R> runKeyQuery(final Query<R> query,
                                                  final PositionBound positionBound,
                                                  final QueryConfig config,
                                                  final StateStore store) {

        if (store instanceof KeyValueStore) {
            final KeyQuery<Bytes, byte[]> rawKeyQuery = (KeyQuery<Bytes, byte[]>) query;
            final KeyValueStore<Bytes, byte[]> keyValueStore =
                (KeyValueStore<Bytes, byte[]>) store;
            try {
                final byte[] bytes = keyValueStore.get(rawKeyQuery.getKey());
                return (QueryResult<R>) QueryResult.forResult(bytes);
            } catch (final Exception e) {
                final String message = parseStoreException(e, store, query);
                return QueryResult.forFailure(
                    FailureReason.STORE_EXCEPTION,
                    message
                );
            }
        } else {
            return QueryResult.forUnknownQueryType(query, store);
        }
    }
a utility interface to facilitate stores' query dispatch logic,
allowing them to generically store query execution logic as the values
in a map.
streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java