AlanConfluent commented on code in PR #29657:
URL: https://github.com/apache/flink/pull/29657#discussion_r4404115861
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/WritableInternalTimeContext.java:
##########
@@ -111,11 +116,23 @@ private void deleteUnnamedTimer(long time) {
}
private void clearUnnamedTimers() {
+ final RowData currentKey = keyedStateBackend.getCurrentKey();
final List<Long> unnamedTimers = new ArrayList<>();
try {
- unnamedTimerService.forEachEventTimeTimer((namespace, time) ->
unnamedTimers.add(time));
+ // The timer service iterates the timers of all keys and sets the
key context to the
+ // key of each visited timer. Only timers of the current key must
be cleared.
+ // The equals is safe: keys are BinaryRowData, which compares the
serialized bytes and
+ // not the object identity, so a key deserialized by the timer
service still matches.
+ unnamedTimerService.forEachEventTimeTimer(
Review Comment:
Yeah, that's fair. I think you're right about the tradeoff. It would be a
significant cost elsewhere, so not worth it.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
