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

    @Override
    public void put(final Windowed<K> sessionKey,
                    final V aggregate) {
        Objects.requireNonNull(sessionKey, "sessionKey can't be null");
        try {
            maybeMeasureLatency(
                () -> {
                    final Bytes key = keyBytes(sessionKey.key());
                    wrapped().put(new Windowed<>(key, sessionKey.window()), serdes.rawValue(aggregate));
                },
                time,
                putSensor
            );
            maybeRecordE2ELatency();
        } catch (final ProcessorStateException e) {
            final String message = String.format(e.getMessage(), sessionKey.key(), aggregate);
            throw new ProcessorStateException(message, e);
        }
    }

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

}