class MeteredSessionStore extends WrappedStateStore<SessionStore<Bytes, byte[]>, Windowed<K>, V> implements SessionStore<K, V> {

    @SuppressWarnings("unchecked")
    @Override
    public boolean setFlushListener(final CacheFlushListener<Windowed<K>, V> listener,
                                    final boolean sendOldValues) {
        final SessionStore<Bytes, byte[]> wrapped = wrapped();
        if (wrapped instanceof CachedStateStore) {
            return ((CachedStateStore<byte[], byte[]>) wrapped).setFlushListener(
                (key, newValue, oldValue, timestamp) -> listener.apply(
                    SessionKeySchema.from(key, serdes.keyDeserializer(), serdes.topic()),
                    newValue != null ? serdes.valueFrom(newValue) : null,
                    oldValue != null ? serdes.valueFrom(oldValue) : null,
                    timestamp
                ),
                sendOldValues);
        }
        return false;
    }

      MeteredSessionStore(final SessionStore<Bytes, byte[]> inner,
                        final String metricsScope,
                        final Serde<K> keySerde,
                        final Serde<V> valueSerde,
                        final Time time);

}