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

        if (store instanceof VersionedKeyValueStore) {
            final RocksDBVersionedStore rocksDBVersionedStore = (RocksDBVersionedStore) store;
            final MultiVersionedKeyQuery<Bytes, byte[]> rawKeyQuery = (MultiVersionedKeyQuery<Bytes, byte[]>) query;
            try {
                final VersionedRecordIterator<byte[]> segmentIterator =
                        rocksDBVersionedStore.get(rawKeyQuery.key(),
                                                  rawKeyQuery.fromTime().get().toEpochMilli(),
                                                  rawKeyQuery.toTime().get().toEpochMilli(),
                                                  rawKeyQuery.resultOrder());
                return (QueryResult<R>) QueryResult.forResult(segmentIterator);
            } 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