@Override
    public VersionedRecord<byte[]> get(final Bytes key, final long asOfTimestamp) {
        Objects.requireNonNull(key, "key cannot be null");
        validateStoreOpen();

        if (asOfTimestamp < observedStreamTime - historyRetention) {
            // history retention exceeded. we still check the latest value store in case the
            // latest record version satisfies the timestamp bound, in which case it should
            // still be returned (i.e., the latest record version per key never expires).
            final byte[] rawLatestValueAndTimestamp = latestValueStore.get(key);
            if (rawLatestValueAndTimestamp != null) {
                final long latestTimestamp = LatestValueFormatter.getTimestamp(rawLatestValueAndTimestamp);
                if (latestTimestamp <= asOfTimestamp) {
                    // latest value satisfies timestamp bound
                    return new VersionedRecord<>(
                            LatestValueFormatter.getValue(rawLatestValueAndTimestamp),
                            latestTimestamp
                    );
                }
            }

            // history retention has elapsed and the latest record version (if present) does
            // not satisfy the timestamp bound. return null for predictability, even if data
            // is still present in segments.
            LOG.warn("Returning null for expired get.");
            return null;
        }

        // first check the latest value store
        final byte[] rawLatestValueAndTimestamp = latestValueStore.get(key);
        if (rawLatestValueAndTimestamp != null) {
            final long latestTimestamp = LatestValueFormatter.getTimestamp(rawLatestValueAndTimestamp);
            if (latestTimestamp <= asOfTimestamp) {
                return new VersionedRecord<>(LatestValueFormatter.getValue(rawLatestValueAndTimestamp), latestTimestamp);
            }
        }

        // check segment stores
        final List<LogicalKeyValueSegment> segments = segmentStores.segments(asOfTimestamp, Long.MAX_VALUE, false);
        for (final LogicalKeyValueSegment segment : segments) {
            final byte[] rawSegmentValue = segment.get(key);
            if (rawSegmentValue != null) {
                final long nextTs = RocksDBVersionedStoreSegmentValueFormatter.getNextTimestamp(rawSegmentValue);
                if (nextTs <= asOfTimestamp) {
                    // this segment contains no data for the queried timestamp, so earlier segments
                    // cannot either
                    return null;
                }

                if (RocksDBVersionedStoreSegmentValueFormatter.getMinTimestamp(rawSegmentValue) > asOfTimestamp) {
                    // the segment only contains data for after the queried timestamp. skip and
                    // continue the search to earlier segments. as an optimization, this code
                    // could be updated to skip forward to the segment containing the minTimestamp
                    // in the if-condition above.
                    continue;
                }

                // the desired result is contained in this segment
                final SegmentSearchResult searchResult =
                        RocksDBVersionedStoreSegmentValueFormatter
                                .deserialize(rawSegmentValue)
                                .find(asOfTimestamp, true);
                if (searchResult.value() != null) {
                    return new VersionedRecord<>(searchResult.value(), searchResult.validFrom(), searchResult.validTo());
                } else {
                    return null;
                }
            }
        }

        // checked all segments and no results found
        return null;
    }
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