@Override
    public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
        wrapped().put(sessionKey, aggregate);
        context.logChange(name(), SessionKeySchema.toBinary(sessionKey), aggregate, context.timestamp(), wrapped().getPosition());
    }
Simple wrapper around a {@link SessionStore} to support writing
updates to a changelog
streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingSessionBytesStore.java