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);
            }
        }

    }
A persistent, versioned key-value store based on RocksDB.
<p>
This store implementation consists of a "latest value store" and "segment stores." The latest
record version for each key is stored in the latest value store, while older record versions
are stored in the segment stores. Conceptually, each record version has two associated
timestamps:
<ul>
    <li>a {@code validFrom} timestamp. This timestamp is explicitly associated with the record
    as part of the {@link VersionedKeyValueStore#put(Object, Object, long)}} call to the store;
    i.e., this is the record's timestamp.</li>
    <li>a {@code validTo} timestamp. This is the timestamp of the next record (or deletion)
    associated with the same key, and is implicitly associated with the record. This timestamp
    can change as new records are inserted into the store.</li>
</ul>
The validity interval of a record is from validFrom (inclusive) to validTo (exclusive), and
can change as new record versions are inserted into the store (and validTo changes as a result).
<p>
Old record versions are stored in segment stores according to their validTo timestamps. This
allows for efficient expiry of old record versions, as entire segments can be dropped from the
store at a time, once the records contained in the segment are no longer relevant based on the
store's history retention (for an explanation of "history retention", see
{@link VersionedKeyValueStore}). Multiple record versions (for the same key) within a single
segment are stored together using the format specified in {@link RocksDBVersionedStoreSegmentValueFormatter}.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java