private void truncateRecordsToTimestamp(final long timestamp) {
            if (timestamp <= minTimestamp) {
                // all record versions in this segment will be truncated.
                // if the total number of record versions is not one, or if the truncation timestamp
                // does not match the minTimestamp of this segment, then this is unexpected (under
                // normal, deterministic record processing post-recovery) and we log an extra warning
                final int totalRecords = find(minTimestamp, false).index() + 1;
                if (!((timestamp == minTimestamp) && (totalRecords == 1))) {
                    LOG.warn("The versioned store inconsistency affects more than "
                            + "one record version, even though under normal replay operations only one "
                            + "record should be affected. Full records affected: {} (expected: 1). "
                            + "New record timestamp: {} (expected: {}).",
                        totalRecords,
                        timestamp,
                        unpackedReversedTimestampAndValueSizes.get(0).timestamp);
                }

                // delete everything in this current segment by replacing it with a degenerate segment
                initializeWithRecord(new ValueAndValueSize(null), timestamp, timestamp);
                return;
            }

            final SegmentSearchResult searchResult = find(timestamp, false);
            // all records with later timestamps should be removed
            int fullRecordsToTruncate = searchResult.index();
            // additionally remove the current record as well, if its validFrom equals the
            // timestamp to truncate to
            if (searchResult.validFrom() == timestamp) {
                fullRecordsToTruncate++;
            }

            // if the total number of record versions is not one, or if the truncation timestamp
            // does not match the timestamp of the latest record in this segment, then this is
            // unexpected (under normal, deterministic record processing post-recovery) and we log
            // an extra warning
            if (!((fullRecordsToTruncate == 1) && (searchResult.index == 0))) {
                LOG.warn("The versioned store inconsistency affects more (or less) than "
                    + "one record version, even though under normal replay operations only one "
                    + "record should be affected. Full records affected: {} (expected: 1). "
                    + "New record timestamp: {} (expected: {}).",
                    fullRecordsToTruncate,
                    timestamp,
                    unpackedReversedTimestampAndValueSizes.get(0).timestamp);
            }

            if (fullRecordsToTruncate == 0) {
                // no records to remove; update nextTimestamp and return
                nextTimestamp = timestamp;
                ByteBuffer.wrap(segmentValue, 0, TIMESTAMP_SIZE).putLong(0, timestamp);
                return;
            }

            final int valuesLengthToRemove = cumulativeValueSizes.get(fullRecordsToTruncate - 1);
            final int timestampAndValueSizesLengthToRemove = (TIMESTAMP_SIZE + VALUE_SIZE) * fullRecordsToTruncate;
            final int newSegmentLength = segmentValue.length - valuesLengthToRemove - timestampAndValueSizesLengthToRemove;
            segmentValue = ByteBuffer.allocate(newSegmentLength)
                .putLong(timestamp) // update nextTimestamp as part of truncation
                .putLong(minTimestamp)
                .put(segmentValue, 2 * TIMESTAMP_SIZE + timestampAndValueSizesLengthToRemove, newSegmentLength - 2 * TIMESTAMP_SIZE)
                .array();
            nextTimestamp = timestamp;
            resetDeserHelpers();
        }
Delete all record versions newer than (or at) the provided timestamp, and also truncate
any partial records which extend beyond the provided timestamp. In other words, when
this method returns, nextTimestamp will be equal to the truncation timestamp.

@param timestamp timestamp to truncate to
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreSegmentValueFormatter.java