@Override
    public synchronized void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
        physicalStore.putAll(entries.stream()
            .map(kv -> new KeyValue<>(
                prefixKeyFormatter.addPrefix(kv.key),
                kv.value))
            .collect(Collectors.toList()));
    }
@Test
    public void shouldPutAll() {
        final List<KeyValue<Bytes, byte[]>> segment0Records = new ArrayList<>();
        segment0Records.add(new KeyValue<>(
            new Bytes(serializeBytes("shared")),
            serializeBytes("v1")));
        segment0Records.add(new KeyValue<>(
            new Bytes(serializeBytes("segment0_only")),
            serializeBytes("foo")));

        final List<KeyValue<Bytes, byte[]>> segment1Records = new ArrayList<>();
        segment1Records.add(new KeyValue<>(
            new Bytes(serializeBytes("shared")),
            serializeBytes("v2")));
        segment1Records.add(new KeyValue<>(
            new Bytes(serializeBytes("segment1_only")),
            serializeBytes("bar")));

        final List<KeyValue<Bytes, byte[]>> negativeSegmentRecords = new ArrayList<>();
        negativeSegmentRecords.add(new KeyValue<>(
            new Bytes(serializeBytes("shared")),
            serializeBytes("v3")));
        negativeSegmentRecords.add(new KeyValue<>(
            new Bytes(serializeBytes("negative_segment_only")),
            serializeBytes("baz")));

        segment0.putAll(segment0Records);
        segment1.putAll(segment1Records);
        negativeIdSegment.putAll(negativeSegmentRecords);

        assertEquals("v1", getAndDeserialize(segment0, "shared"));
        assertEquals("v2", getAndDeserialize(segment1, "shared"));
        assertEquals("v3", getAndDeserialize(negativeIdSegment, "shared"));

        assertEquals("foo", getAndDeserialize(segment0, "segment0_only"));
        assertNull(getAndDeserialize(segment1, "segment0_only"));
        assertNull(getAndDeserialize(negativeIdSegment, "segment0_only"));

        assertNull(getAndDeserialize(segment0, "segment1_only"));
        assertEquals("bar", getAndDeserialize(segment1, "segment1_only"));
        assertNull(getAndDeserialize(negativeIdSegment, "segment1_only"));

        assertNull(getAndDeserialize(segment0, "negative_segment_only"));
        assertNull(getAndDeserialize(segment1, "negative_segment_only"));
        assertEquals("baz", getAndDeserialize(negativeIdSegment, "negative_segment_only"));
    }