private <T extends VersionedStoreSegment> long finishPut(
            final VersionedStoreClient<T> versionedStoreClient,
            final long observedStreamTime,
            final Bytes key,
            final byte[] value,
            final long timestamp,
            final long foundTs
    ) {
        if (foundTs == SENTINEL_TIMESTAMP) {
            // insert into latest value store
            if (value != null) {
                versionedStoreClient.putLatestValue(key, LatestValueFormatter.from(value, timestamp));
            } else {
                // tombstones are not inserted into the latest value store. insert into segment instead.
                // the specific segment to insert to is determined based on the tombstone's timestamp
                final T segment = versionedStoreClient.getOrCreateSegmentIfLive(
                        versionedStoreClient.segmentIdForTimestamp(timestamp), context, observedStreamTime);
                if (segment == null) {
                    // the record being inserted does not affect version history. discard and return.
                    // this can happen during restore because individual put calls are executed after
                    // observedStreamTime is fast-forwarded for the entire batch of records being
                    // restored at once.
                    return PUT_RETURN_CODE_NOT_PUT;
                }

                final byte[] rawSegmentValue = segment.get(key);
                if (rawSegmentValue == null) {
                    // in this special case where the latest record version (for a particular key)
                    // is a tombstone, and the segment that the tombstone belongs in contains no
                    // record versions for this key, create a new "degenerate" segment with the
                    // tombstone's timestamp as both validFrom and validTo timestamps for the segment
                    segment.put(
                            key,
                            RocksDBVersionedStoreSegmentValueFormatter
                                    .newSegmentValueWithRecord(null, timestamp, timestamp)
                                    .serialize()
                    );
                } else {
                    // insert as latest, since foundTs = sentinel means nothing later exists
                    if (RocksDBVersionedStoreSegmentValueFormatter.getNextTimestamp(rawSegmentValue) == timestamp) {
                        // next timestamp equal to put() timestamp already represents a tombstone,
                        // so no additional insertion is needed in this case
                        return foundTs;
                    }
                    final SegmentValue segmentValue
                            = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawSegmentValue);
                    segmentValue.insertAsLatest(
                            RocksDBVersionedStoreSegmentValueFormatter.getNextTimestamp(rawSegmentValue),
                            timestamp,
                            null
                    );
                    segment.put(key, segmentValue.serialize());
                }
            }
        } else {
            // insert into segment corresponding to foundTs, as foundTs represents the validTo
            // timestamp of the current put.
            // the new record is either the earliest or the latest in this segment, depending on the
            // circumstances of the fall-through. (it cannot belong in the middle because otherwise
            // putSegments() above would have identified a segment for which
            // minTimestamp <= timestamp < nextTimestamp, and putSegments would've completed the
            // put procedure without reaching this fall-through case.)
            final T segment = versionedStoreClient.getOrCreateSegmentIfLive(
                    versionedStoreClient.segmentIdForTimestamp(foundTs), context, observedStreamTime);
            if (segment == null) {
                // the record being inserted does not affect version history. discard and return.
                // this can happen during restore because individual put calls are executed after
                // observedStreamTime is fast-forwarded for the entire batch of records being
                // restored at once.
                return PUT_RETURN_CODE_NOT_PUT;
            }

            final byte[] rawSegmentValue = segment.get(key);
            if (rawSegmentValue == null) {
                segment.put(
                        key,
                        RocksDBVersionedStoreSegmentValueFormatter
                                .newSegmentValueWithRecord(value, timestamp, foundTs)
                                .serialize()
                );
            } else {
                final long foundNextTs = RocksDBVersionedStoreSegmentValueFormatter.getNextTimestamp(rawSegmentValue);
                if (foundNextTs <= timestamp) {
                    // insert as latest. this case is possible if the found segment is "degenerate"
                    // (cf RocksDBVersionedStoreSegmentValueFormatter.java for details) as older
                    // degenerate segments may result in "gaps" between one record version's validTo
                    // timestamp and the next record version's validFrom timestamp
                    final SegmentValue segmentValue = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawSegmentValue);
                    segmentValue.insertAsLatest(timestamp, foundTs, value);
                    segment.put(key, segmentValue.serialize());
                } else {
                    // insert as earliest
                    final SegmentValue segmentValue = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawSegmentValue);
                    segmentValue.insertAsEarliest(timestamp, value);
                    segment.put(key, segmentValue.serialize());
                }
            }
        }
        return foundTs;
    }
@return updated {@code foundTs} value, i.e., the validTo timestamp of the record being put
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java