private static Stream<Arguments> exceptionalData() {
        return Stream.of(
                Arguments.of(new TestCase("truncate all, single record", 15, new TestRecord(null, 12))),
                Arguments.of(new TestCase("truncate all, single record, exact timestamp match", 15, new TestRecord(null, INSERT_VALID_FROM_TIMESTAMP))),
                Arguments.of(new TestCase("truncate all, multiple records", 15, new TestRecord(null, 12), new TestRecord("foo".getBytes(), 11))),
                Arguments.of(new TestCase("truncate all, multiple records, exact timestamp match", 15, new TestRecord(null, 12), new TestRecord("foo".getBytes(), 11), new TestRecord(null, INSERT_VALID_FROM_TIMESTAMP))),
                Arguments.of(new TestCase("partial truncation, single record", 15, new TestRecord(null, 8))),
                Arguments.of(new TestCase("partial truncation, multiple records", 15, new TestRecord("foo".getBytes(), 12), new TestRecord("bar".getBytes(), 8))),
                Arguments.of(new TestCase("partial truncation, on record boundary", 15, new TestRecord("foo".getBytes(), 12), new TestRecord("bar".getBytes(), INSERT_VALID_FROM_TIMESTAMP), new TestRecord("baz".getBytes(), 8)))
        );
    }
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