private <T extends VersionedStoreSegment> long putToSegment(
            final VersionedStoreClient<T> versionedStoreClient,
            final long observedStreamTime,
            final T segment,
            final byte[] rawSegmentValue,
            final Bytes key,
            final byte[] value,
            final long timestamp
    ) {
        final long segmentIdForTimestamp = versionedStoreClient.segmentIdForTimestamp(timestamp);
        // it's possible that putting the current record into a segment will require moving an
        // existing record from this segment into an older segment. this is because records belong
        // in segments based on their validTo timestamps, and putting a new record into a segment
        // updates the validTo timestamp for the previous record. the new validTo timestamp
        // of the previous record will be the (validFrom) timestamp of the record being put.
        // if this (validFrom) timestamp belongs in a different segment, then it may be required
        // to move an existing record from this segment into an older one.
        final boolean writeToOlderSegmentMaybeNeeded = segmentIdForTimestamp != segment.id();

        final SegmentValue segmentValue = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawSegmentValue);
        // we pass `writeToOlderSegmentMaybeNeeded` as `includeValue` in the call to find() below
        // because moving a record from this segment into an older one will require knowing its
        // value. unless such a move is required, we do not need the existing record value.
        final SegmentSearchResult searchResult = segmentValue.find(timestamp, writeToOlderSegmentMaybeNeeded);

        if (searchResult.validFrom() == timestamp) {
            // this put() replaces an existing entry, rather than adding a new one. as a result,
            // the validTo timestamp of the previous record does not need to be updated either,
            // and we do not need to move an existing record to an older segment (i.e., we can
            // ignore `writeToOlderSegmentMaybeNeeded`).
            segmentValue.updateRecord(timestamp, value, searchResult.index());
            segment.put(key, segmentValue.serialize());
            return searchResult.validTo();
        }

        if (writeToOlderSegmentMaybeNeeded) {
            // existing record needs to be moved to an older segment.
            // it's important that this step happens before updating the current
            // segment. if there is a partial failure (this step succeeds but the
            // update to the current segment fails), updating the older 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 T olderSegment = versionedStoreClient
                    .getOrCreateSegmentIfLive(segmentIdForTimestamp, context, observedStreamTime);
            // `olderSegment == null` implies that all data in the older 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 newer segment
            // into the older segment if `olderSegment != 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 newer segment is
            // expired.) so, there is nothing to do for this step if `olderSegment == null`,
            // but we do still update the newer segment with the new record below.
            if (olderSegment != null) {
                final byte[] rawOlderSegmentValue = olderSegment.get(key);
                if (rawOlderSegmentValue == null) {
                    olderSegment.put(
                            key,
                            RocksDBVersionedStoreSegmentValueFormatter.newSegmentValueWithRecord(
                                    searchResult.value(), searchResult.validFrom(), timestamp
                            ).serialize()
                    );
                } else {
                    final SegmentValue olderSegmentValue
                            = RocksDBVersionedStoreSegmentValueFormatter.deserialize(rawOlderSegmentValue);
                    olderSegmentValue.insertAsLatest(searchResult.validFrom(), timestamp, searchResult.value());
                    olderSegment.put(key, olderSegmentValue.serialize());
                }
            }

            // update in newer segment (replace the record that was just moved with the new one)
            segmentValue.updateRecord(timestamp, value, searchResult.index());
            segment.put(key, segmentValue.serialize());
            return searchResult.validTo();
        }

        // plain insert into segment. no additional handling required.
        segmentValue.insert(timestamp, value, searchResult.index());
        segment.put(key, segmentValue.serialize());
        return searchResult.validTo();
    }
@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