You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1fanwang (via GitHub)" <gi...@apache.org> on 2026/09/25 07:35:43 UTC
[PR] [FLINK-33404][python] Add default timer callbacks [flink]
1fanwang opened a new pull request, #29286:
URL: https://github.com/apache/flink/pull/29286
## What is the purpose of the change
This restores the default timer callback that the PyFlink `ProcessFunction` and `CoProcessFunction` docs already promise.
On keyed connected streams, the runtime routes timer firings through `on_timer`. Using a plain `CoProcessFunction` there currently fails with `AttributeError: 'MyCoProcessFunction' object has no attribute 'on_timer'` instead of behaving like the Java API's default no-op `onTimer`.
## Brief change log
- align the base PyFlink process-function APIs with the existing timer contract by making timer callbacks default to a no-op
- let keyed connected-stream timer firings fall through cleanly when a user relies on that default behavior
- cover the regression with a process-mode timer test
## Verifying this change
This change added tests and can be verified as follows:
- reproduced the pre-fix failure in `pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override`
- reran the same regression plus nearby process tests after the fix:
- `ProcessDataStreamStreamingTests::test_process_function`
- `ProcessDataStreamStreamingTests::test_keyed_co_process`
- `ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override`
<details>
<summary>Raw logs</summary>
Before:
```bash
git checkout HEAD^ -- \
flink-python/pyflink/datastream/functions.py \
flink-python/pyflink/fn_execution/datastream/embedded/process_function.py \
flink-python/pyflink/fn_execution/datastream/process/process_function.py
source .venv-flink/bin/activate
SITE_PYFLINK=$(python - <<'PY'
import pyflink
print(pyflink.__path__[0])
PY
)
export FLINK_HOME="$SITE_PYFLINK"
export PYTHONPATH="$PWD/flink-python"
cd flink-python
pytest -q -vv pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override
```
```text
pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override FAILED [100%]
...
E AttributeError: 'MyCoProcessFunction' object has no attribute 'on_timer'
...
FAILED pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override
```
After:
```bash
git checkout HEAD -- \
flink-python/pyflink/datastream/functions.py \
flink-python/pyflink/fn_execution/datastream/embedded/process_function.py \
flink-python/pyflink/fn_execution/datastream/process/process_function.py
source .venv-flink/bin/activate
SITE_PYFLINK=$(python - <<'PY'
import pyflink
print(pyflink.__path__[0])
PY
)
export FLINK_HOME="$SITE_PYFLINK"
export PYTHONPATH="$PWD/flink-python"
cd flink-python
pytest -q \
pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_process_function \
pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process \
pyflink/datastream/tests/test_data_stream.py::ProcessDataStreamStreamingTests::test_keyed_co_process_without_on_timer_override
```
```text
... [100%]
3 passed in 21.73s
```
</details>
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: GitHub Copilot CLI (Claude Opus 5.5)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33404][python] Add default timer callbacks [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29286:
URL: https://github.com/apache/flink/pull/29286#issuecomment-5828781336
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "71440893d424a93be6f20bfd43007d2c31f0333f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "71440893d424a93be6f20bfd43007d2c31f0333f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 71440893d424a93be6f20bfd43007d2c31f0333f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33404][python] Fix CoProcessFunction timers without on_timer [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29286:
URL: https://github.com/apache/flink/pull/29286#discussion_r4310839629
##########
flink-python/pyflink/fn_execution/datastream/embedded/operations.py:
##########
@@ -211,7 +213,8 @@ def on_timer_func(timestamp):
yield from process_func(on_timer(timestamp, timer_context))
return TwoInputOperation(
- open_func, close_func, process_element_func1, process_element_func2, on_timer_func)
+ open_func, close_func, process_element_func1, process_element_func2,
+ on_timer_func if on_timer else None)
Review Comment:
Could we revert this case and handle the missing callback inside `on_timer_func`?
For example, the new test registers an event-time timer at 3 without defining `on_timer`. In thread mode, when the watermark reaches 3, `TwoInputOperation.on_timer()` returns `None`, which becomes Java `null`. `EmbeddedPythonIterator.from(null)` then throws an NPE. We still need an empty generator when the user callback is absent.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33404][python] Fix CoProcessFunction timers without on_timer [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29286:
URL: https://github.com/apache/flink/pull/29286#discussion_r4310839629
##########
flink-python/pyflink/fn_execution/datastream/embedded/operations.py:
##########
@@ -211,7 +213,8 @@ def on_timer_func(timestamp):
yield from process_func(on_timer(timestamp, timer_context))
return TwoInputOperation(
- open_func, close_func, process_element_func1, process_element_func2, on_timer_func)
+ open_func, close_func, process_element_func1, process_element_func2,
+ on_timer_func if on_timer else None)
Review Comment:
Could we revert this change and handle the missing callback inside `on_timer_func`?
For example, the new test registers an event-time timer at 3 without defining `on_timer`. In thread mode, when the watermark reaches 3, `TwoInputOperation.on_timer()` returns `None`, which becomes Java `null`. `EmbeddedPythonIterator.from(null)` then throws an NPE. We still need an empty generator when the user callback is absent.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33404][python] Add default on_timer to CoProcessFunction [flink]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on PR #29286:
URL: https://github.com/apache/flink/pull/29286#issuecomment-5943111217
> I don't think we need to introduce the proposed interface in ProcessFunction. It's available in KeyedProcessFunction. Same for CoProcessFunction, the on_timer callable is available in KeyedCoProcessFunction.
@dianfu agreed on ProcessFunction, so I dropped it in https://github.com/apache/flink/commit/14d1146401c44958ab1f38e75fe2ac2fde96811c. I kept the CoProcessFunction default because ConnectedStreams.process accepts a plain CoProcessFunction on keyed streams and runs it as a keyed co-process operator, which calls on_timer on every timer. Without the default, a job that registers a timer fails with "AttributeError: 'Join' object has no attribute 'on_timer'"; the reproducer is in the description. Would you rather have ConnectedStreams.process reject a plain CoProcessFunction on keyed inputs instead?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33404][python] Add default timer callbacks [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on PR #29286:
URL: https://github.com/apache/flink/pull/29286#issuecomment-5924707942
@1fanwang Hey, I don't think we need to introduce the proposed interface in ProcessFunction. It's available in KeyedProcessFunction. Same for CoProcessFunction, the on_timer callable is available in KeyedCoProcessFunction.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org