@Override
    public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
        wrapped().putAll(entries);
        for (final KeyValue<Bytes, byte[]> entry : entries) {
            final byte[] valueAndTimestamp = entry.value;
            log(entry.key, rawValue(valueAndTimestamp), valueAndTimestamp == null ? context.timestamp() : timestamp(valueAndTimestamp));
        }
    }
@Test
    public void shouldLogChangesOnPutAll() {
        store.putAll(Arrays.asList(KeyValue.pair(hi, rawThere),
                                   KeyValue.pair(hello, rawWorld)));

        assertThat(collector.collected().size(), equalTo(2));
        assertThat(collector.collected().get(0).key(), equalTo(hi));
        assertThat(collector.collected().get(0).value(), equalTo(there.value()));
        assertThat(collector.collected().get(0).timestamp(), equalTo(there.timestamp()));
        assertThat(collector.collected().get(1).key(), equalTo(hello));
        assertThat(collector.collected().get(1).value(), equalTo(world.value()));
        assertThat(collector.collected().get(1).timestamp(), equalTo(world.timestamp()));
    }