@SuppressWarnings("unchecked")
    private  <R> QueryResult<R> runKeyQuery(final Query<R> query,
                                            final PositionBound positionBound,
                                            final QueryConfig config) {
        final QueryResult<R> result;
        final KeyQuery<K, V> typedKeyQuery = (KeyQuery<K, V>) query;
        final KeyQuery<Bytes, byte[]> rawKeyQuery =
            KeyQuery.withKey(keyBytes(typedKeyQuery.getKey()));
        final QueryResult<byte[]> rawResult =
            wrapped().query(rawKeyQuery, positionBound, config);
        if (rawResult.isSuccess()) {
            final Function<byte[], V> deserializer = getDeserializeValue(serdes, wrapped());
            final V value = deserializer.apply(rawResult.getResult());
            final QueryResult<V> typedQueryResult =
                InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult, value);
            result = (QueryResult<R>) typedQueryResult;
        } else {
            // the generic type doesn't matter, since failed queries have no result set.
            result = (QueryResult<R>) rawResult;
        }
        return result;
    }
A Metered {@link KeyValueStore} wrapper that is used for recording operation metrics, and hence its
inner KeyValueStore implementation do not need to provide its own metrics collecting functionality.
The inner {@link KeyValueStore} of this class is of type &lt;Bytes,byte[]&gt;, hence we use {@link Serde}s
to convert from &lt;K,V&gt; to &lt;Bytes,byte[]&gt;

@param <K>
@param <V>
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java