Skip to content

[FLINK-38561][python] Add retry support for async function in Python DataStream API #27164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Closed
wants to merge 3 commits into from
Closed

[FLINK-38561][python] Add retry support for async function in Python DataStream API #27164

wants to merge 3 commits into from

Conversation

@dianfu
Copy link
Contributor

@dianfu dianfu commented Oct 29, 2025

What is the purpose of the change

*This pull request adds retry support for async function in Python DataStream API. *

Brief change log

    • Introduces API (AsyncDataStream.unordered_wait_with_retry) for retry support*
    • Introduces RetryableResultHandler to handle retry*

Verifying this change

This change added tests and can be verified as follows:

  • Added tests test_async_with_retry

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

  • Dependencies (does it add or upgrade a dependency): (yes / no)
  • The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
  • The serializers: (yes / no / don't know)
  • The runtime per-record code paths (performance sensitive): (yes / no / don't know)
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
  • The S3 file system connector: (yes / no / don't know)

Documentation

  • Does this pull request introduce a new feature? (yes)
  • If yes, how is the feature documented? (not documented, will be added in a separate pull request)

@flinkbot
Copy link
Collaborator

flinkbot commented Oct 29, 2025 •
edited
Loading

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

bgeng777
bgeng777 reviewed Oct 29, 2025
View reviewed changes
Copy link
Contributor

@bgeng777 bgeng777 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the PR, just some minor comments

flink-python/pyflink/datastream/tests/test_async_function.py Outdated
async_retry_strategy = AsyncRetryStrategy.fixed_delay(
max_attempts=5,
backoff_time_millis=1000,
result_predicate=None,
Copy link
Contributor

@bgeng777 bgeng777 Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It could be nicer if we add a test for result_predicate

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

if (not self._is_timeout() and satisfy and
self._retry_strategy.can_retry(self._current_attempts)):

Copy link
Contributor

@bgeng777 bgeng777 Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

extra blank line

flink-python/pyflink/datastream/async_data_stream.py
capacity, output_type)

@staticmethod
def unordered_wait_with_retry(
Copy link
Contributor

@bgeng777 bgeng777 Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we prefer introducing unordered_wait_with_retry instead of adding a new parameter(AsyncRetryStrategy, default is None) to unordered_wait

Copy link
Contributor Author

@dianfu dianfu Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is to follow the design of Java API.

Copy link
Contributor

@bgeng777 bgeng777 Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java's AsyncWaitOperator has only 1 constructor with asyncRetryStrategy as an parameter; so here you mean org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorFactory#AsyncWaitOperatorFactory ?

Copy link
Contributor Author

@dianfu dianfu Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean this

flink/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AsyncDataStream.java

Line 217 in 5b61b1c

public static <IN, OUT> SingleOutputStreamOperator<OUT> unorderedWaitWithRetry(

davidradl
davidradl reviewed Oct 29, 2025
View reviewed changes
flink-python/pyflink/datastream/async_data_stream.py
output_type: TypeInformation = None) -> 'DataStream':
"""
Adds an async function with an AsyncRetryStrategy to support retry of AsyncFunction to the
data stream. The order of output stream records may be reordered.
Copy link
Contributor

@davidradl davidradl Oct 29, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I assume this is because we hard code AsyncFunctionDescriptor.OutputMode.UNORDERED. Is there thinking about an ordered version of this?

wording nit: The order of output stream records may be reordered. -> The output stream records may be reordered.

Copy link
Contributor Author

@dianfu dianfu Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The ordered version is to be supported in https://issues.apache.org/jira/browse/FLINK-38560 (PR is already ready and will submit for review once this PR is merged.)

Copy link
Contributor Author

@dianfu dianfu Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wording nit: The order of output stream records may be reordered. -> The output stream records may be reordered.

Oh, this description is copied from the Java API.

bgeng777
bgeng777 approved these changes Oct 29, 2025
View reviewed changes
Copy link
Contributor

@bgeng777 bgeng777 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM!

@dianfu dianfu closed this in 2c4a45c Oct 29, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

2 more reviewers

@davidradl davidradl davidradl left review comments

@bgeng777 bgeng777 bgeng777 approved these changes

Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

4 participants

@dianfu @flinkbot @davidradl @bgeng777