KAFKA-13499: Avoid restoring outdated records#21779
KAFKA-13499: Avoid restoring outdated records#21779gabriellefu wants to merge 1 commit intoapache:trunkfrom
Conversation
gabriellefu
force-pushed
the
restoring_window
branch
from
deb759b to
1c4b635
Compare
bbejeck
left a comment
bbejeck
left a comment
There was a problem hiding this comment.
Thanks for the PR @gabriellefu! Overall this looks good with a few comments.
| current = ((WrappedStateStore<?, ?, ?>) current).wrapped(); | ||
| } | ||
| // Now 'current' is the innermost store. Check what type it is. | ||
| if (current instanceof AbstractRocksDBSegmentedBytesStore) { |
There was a problem hiding this comment.
I thinking instead of the mulitple instanceof checks we could introduce an interface WithRetentionPeriod (the name is up for debate) containing the single method retentionPeriod and have the store types here implement it. Since all of the instances here are internal API this is possble without requiring a KIP. So the multiple checks would go to one
if (current instanceof WithRetentionPeriod) {
return ((WithRetentionPeriod) current).retentionPeriod();
}This would also provide the added benefit of automatically picking up any store needing/using a retention period.
| @@ -88,6 +88,9 @@ public class InMemorySessionStore implements SessionStore<Bytes, byte[]> { | |||
| this.metricScope = metricScope; | |||
| this.position = Position.emptyPosition(); | |||
| } | |||
| } | ||
|
|
||
| @Test | ||
| public void shouldSeekByTimestampForWindowedStoreWithoutCheckpoint() { |
There was a problem hiding this comment.
Maybe add tests for negative test cases, broker doesn't have info and returns null for offsetsForTimes or for the non-windowed store case
2 participants
checkpoint doesn't exist to seek to certain timestamp to avoid
restoring outdated records.
Reviewers: Bill Bejeck bbejeck@apache.org