@Override
        public void insertAsLatest(final long validFrom, final long validTo, final byte[] valueOrNull) {
            final ValueAndValueSize value = new ValueAndValueSize(valueOrNull);

            if (validFrom < nextTimestamp) {
                // detected inconsistency edge case due to partial update, (i.e., a previous put()
                // call required updating two different segments (or one segment and the latest
                // value store) and the write to the older segment was successful whereas the write
                // to the newer segment (or latest value store) failed. when this happened, a fatal
                // error was thrown. streams has now restarted and we need to reconcile the store
                // inconsistency caused by the partial update.
                //
                // because record versions are inserted into segments based on their validTo
                // timestamps, the only situation in which two segments need to be updated for a
                // single put() is when an existing record moves from a newer segment into an
                // older segment (because its validTo timestamp is updated by the put record).
                // when this happens, we first move the existing record from the newer to the
                // older segment and afterwards update the newer segment with the new record.
                // if only the first write (i.e., moving the existing record into the older segment)
                // succeeded, then we have an overlap in [minTimestamp, nextTimestamp) ranges for
                // the two segments, and this older segment is corrupted:
                //  - the latest record of this older segment has [old_value, old_validFrom, updated_validTo),
                //    which is incorrect because the dual write was not completed
                //  - the earliest record of the newer segment has [old_value, old_validFrom, old_validTo),
                //    which is still correct as it was never modified
                //
                // we have the older segment at hand, and need to truncate the partial write.
                // do this by removing the latest entries from this segment until the overlap is resolved.
                LOG.warn("Detected inconsistency among versioned store segments. "
                    + "This indicates a previous failure to write to a state store. "
                    + "Automatically recovering and continuing.");
                truncateRecordsToTimestamp(validFrom);
            }

            if (nextTimestamp == validFrom) {
                // nextTimestamp is moved into segment automatically as record is added on top
                if (isDegenerate) {
                    initializeWithRecord(value, validFrom, validTo);
                } else {
                    doInsert(validFrom, value, 0);
                }
            } else {
                // move nextTimestamp into list as tombstone and add new record on top
                if (isDegenerate) {
                    initializeWithRecord(new ValueAndValueSize(null), nextTimestamp, validFrom);
                } else {
                    doInsert(nextTimestamp, new ValueAndValueSize(null), 0);
                }
                doInsert(validFrom, value, 0);
            }
            // update nextTimestamp
            nextTimestamp = validTo;
            ByteBuffer.wrap(segmentValue, 0, TIMESTAMP_SIZE).putLong(nextTimestamp);
        }
This value will be null if the caller did not specify that the value should be
included with the return result from {@link SegmentValue#find(long, boolean)}.
In this case, it is up to the caller to not call this method, as the "value"
returned will not be meaningful.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreSegmentValueFormatter.java