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

        // test updating at each possible index
        for (int updateIdx = 0; updateIdx < testCase.records.size(); updateIdx++) {
            // build updated record
            long updatedRecordTimestamp = testCase.records.get(updateIdx).timestamp - 1;
            if (updatedRecordTimestamp < 0 || (updateIdx < testCase.records.size() - 1 && updatedRecordTimestamp == testCase.records.get(updateIdx + 1).timestamp)) {
                // found timestamp conflict. try again
                updatedRecordTimestamp = testCase.records.get(updateIdx).timestamp + 1;
                if (updateIdx > 0 && updatedRecordTimestamp == testCase.records.get(updateIdx - 1).timestamp) {
                    // found timestamp conflict. use original timestamp
                    updatedRecordTimestamp = testCase.records.get(updateIdx).timestamp;
                }
            }
            final TestRecord updatedRecord = new TestRecord("updated".getBytes(), updatedRecordTimestamp);

            final SegmentValue segmentValue = buildSegmentWithInsertLatest(testCase);

            // updateRecord() first requires a call to find()
            segmentValue.find(testCase.records.get(updateIdx).timestamp, false);
            segmentValue.updateRecord(updatedRecord.timestamp, updatedRecord.value, updateIdx);

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

            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