@SuppressWarnings("unchecked")
    private static <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query,
                                                           final PositionBound positionBound,
                                                           final QueryConfig config,
                                                           final StateStore store) {
        if (store instanceof VersionedKeyValueStore) {
            final VersionedKeyValueStore<Bytes, byte[]> versionedKeyValueStore =
                (VersionedKeyValueStore<Bytes, byte[]>) store;
            final VersionedKeyQuery<Bytes, byte[]> rawKeyQuery =
                (VersionedKeyQuery<Bytes, byte[]>) query;
            try {
                final VersionedRecord<byte[]> bytes;
                if (((VersionedKeyQuery<?, ?>) query).asOfTimestamp().isPresent()) {
                    bytes = versionedKeyValueStore.get(rawKeyQuery.key(),
                        ((VersionedKeyQuery<?, ?>) query).asOfTimestamp().get().toEpochMilli());
                } else {
                    bytes = versionedKeyValueStore.get(rawKeyQuery.key());
                }
                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