@SuppressWarnings("unchecked")
        private <R> QueryResult<R> runMultiVersionedKeyQuery(final Query<R> query, final PositionBound positionBound, final QueryConfig config) {
            final QueryResult<R> result;
            final MultiVersionedKeyQuery<K, V> typedKeyQuery = (MultiVersionedKeyQuery<K, V>) query;

            final Instant fromTime = typedKeyQuery.fromTime().orElse(Instant.ofEpochMilli(Long.MIN_VALUE));
            final Instant toTime = typedKeyQuery.toTime().orElse(Instant.ofEpochMilli(Long.MAX_VALUE));
            if (fromTime.compareTo(toTime) > 0) {
                throw new IllegalArgumentException("The `fromTime` timestamp must be smaller than the `toTime` timestamp.");
            }
            MultiVersionedKeyQuery<Bytes, byte[]> rawKeyQuery = MultiVersionedKeyQuery.withKey(keyBytes(typedKeyQuery.key()));
            rawKeyQuery = rawKeyQuery.fromTime(fromTime).toTime(toTime);
            if (typedKeyQuery.resultOrder().equals(ResultOrder.DESCENDING)) {
                rawKeyQuery = rawKeyQuery.withDescendingTimestamps();
            } else if (typedKeyQuery.resultOrder().equals(ResultOrder.ASCENDING)) {
                rawKeyQuery = rawKeyQuery.withAscendingTimestamps();
            }

            final QueryResult<VersionedRecordIterator<byte[]>> rawResult = wrapped().query(rawKeyQuery, positionBound, config);
            if (rawResult.isSuccess()) {
                final MeteredMultiVersionedKeyQueryIterator<V> typedResult =
                        new MeteredMultiVersionedKeyQueryIterator<V>(
                            rawResult.getResult(),
                            iteratorDurationSensor,
                            time,
                            StoreQueryUtils.getDeserializeValue(plainValueSerdes),
                            numOpenIterators,
                            openIterators
                        );
                final QueryResult<MeteredMultiVersionedKeyQueryIterator<V>> typedQueryResult =
                        InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult, typedResult);
                result = (QueryResult<R>) typedQueryResult;
            } else {
                // the generic type doesn't matter, since failed queries have no result set.
                result = (QueryResult<R>) rawResult;
            }
            return result;
        }
Conceptually, {@link MeteredVersionedKeyValueStore} should {@code extend}
{@link MeteredKeyValueStore}, but due to type conflicts, we cannot do this. (Specifically,
the first needs to be {@link VersionedKeyValueStore} while the second is {@link KeyValueStore}
and the two interfaces conflict.) Thus, we use an internal <it>instance</it> of
{@code MeteredKeyValueStore} to mimic inheritance instead.
<p>
It's not ideal because it requires an extra step to translate between the APIs of
{@link VersionedKeyValueStore} in {@link MeteredVersionedKeyValueStore} and
the APIs of {@link TimestampedKeyValueStore} in {@link MeteredVersionedKeyValueStoreInternal}.
This extra step is all that the methods of {@code MeteredVersionedKeyValueStoreInternal} do.
<p>
Note that the addition of {@link #get(Object, long)} and {@link #delete(Object, long)} in
this class are to match the interface of {@link VersionedKeyValueStore}.
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java