KAFKA-20876: Retry windowed restore after control record

[lh0156]

Summary

When a windowed changelog ends with a transaction control record, the
restore consumer filters that record from poll results. Probing only
endOffset - 1 therefore returns no record and incorrectly abandons the
retention-based restore optimization.

This change retries the probe from the preceding offset before falling
back to the beginning. The regression test covers both a visible latest
record and the filtered control-record case.

Testing

• TDD RED: the new filtered-record case failed on the baseline branch
  because restoration fell back to the beginning.
• TDD GREEN: ./gradlew :streams:test --tests org.apache.kafka.streams.processor.internals.StoreChangelogReaderTest.shouldSeekByTimestampForWindowedStoreWithoutCheckpoint --no-build-cache --console=plain
• Full module: ./gradlew :streams:test --no-build-cache --console=plain (BUILD SUCCESSFUL, 6m 54s)
• The targeted Gradle run completed the affected streams checkstyle and
  SpotBugs tasks successfully.
• git diff --check

Jira: https://issues.apache.org/jira/browse/KAFKA-20876 Reviewers: Mingi
Cho (github:ChoMinGi)

[lh0156]

Verification update: the full :streams:test suite passed locally in 6m 54s after the targeted RED/GREEN regression test and affected checkstyle/SpotBugs tasks also passed. The branch remains clean and is ready for maintainer review.

[ChoMinGi]

Thanks for working on this. I tried the probe sequence with EmbeddedKafkaCluster and a real read_committed consumer using pollTime = 100 ms. I saw two cases worth considering.

First, the retry still returned empty:

Committed transaction: 5 records + COMMIT marker, LSO=6

seek(5) -> 0 records,   2 ms
seek(4) -> 0 records, 101 ms   <- retry
...
record returned on poll #6, 508 ms in

Control: fresh consumer, seek(4) -> 1 record in 2 ms

Second, the endOffset - 2 retry did not reach the latest visible record when the log ended with an aborted transaction:

Committed transaction followed by an aborted transaction, LSO=12

seek(11) -> 0
seek(10) -> 0   <- retry
seek(9..5) -> 0
seek(4)  -> 1

Here, the latest visible record was eight offsets back because the aborted batch was filtered as a whole.

Would it be better to start the probe from an earlier offset, clamped to logStartOffset, instead of retrying at endOffset - 2?

I can share the reproduction test if needed.