hanyuzheng7 commented on code in PR #29658:
URL: https://github.com/apache/flink/pull/29658#discussion_r4407662367
##########
flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/AbstractPythonFunctionOperator.java:
##########
@@ -219,6 +219,12 @@ public void setCurrentKey(Object key) {
super.setCurrentKey(key);
}
+ @Override
+ public void moveToNextKey(Object key) {
+ processElementsOfCurrentKeyIfNeeded(key);
Review Comment:
For PyFlink in BATCH, timers that Python registers while this flushes the
previous key's bundle are queued as mailbox mails
(`BeamPythonFunctionRunner.createTimerReceiverFactory`) and only applied after
the key has moved on. `TimerRegistration.setTimer` then calls
`keyContext.setCurrentKey`(`previousKey`), which I think would now throw in
`BatchExecutionKeyedStateBackend.setCurrentKey`.
`processElementsOfCurrentKeyIfNeeded` doesn't call `drainUnregisteredTimers()`
the way advanceWatermark does. Would draining them after `invokeFinishBundle()`
help here? I only traced the code and haven't run a PyFlink batch job with
timers, so please correct me if I'm missing something.
--
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]
