@ParameterizedTest
    @MethodSource("nonExceptionalData")
    public void shouldInsertAtIndex(final TestCase testCase) {
        if (testCase.isDegenerate) {
            // cannot insert into degenerate segment
            return;
        }

        // test inserting at each possible index
        for (int insertIdx = 0; insertIdx <= testCase.records.size() - 1; insertIdx++) {
            // build record to insert
            final long newRecordTimestamp;
            if (insertIdx == 0) {
                newRecordTimestamp = testCase.records.get(0).timestamp + 1;
                if (newRecordTimestamp == testCase.nextTimestamp) {
                    // cannot insert because no timestamp exists between last record and nextTimestamp
                    continue;
                }
            } else {
                newRecordTimestamp = testCase.records.get(insertIdx - 1).timestamp - 1;
                if (newRecordTimestamp < 0 || (newRecordTimestamp == testCase.records.get(insertIdx).timestamp)) {
                    // cannot insert because timestamps of existing records are adjacent
                    continue;
                }
            }
            final TestRecord newRecord = new TestRecord("new".getBytes(), newRecordTimestamp);

            final SegmentValue segmentValue = buildSegmentWithInsertLatest(testCase);

            // insert() first requires a call to find()
            segmentValue.find(testCase.records.get(insertIdx).timestamp, false);

            segmentValue.insert(newRecord.timestamp, newRecord.value, insertIdx);

            // create expected results
            final List<TestRecord> expectedRecords = new ArrayList<>(testCase.records);
            expectedRecords.add(insertIdx, newRecord);

            verifySegmentContents(segmentValue, new TestCase("expected", testCase.nextTimestamp, expectedRecords));
        }
    }
These scenarios may only be hit in the event of an earlier exception, such as failure to
write to a particular segment store of {@link RocksDBVersionedStore} resulting in an
inconsistency among segments. These tests verify that the store is able to recover
gracefully even if this happens.
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreSegmentValueFormatterTest.java