@Override
    Map<KeyValueSegment, WriteBatch> getWriteBatches(final Collection<ConsumerRecord<byte[], byte[]>> records) {
        final Map<KeyValueSegment, WriteBatch> writeBatchMap = new HashMap<>();
        for (final ConsumerRecord<byte[], byte[]> record : records) {
            final long timestamp = WindowKeySchema.extractStoreTimestamp(record.key());
            observedStreamTime = Math.max(observedStreamTime, timestamp);
            minTimestamp = Math.min(minTimestamp, timestamp);
            final long segmentId = segments.segmentId(timestamp);
            final KeyValueSegment segment = segments.getOrCreateSegmentIfLive(segmentId, context, observedStreamTime);
            if (segment != null) {
                //null segment is if it has expired, so  we don't want those records
                ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
                    record,
                    consistencyEnabled,
                    position
                );
                try {
                    final WriteBatch batch = writeBatchMap.computeIfAbsent(segment, s -> new WriteBatch());

                    final byte[] baseKey = TimeFirstWindowKeySchema.fromNonPrefixWindowKey(record.key());
                    segment.addToBatch(new KeyValue<>(baseKey, record.value()), batch);
                } catch (final RocksDBException e) {
                    throw new ProcessorStateException("Error restoring batch to store " + name(), e);
                }
            }
        }
        return writeBatchMap;
    }
@Test
    public void shouldCreateEmptyWriteBatches() {
        final Collection<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
        final Map<KeyValueSegment, WriteBatch> writeBatchMap = bytesStore.getWriteBatches(records);
        assertEquals(0, writeBatchMap.size());
    }