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

        final SegmentValue segmentValue = buildSegmentWithInsertLatest(testCase);

        // build expected mapping from timestamp -> record
        final Map<Long, Integer> expectedRecordIndices = new HashMap<>();
        // it's important that this for-loop iterates backwards through the record indices, so that
        // when adjacent records have adjacent timestamps, then the record with the later timestamp
        // (i.e., the earlier index) takes precedence
        for (int recordIdx = testCase.records.size() - 1; recordIdx >= 0; recordIdx--) {
            if (recordIdx < testCase.records.size() - 1) {
                expectedRecordIndices.put(testCase.records.get(recordIdx).timestamp - 1, recordIdx + 1);
            }
            if (recordIdx > 0) {
                expectedRecordIndices.put(testCase.records.get(recordIdx).timestamp + 1, recordIdx);
            }
            expectedRecordIndices.put(testCase.records.get(recordIdx).timestamp, recordIdx);
        }

        // verify results
        for (final Map.Entry<Long, Integer> entry : expectedRecordIndices.entrySet()) {
            final TestRecord expectedRecord = testCase.records.get(entry.getValue());
            final long expectedValidTo = entry.getValue() == 0 ? testCase.nextTimestamp : testCase.records.get(entry.getValue() - 1).timestamp;

            final SegmentSearchResult result = segmentValue.find(entry.getKey(), true);

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

        // verify exception when timestamp is out of range
        assertThrows(IllegalArgumentException.class, () -> segmentValue.find(testCase.nextTimestamp, false));
        assertThrows(IllegalArgumentException.class, () -> segmentValue.find(testCase.nextTimestamp + 1, false));
        assertThrows(IllegalArgumentException.class, () -> segmentValue.find(testCase.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