@SuppressWarnings("unchecked")
    private static <R> QueryResult<R> runWindowKeyQuery(final Query<R> query,
                                                        final PositionBound positionBound,
                                                        final QueryConfig config,
                                                        final StateStore store) {
        if (store instanceof WindowStore) {
            final WindowKeyQuery<Bytes, byte[]> windowKeyQuery =
                (WindowKeyQuery<Bytes, byte[]>) query;
            final WindowStore<Bytes, byte[]> windowStore = (WindowStore<Bytes, byte[]>) store;
            try {
                if (windowKeyQuery.getTimeFrom().isPresent() && windowKeyQuery.getTimeTo().isPresent()) {
                    final WindowStoreIterator<byte[]> iterator = windowStore.fetch(
                        windowKeyQuery.getKey(),
                        windowKeyQuery.getTimeFrom().get(),
                        windowKeyQuery.getTimeTo().get()
                    );
                    return (QueryResult<R>) QueryResult.forResult(iterator);
                } else {
                    return QueryResult.forFailure(
                        FailureReason.UNKNOWN_QUERY_TYPE,
                        "This store (" + store.getClass() + ") doesn't know how to"
                            + " execute the given query (" + query + ") because it only supports"
                            + " closed-range queries."
                            + " Contact the store maintainer if you need support"
                            + " for a new query type."
                    );
                }
            } 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