@SuppressWarnings("unchecked")
    @Override
    public boolean setFlushListener(final CacheFlushListener<K, V> listener,
                                    final boolean sendOldValues) {
        final KeyValueStore<Bytes, byte[]> wrapped = wrapped();
        if (wrapped instanceof CachedStateStore) {
            return ((CachedStateStore<byte[], byte[]>) wrapped).setFlushListener(
                record -> listener.apply(
                    record.withKey(serdes.keyFrom(record.key()))
                        .withValue(new Change<>(
                            record.value().newValue != null ? serdes.valueFrom(record.value().newValue) : null,
                            record.value().oldValue != null ? serdes.valueFrom(record.value().oldValue) : null,
                            record.value().isLatest
                        ))
                ),
                sendOldValues);
        }
        return false;
    }
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