@Override
    public boolean put(final long time,
                       final Record<K, Change<V>> record,
                       final ProcessorRecordContext recordContext) {
        requireNonNull(record.value(), "value cannot be null");
        requireNonNull(recordContext, "recordContext cannot be null");

        final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(changelogTopic, record.key()));
        final Change<byte[]> serialChange = valueSerde.serializeParts(changelogTopic, record.value());

        final BufferValue buffered = getBuffered(serializedKey);
        final byte[] serializedPriorValue;
        if (buffered == null) {
            serializedPriorValue = serialChange.oldValue;
        } else {
            serializedPriorValue = buffered.priorValue();
        }

        cleanPut(
            time,
            serializedKey,
            new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, recordContext)
        );
        if (loggingEnabled) {
            dirtyKeys.add(serializedKey);
        }
        updateBufferMetrics();
        return true;
    }
@Test
    public void bufferShouldAllowLoggingEnablement() {
        final String expect = "3";
        final Map<String, String> logConfig = new HashMap<>();
        logConfig.put("min.insync.replicas", expect);
        final StoreBuilder<InMemoryTimeOrderedKeyValueChangeBuffer<Object, Object, Change<Object>>> builder =
            new InMemoryTimeOrderedKeyValueChangeBuffer.Builder<>(null, null, null)
                .withLoggingEnabled(logConfig);

        assertThat(builder.logConfig(), is(singletonMap("min.insync.replicas", expect)));
        assertThat(builder.loggingEnabled(), is(true));
    }