private static void verifySegmentContents(final SegmentValue segmentValue, final TestCase expectedRecords) {
        if (!expectedRecords.isDegenerate) { // cannot find() on degenerate segment
            // verify expected records
            for (int recordIdx = 0; recordIdx < expectedRecords.records.size(); recordIdx++) {
                final TestRecord expectedRecord = expectedRecords.records.get(recordIdx);
                final long expectedValidTo = recordIdx == 0 ? expectedRecords.nextTimestamp : expectedRecords.records.get(recordIdx - 1).timestamp;

                final SegmentSearchResult result = segmentValue.find(expectedRecord.timestamp, true);

                assertThat(result.index(), equalTo(recordIdx));
                assertThat(result.value(), equalTo(expectedRecord.value));
                assertThat(result.validFrom(), equalTo(expectedRecord.timestamp));
                assertThat(result.validTo(), equalTo(expectedValidTo));
            }
        }

        // verify expected exceptions from timestamp out-of-bounds
        assertThrows(IllegalArgumentException.class, () -> segmentValue.find(expectedRecords.nextTimestamp, false));
        assertThrows(IllegalArgumentException.class, () -> segmentValue.find(expectedRecords.minTimestamp - 1, false));
    }
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