@Override
        public KeyValue<K, V> next() {
            final KeyValue<Bytes, byte[]> keyValue = iter.next();
            if (returnPlainValue) {
                final V plainValue = valueAndTimestampDeserializer.apply(keyValue.value).value();
                return KeyValue.pair(
                        serdes.keyFrom(keyValue.key.get()), plainValue);
            }
            return (KeyValue<K, V>) KeyValue.pair(
                    serdes.keyFrom(keyValue.key.get()),
                    valueAndTimestampDeserializer.apply(keyValue.value));
        }
A Metered {@link TimestampedKeyValueStore} 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,ValueAndTimestamp&lt;V&gt&gt; to &lt;Bytes,byte[]&gt;
@param <K>
@param <V>
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java