void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> records) {

        // compute the observed stream time at the end of the restore batch, in order to speed up
        // restore by not bothering to read from/write to segments which will have expired by the
        // time the restoration process is complete.
        long endOfBatchStreamTime = observedStreamTime;
        for (final ConsumerRecord<byte[], byte[]> record : records) {
            endOfBatchStreamTime = Math.max(endOfBatchStreamTime, record.timestamp());
        }

        final VersionedStoreClient<?> restoreClient = restoreWriteBuffer.getClient();

        // note: there is increased risk for hitting an out-of-memory during this restore loop,
        // compared to for non-versioned key-value stores, because this versioned store
        // implementation stores multiple records (for the same key) together in a single RocksDB
        // "segment" entry -- restoring a single changelog entry could require loading multiple
        // records into memory. how high this memory amplification will be is very much dependent
        // on the specific workload and the value of the "segment interval" parameter.
        synchronized (position) {
            for (final ConsumerRecord<byte[], byte[]> record : records) {
                if (record.timestamp() < observedStreamTime - gracePeriod) {
                    // record is older than grace period and was therefore never written to the store
                    continue;
                }
                // advance observed stream time as usual, for use in deciding whether records have
                // exceeded the store's grace period and should be dropped.
                observedStreamTime = Math.max(observedStreamTime, record.timestamp());

                ChangelogRecordDeserializationHelper.applyChecksAndUpdatePosition(
                    record,
                    consistencyEnabled,
                    position
                );

                // put records to write buffer
                doPut(
                    restoreClient,
                    endOfBatchStreamTime,
                    new Bytes(record.key()),
                    record.value(),
                    record.timestamp()
                );
            }

            try {
                restoreWriteBuffer.flush();
            } catch (final RocksDBException e) {
                throw new ProcessorStateException("Error restoring batch to store " + name, e);
            }
        }

    }
@Test
    public void shouldRestoreEvenIfRecordWouldBeExpiredByEndOfBatch() {
        final List<DataRecord> records = new ArrayList<>();
        records.add(new DataRecord("k2", "v2", HISTORY_RETENTION - GRACE_PERIOD)); // this record will be older than grace period by the end of the batch, but should still be restored
        records.add(new DataRecord("k", "v", HISTORY_RETENTION + 10));

        store.restoreBatch(getChangelogRecords(records));

        verifyGetValueFromStore("k2", "v2", HISTORY_RETENTION - GRACE_PERIOD);
        verifyGetValueFromStore("k", "v", HISTORY_RETENTION + 10);
    }