vcrfxia commented on code in PR #13243: URL: https://github.com/apache/kafka/pull/13243#discussion_r1107567122
########## streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java: ########## @@ -283,7 +293,12 @@ public void init(final StateStoreContext context, final StateStore root) { // VisibleForTesting void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> records) { - // advance stream time to the max timestamp in the batch + // copy the observed stream time, for use in deciding whether to drop records during restore, + // when records have exceeded the store's grace period. + long streamTimeForRestore = observedStreamTime; Review Comment: > I guess the question is, what is the value of `observedStreamTime` when we start the restore? Are you saying it's `-1` and we basically "replay" `observedStreamTime` during restore? Yes, that's exactly right. `observedStreamTime` is tracked locally per store. It is initialized to `-1` and only updated on `put()` or during restore. (This is the same as the existing behavior for window stores today.) > Maybe best to update some variable names? Are you proposing that `doPut()` takes stream time as a parameter, so that during normal `put()` operation we pass `observedStreamTime` and during restore we pass `endOfBatchStreamTime`, which means we can rename `streamTimeForRestore` to be `observedStreamTime` instead? This SGTM, just want to check whether that's also what you have in mind, since we removed a number of parameters from `doPut()` in a previous PR revision in order to keep the parameter list small. > I guess follow up work (independent for this KIP) might be, to actually make use of KS runtime streamTime instead of tracking inside the store, and thus won't need `observedStreamTime` any longer, as we could look ahead to the "end-of-restore stream-time" (not just "end-of batch"). What's the scope of the "streamTime" which is tracked by the KS runtime? Is it per-task? Per-processor? Global? I'm wondering how this would work in situations with multiple partitions, or with multiple processors where some processors are expected to see new data earlier than other (downstream) processors. I guess we'd also need to implement the change from your other comment about not writing records which are expired (based on grace period) into the changelog topic first before we can make this change, otherwise we would not have a way to determine during restore whether records are expired or not. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org