KAFKA-13499: Avoid restoring outdated records

[gabriellefu]

1. Expose the retentionPeriod length to storeMetadata
2. In prepareChangelogs(), switch it from always seektobeginning if
   checkpoint doesn't exist to seek to certain timestamp to avoid restoring
   outdated records.
3. Change from the :
   Instead of the wall clock, use the latest timestamp in the changelog as
   the latest time, and seek from the timestamp of
   latest_changelog_stamp_time-rention_period.

Reviewers: TengYao Chi frankvicky@apache.org, Bill Bejeck
bbejeck@apache.org

[gabriellefu]

The failed smoke test now is passed:

[frankvicky]

@gabriellefu Thanks for the PR.
Please run ./gradlew clean spotlessApply to fix the CI fail.

[bbejeck]

Thanks @gabriellefu I made a pass

[bbejeck]

@gabriellefu I was playing around some more and I think I found something else - New standby tasks won't have a valid endOffset, so they need to be filtered out. Otherwise with the restore consumer's auto.offset.reset=none every batched partition falls back to seek-to-beginning.

So we can update the if block to this
if (retentionPeriod > 0 && retentionPeriod != Long.MAX_VALUE && endOffset != null && endOffset > 0)

[gabriellefu]

I used restoreConsumer.endOffsets() not which should be able to solve the standby task problem

[bbejeck]

System tests pass for this PR

[bbejeck]

Thanks @gabriellefu ! LGTM

[bbejeck]

Merged #22115 into trunk