@Override
    public long put(final Bytes key, final byte[] value, final long timestamp) {
        final long validTo = inner.put(key, value, timestamp);
        log(key, value, timestamp);
        return validTo;
    }
A {@link VersionedBytesStore} wrapper for writing changelog records on calls to
{@link VersionedBytesStore#put(Object, Object)} and {@link VersionedBytesStore#delete(Bytes, long)}.
streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingVersionedKeyValueBytesStore.java