sangkyoonnam opened a new pull request, #29635: URL: https://github.com/apache/flink/pull/29635
## What is the purpose of the change With `python.execution-mode: thread`, `AsyncDataStream.unordered_wait` and `ordered_wait` fail with "AsyncFunction is still not supported for 'thread' mode." This adds a thread-mode operator that follows `AsyncWaitOperator`, as agreed on [FLINK-38562](https://issues.apache.org/jira/browse/FLINK-38562). Retry in thread mode isn't part of this PR, so thread mode accepts only `NO_RETRY_STRATEGY`. ## Brief change log - `EmbeddedPythonAsyncWaitOperator` ports `AsyncWaitOperator`'s control logic onto its ordered and unordered queues: mailbox backpressure, processing-time timeouts and `SupportsChainAvailability`. Its factory implements `YieldingOperatorFactory`, so it keeps `ALWAYS` chaining but isn't chained to a legacy source. Like `AsyncWaitOperator`, and unlike process mode, it stores in-flight inputs in operator state and calls the function again for them after a restore. - `async_operation.py` runs `async_invoke` on an asyncio event loop thread and reports each result or error to a Java handler, which converts and copies results in the mailbox. It has its own runner since Java owns completion and timers here. Python `timeout` runs on the task thread only if no result has arrived; a result arriving meanwhile still wins. If the loop exits unexpectedly, pending inputs fail. Like process mode, shutdown waits up to one second for the loop and doesn't cancel Python work. - `AsyncDataStream` uses it in thread mode, and `execution_mode.rst` marks Async I/O supported in both modes, retry in process mode only. ## Verifying this change This change added tests and can be verified as follows: - `EmbeddedPythonAsyncWaitOperatorTest` drives the operator with a fake Python side: ordering around watermarks, several rows per input, timeout races, failures, snapshot and restore, and chain availability. `test_embedded_async_operation.py` covers the loop thread. - `test_async_function.py` runs shared tests in both modes, plus new ones for parallelism 2, capacity 1 behind a chained Python `map`, several rows per input, and `async_invoke` errors. Thread mode adds the retry checks and a restore after a failure induced once a checkpoint holds an in-flight input, checking no value is lost. - Breaking the copy, snapshot, factory marker, completion check, availability or error reporting fails the matching tests. `./mvnw clean verify -pl flink-python` passes (474 tests and 1 IT), and the async Python tests passed 20 consecutive runs. ## 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)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): yes, but only async functions in thread mode, which were rejected before - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, the new operator keeps its in-flight inputs in operator state, like `AsyncWaitOperator` - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? docs (`flink-python/docs/user_guide/execution_mode.rst`) --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) AI was used for the implementation, the verification runs and this description. Generated-by: Claude Code 2.1.308 (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]
