[FLINK-38559][python] Support unordered mode of async function in Python DataStream API #27145

dianfu · 2025-10-24T08:50:17Z

What is the purpose of the change

This pull request provides basic support of async function in Python DataStream API. It currently only supports unordered mode. *

Brief change log

- Introduce basic structures, e.g. UnorderedStreamElementQueue, Emitter, AsyncFunctionRunner *
- Introduce AsyncOperation which is responsible for async function execution *

Verifying this change

- Added tests in test_async_function.py *

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? (yes)
If yes, how is the feature documented? (not documented: will add documentation in a separate pull request)

…hon DataStream API

flinkbot · 2025-10-24T08:56:26Z

CI report:

30433a1 Azure: FAILURE
29e64c5 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

charlesdong1991 · 2025-10-25T13:26:35Z

flink-python/pyflink/datastream/async_data_stream.py

+    def _validate(data_stream: DataStream, async_function: AsyncFunction):
+        if not inspect.iscoroutinefunction(async_function.async_invoke):
+            raise Exception("Method 'async_invoke' of class '%s' should be declared as 'async def'."
+                            % type(async_function))


Suggested change

% type(async_function))

% type(async_function).__name__)

i think type(function) will return a verbose result, maybe this change gives actual name

type(function) returns the full path of the class name and type(async_function).__name__ returns just the class name. My intention here is to print the full path which I think will be more helpful.

charlesdong1991 · 2025-10-25T13:27:57Z

flink-python/pyflink/datastream/async_data_stream.py

+            j_python_data_stream_function_operator))
+
+    @staticmethod
+    def _validate(data_stream: DataStream, async_function: AsyncFunction):


Suggested change

def _validate(data_stream: DataStream, async_function: AsyncFunction):

def _validate(data_stream: DataStream, async_function: AsyncFunction) -> None:

charlesdong1991 · 2025-10-25T14:38:39Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+            return False
+
+
+class ResultHandler(ResultFuture, Generic[OUT]):


do i understand correctly that retry is not supported?

Yes, you are right. It's planned to be supported in a separate ticket https://issues.apache.org/jira/browse/FLINK-38561

charlesdong1991 · 2025-10-25T14:42:04Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+
+class ResultHandler(ResultFuture, Generic[OUT]):
+
+    def __init__(self, classname, timeout_func, exception_handler, record,


would be nice to add type hints here too?

charlesdong1991 · 2025-10-25T15:08:34Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+            self._async_function_runner.stop()
+            self._async_function_runner = None
+
+    def process_element(self, windowed_value, element):


i wonder what will happen if the element is a watermark record?

Currently, the watermark and element are sent together, the structure of element is as following:

CURRENT_TIMESTAMP, CURRENT_WATERMARK, NORMAL_DATA

charlesdong1991 · 2025-10-25T15:20:17Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+        watermark = element[1]
+        record = element[2]
+
+        self._queue.advance_watermark(watermark)


for my own curiosity, what is the reason to advance watermark before the entry (the line below)?

Because the data is behind the watermark. You can refer to ExternalPythonProcessOperator for more details.

charlesdong1991 · 2025-10-25T15:24:16Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+            return
+
+        self._exception_handler(
+            Exception("Could not complete the element:" + str(self._record), error))


i wonder if it is possible to have a test case to trigger this exception?

dianfu · 2025-10-27T05:46:11Z

@charlesdong1991 Thanks for the review! Have updated the PR accordingly ~

bgeng777

Thanks for the PR! leave some comments

bgeng777 · 2025-10-27T07:32:04Z

flink-python/pyflink/fn_execution/datastream/process/async_function/queue.py

+
+        def __init__(self, capacity):
+            self._incomplete_elements = set()
+            self._complete_elements = collections.deque(maxlen=capacity)


maybe what we mean here is "completed" (i.e. something is finished) instead of "complete"(i.e. entire)

bgeng777 · 2025-10-27T07:46:08Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+        if not isinstance(result, Iterable):
+            raise RuntimeError("The 'result_future' of AsyncFunction should be completed with "
+                               "data of list type, please check the methods 'async_invoke' and "
+                               "'timeout' of class '%s'." % self._classname)


It looks like we can move 'RuntimeError' int the previous "else" instead of doing another if not isinstance(result, Iterable) again, Like

if not isinstance(result, Iterable): # complete with empty result, so that we remove timer and move ahead processing self._process_results([]) raise RuntimeError("The 'result_future' of AsyncFunction should be completed with " "data of list type, please check the methods 'async_invoke' and " "'timeout' of class '%s'." % self._classname) else: self._process_results(result)

bgeng777 · 2025-10-27T08:02:04Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+
+    def stop(self):
+        if self._loop is not None:
+            self._loop.stop()


Should we do self._loop.close() to free underlying resources?

Suggested change

self._loop.stop()

self._loop.stop()

self._loop.close()

Added self._loop.close() in method run.

bgeng777 · 2025-10-27T09:29:40Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+
+    def run_async(self, async_function, *arg):
+        wrapped_function = self.exception_handler_wrapper(async_function, *arg)
+        asyncio.run_coroutine_threadsafe(wrapped_function, self._loop)


I want to make sure I understand the error handling here correctly so I try to summarize the error handling process with a simple case:

there are 3 records that should be processed by AsyncOperation'sprocess_element.

self._async_function_runner.run_async is called for element1 in main thread.

self._async_function_runner.run_async is called for element2 in main thread but the task has not been submitted to theloop thread(i.e. line 172 has not been executed).

theloop thread executes the processing of element1, then exception happens, AsyncOperation marks exception.

AsyncOperation in main threads submits the element2 to loop thread.

AsyncOperation in main thread finds exception when processing element3 and raises. After some population, the main thread calls AsyncOperation's close() and stops the loop thread.
According to the doc of async io:

if the `loop` thread has not process element2's task, then the thread just stops and no extra processing for element2. If the `loop` thread is processing element2, the `loop` thread will finish the processing. But as AsyncOperation has been closed, is it possible that the `loop` thread runs into some bad cases(e.g. if the async function uses some freed resources setup in advance)?

I think your understanding is correct. I make a minor update to reflect this: moving self.close_func() to the end of the method close.

bgeng777 · 2025-10-27T09:32:01Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+        if output_mode == AsyncFunctionDescriptor.OutputMode.UNORDERED:
+            self._queue = UnorderedStreamElementQueue(capacity, self._raise_exception_if_exists)
+        else:
+            raise NotImplementedError()


Suggested change

raise NotImplementedError()

raise NotImplementedError("ORDERED mode is supported.")

bgeng777 · 2025-10-27T09:36:09Z

flink-python/pyflink/fn_execution/datastream/process/async_function/operation.py

+
+        if self._async_function_runner is not None:
+            self._async_function_runner.stop()
+            self._async_function_runner = None


do we need self._exception = None here?
In other words, when failover happens, will we reuse AsyncOperation object or just create a new one?

dianfu · 2025-10-28T03:15:23Z

@bgeng777 Thanks for the review. Updated the PR~

[FLINK-38190][python] Support unordered mode of async function in Pyt…

3625e30

…hon DataStream API

dianfu force-pushed the async_function branch from f4259c1 to 3625e30 Compare October 24, 2025 08:52

dianfu changed the title ~~[FLINK-38190][python] Support unordered mode of async function in Python DataStream API~~ [FLINK-38559][python] Support unordered mode of async function in Python DataStream API Oct 24, 2025

charlesdong1991 reviewed Oct 25, 2025

View reviewed changes

github-actions bot added the community-reviewed PR has been reviewed by the community. label Oct 25, 2025

dianfu added 2 commits October 27, 2025 12:59

add test_watermark test case

06251fa

address review comments

30433a1

bgeng777 reviewed Oct 27, 2025

View reviewed changes

address review comments

29e64c5

	def _validate(data_stream: DataStream, async_function: AsyncFunction):
	def _validate(data_stream: DataStream, async_function: AsyncFunction) -> None:

		return False


		class ResultHandler(ResultFuture, Generic[OUT]):


		class ResultHandler(ResultFuture, Generic[OUT]):

		def __init__(self, classname, timeout_func, exception_handler, record,

	raise NotImplementedError()
	raise NotImplementedError("ORDERED mode is supported.")

Uh oh!

[FLINK-38559][python] Support unordered mode of async function in Python DataStream API #27145

Are you sure you want to change the base?

[FLINK-38559][python] Support unordered mode of async function in Python DataStream API #27145

Conversation

dianfu commented Oct 24, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Oct 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dianfu Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dianfu commented Oct 27, 2025

Uh oh!

bgeng777 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dianfu commented Oct 28, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Oct 24, 2025 •

edited

Loading

dianfu Oct 27, 2025 •

edited

Loading