private <T extends VersionedStoreSegment> PutStatus maybePutToLatestValueStore(
            final VersionedStoreClient<T> versionedStoreClient,
            final long observedStreamTime,
            final Bytes key,
            final byte[] value,
            final long timestamp
    ) {
        // initialize with a starting "sentinel timestamp" which represents
        // that the segment should be inserted into the latest value store.
        long foundTs = SENTINEL_TIMESTAMP;

        final byte[] rawLatestValueAndTimestamp = versionedStoreClient.getLatestValue(key);
        if (rawLatestValueAndTimestamp != null) {
            final long latestValueStoreTimestamp = LatestValueFormatter.getTimestamp(rawLatestValueAndTimestamp);
            if (timestamp >= latestValueStoreTimestamp) {
                // new record belongs in the latest value store
                if (timestamp > latestValueStoreTimestamp) {
                    // move existing latest value into segment.
                    // it's important that this step happens before the update to the latest value
                    // store. if there is a partial failure (this step succeeds but the update to
                    // the latest value store fails), updating the segment first means there will
                    // not be data loss. (rather, there will be duplicated data which is fine as
                    // it can/will be reconciled later.)
                    final long segmentId = versionedStoreClient.segmentIdForTimestamp(timestamp);
                    final T segment = versionedStoreClient.getOrCreateSegmentIfLive(segmentId, context, observedStreamTime);
                    // `segment == null` implies that all data in the segment is older than the
                    // history retention of this store, and therefore does not need to tracked.
                    // as a result, we only need to move the existing record from the latest value
                    // store into a segment if `segment != null`. (also, we do not call the
                    // `expiredRecordSensor` in this case because the expired record sensor is only
                    // called when the record being put is itself expired, which is not the case
                    // here. only the existing record already present in the latest value store
                    // is expired.) so, there is nothing to do for this step if `segment == null`,
                    // but we do still update the latest value store with the new record below.
                    if (segment != null) {
                        final byte[] rawValueToMove = LatestValueFormatter.getValue(rawLatestValueAndTimestamp);
                        final byte[] rawSegmentValue = segment.get(key);
                        if (rawSegmentValue == null) {
                            segment.put(
                                    key,
                                    RocksDBVersionedStoreSegmentValueFormatter
                                            .newSegmentValueWithRecord(rawValueToMove, latestValueStoreTimestamp, timestamp)
                                            .serialize()
                            );
                        } else {
                            final SegmentValue segmentValue = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawSegmentValue);
                            segmentValue.insertAsLatest(latestValueStoreTimestamp, timestamp, rawValueToMove);
                            segment.put(key, segmentValue.serialize());
                        }
                    }
                }

                // update latest value store
                if (value != null) {
                    versionedStoreClient.putLatestValue(key, LatestValueFormatter.from(value, timestamp));
                } else {
                    versionedStoreClient.deleteLatestValue(key);
                }
                return new PutStatus(true, foundTs);
            } else {
                foundTs = latestValueStoreTimestamp;
            }
        }
        return new PutStatus(false, foundTs);
    }
The smallest timestamp seen so far, as part of this current put() operation, 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.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java