AlanConfluent commented on code in PR #29657:
URL: https://github.com/apache/flink/pull/29657#discussion_r4402904102
##########
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:
This scans every timer, including for other keys, to do this. If you also
kept `MapState<Long, Void>` to essentially create a set of unnamed timers, or
something like that, you could just iterate through all the timers for this
particular key.
Given that this would require a migration for existing PTFs without that
state, it's probably not worth the trouble for an uncommon call. But just a
thought.
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/process/ProcessSetTableOperatorTest.java:
##########
@@ -128,6 +129,20 @@ void testWithoutNotifyStatefulSets() throws Exception {
}
}
+ @Test
+ void testNonBinaryKey() throws Exception {
+ try (KeyedMultiInputStreamOperatorTestHarness<RowData, RowData>
harness =
+ createTtlBroadcastHarness(new
NotifyStatefulSetsOperatorCreator(false, 0))) {
Review Comment:
Nit: Can you just call createHarness since this doesn't touch broadcast
state?
--
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]
