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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
