private <T extends VersionedStoreSegment> long doPut(
            final VersionedStoreClient<T> versionedStoreClient,
            final long observedStreamTime,
            final Bytes key,
            final byte[] value,
            final long timestamp
    ) {
        segmentStores.cleanupExpiredSegments(observedStreamTime);

        // track the smallest timestamp seen so far that is larger than insertion timestamp.
        // this timestamp determines, based on all segments searched so far, which segment the
        // new record should be inserted into.
        long foundTs;

        // check latest value store
        PutStatus status = maybePutToLatestValueStore(
                versionedStoreClient,
                observedStreamTime,
                key,
                value,
                timestamp
        );
        if (status.isComplete) {
            return status.foundTs == SENTINEL_TIMESTAMP ? PUT_RETURN_CODE_VALID_TO_UNDEFINED : status.foundTs;
        } else {
            foundTs = status.foundTs;
        }

        // continue search in segments
        status = maybePutToSegments(
                versionedStoreClient,
                observedStreamTime,
                key,
                value,
                timestamp,
                foundTs
        );
        if (status.isComplete) {
            return status.foundTs == SENTINEL_TIMESTAMP ? PUT_RETURN_CODE_VALID_TO_UNDEFINED : status.foundTs;
        } else {
            foundTs = status.foundTs;
        }

        // the record did not unconditionally belong in any specific store (latest value store
        // or segments store). insert based on foundTs here instead.
        foundTs = finishPut(
                versionedStoreClient,
                observedStreamTime,
                key,
                value,
                timestamp,
                foundTs
        );
        return foundTs == SENTINEL_TIMESTAMP ? PUT_RETURN_CODE_VALID_TO_UNDEFINED : foundTs;
    }
Helper method shared between put and restore.
<p>
This method does not check whether the record being put is expired based on grace period
or not; that is the caller's responsibility. This method does, however, check whether the
record is expired based on history retention, by using the current
{@code observedStreamTime}, and returns without inserting into the store if so. It can be
possible that a record is not expired based on grace period but is expired based on
history retention, even though history retention is always at least the grace period,
during restore because restore advances {@code observedStreamTime} to the largest timestamp
in the entire restore batch at the beginning of restore, in order to optimize for not
putting records into the store which will have expired by the end of the restore.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java