[FLINK-38563][python][docs] Update documentation of Python AsyncFunction (#27195)

Author: dianfu

docs/content.zh/docs/dev/datastream/operators/asyncio.md

@@ -61,11 +61,14 @@ Flink 的异步 I/O API 允许用户在流处理中使用异步请求客户端
 在具备异步数据库客户端的基础上，实现数据流转换操作与数据库的异步 I/O 交互需要以下三部分：
 
 - 实现分发请求的 `AsyncFunction`
-- 获取数据库交互的结果并发送给 `ResultFuture` 的 *回调* 函数
+- 如果使用的 Java API，获取数据库交互的结果并发送给 `ResultFuture` 的 *回调* 函数；如果使用的 Python API，可以通过 await 获取数据库交互的结果
 - 将异步 I/O 操作应用于 `DataStream` 作为 `DataStream` 的一次转换操作, 启用或者不启用重试。
 
 下面是基本的代码模板：
 
+{{< tabs "6c8c009c-4c12-4338-9eeb-3be83cfa9e36" >}}
+{{< tab "Java" >}}
+
 ```java
 // 这个例子使用 Java 8 的 Future 接口（与 Flink 的 Future 相同）实现了异步请求和回调。
 
@@ -132,7 +135,74 @@ DataStream<Tuple2<String, String>> resultStream =
 	AsyncDataStream.unorderedWaitWithRetry(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100, asyncRetryStrategy);
 ```
 
-**重要提示**： 第一次调用 `ResultFuture.complete` 后 `ResultFuture` 就完成了。
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+from typing import List
+
+from pyflink.common import Time, Types
+from pyflink.datastream import AsyncFunction, AsyncDataStream, async_retry_predicates
+from pyflink.datastream.functions import RuntimeContext, AsyncRetryStrategy
+
+
+class AsyncDatabaseRequest(AsyncFunction[str, (str, str)]):
+
+    def __init__(self, host, port, credentials):
+        self._host = host
+        self._port = port
+        self._credentials = credentials
+
+    def open(self, runtime_context: RuntimeContext):
+        # The database specific client that can issue concurrent requests with callbacks
+        self._client = DatabaseClient(self._host, self._port, self._credentials)
+
+    def close(self):
+        if self._client:
+            self._client.close()
+
+    async def async_invoke(self, value: str) -> List[(str, str)]:
+        try:
+            # issue the asynchronous request
+            result = await self._client.query(value)
+            return [(value, str(result))]
+        except Exception:
+            return [(value, None)]
+
+
+# 创建初始 DataStream
+stream = ...
+
+# 应用异步 I/O 转换操作，不启用重试
+result_stream = AsyncDataStream.unordered_wait(
+    data_stream=stream,
+    async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+    timeout=Time.seconds(10),
+    capacity=100,
+    output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+
+# 或应用异步 I/O 转换操作并启用重试
+# 通过工具类创建一个异步重试策略, 或用户实现自定义的策略
+async_retry_strategy = AsyncRetryStrategy.fixed_delay(
+    max_attempts=3,
+    backoff_time_millis=100,
+    result_predicate=async_retry_predicates.empty_result_predicate,
+    exception_predicate=async_retry_predicates.has_exception_predicate)
+
+# 应用异步 I/O 转换操作并启用重试
+result_stream_with_retry = AsyncDataStream.unordered_wait_with_retry(
+    data_stream=stream,
+    async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+    timeout=Time.seconds(10),
+    async_retry_strategy=async_retry_strategy,
+    capacity=1000,
+    output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+**重要提示**： 在 Java API 中，第一次调用 `ResultFuture.complete` 后 `ResultFuture` 就完成了。
 后续的 `complete` 调用都将被忽略。
 
 下面两个参数控制异步操作：
@@ -149,9 +219,13 @@ DataStream<Tuple2<String, String>> resultStream =
 
 当异步 I/O 请求超时的时候，默认会抛出异常并重启作业。
 如果你想处理超时，可以重写 `AsyncFunction#timeout` 方法。
-重写 `AsyncFunction#timeout` 时别忘了调用 `ResultFuture.complete()` 或者 `ResultFuture.completeExceptionally()`
+
+在 Java API 中，重写 `AsyncFunction#timeout` 时别忘了调用 `ResultFuture.complete()` 或者 `ResultFuture.completeExceptionally()`
 以便告诉Flink这条记录的处理已经完成。如果超时发生时你不想发出任何记录，你可以调用 `ResultFuture.complete(Collections.emptyList())` 。
 
+在 Python API 中，可以返回一个 List 或者抛异常以便告诉Flink这条记录的处理已经完成。如果超时发生时你不想发出任何记录，
+你可以调用 `return []` 以返回一个空列表。
+
 ### 结果的顺序
 
 `AsyncFunction` 发出的并发请求经常以不确定的顺序完成，这取决于请求得到响应的顺序。
@@ -160,9 +234,9 @@ Flink 提供两种模式控制结果记录以何种顺序发出。
   - **无序模式**： 异步请求一结束就立刻发出结果记录。
     流中记录的顺序在经过异步 I/O 算子之后发生了改变。
     当使用 *处理时间* 作为基本时间特征时，这个模式具有最低的延迟和最少的开销。
-    此模式使用 `AsyncDataStream.unorderedWait(...)` 方法。
+    此模式使用 `AsyncDataStream.unorderedWait(...)` 或者 `AsyncDataStream.unordered_wait(...)` 方法。
 
-  - **有序模式**: 这种模式保持了流的顺序。发出结果记录的顺序与触发异步请求的顺序（记录输入算子的顺序）相同。为了实现这一点，算子将缓冲一个结果记录直到这条记录前面的所有记录都发出（或超时）。由于记录或者结果要在 checkpoint 的状态中保存更长的时间，所以与无序模式相比，有序模式通常会带来一些额外的延迟和 checkpoint 开销。此模式使用 `AsyncDataStream.orderedWait(...)` 方法。
+  - **有序模式**: 这种模式保持了流的顺序。发出结果记录的顺序与触发异步请求的顺序（记录输入算子的顺序）相同。为了实现这一点，算子将缓冲一个结果记录直到这条记录前面的所有记录都发出（或超时）。由于记录或者结果要在 checkpoint 的状态中保存更长的时间，所以与无序模式相比，有序模式通常会带来一些额外的延迟和 checkpoint 开销。此模式使用 `AsyncDataStream.orderedWait(...)` 或者 `AsyncDataStream.ordered_wait(...)` 方法。
 
 
 ### 事件时间
@@ -203,6 +277,7 @@ Flink 提供两种模式控制结果记录以何种顺序发出。
 `DirectExecutor` 可以通过 `org.apache.flink.util.concurrent.Executors.directExecutor()` 或
 `com.google.common.util.concurrent.MoreExecutors.directExecutor()` 获得。
 
+**注意:** 这仅适用于 Java API，在 Python API 中，您可以使用 await 等待异步执行的结果。
 
 ### 警告
 

docs/content/docs/dev/datastream/operators/asyncio.md

@@ -75,11 +75,14 @@ Assuming one has an asynchronous client for the target database, three parts are
 with asynchronous I/O against the database:
 
   - An implementation of `AsyncFunction` that dispatches the requests
-  - A *callback* that takes the result of the operation and hands it to the `ResultFuture`
+  - A *callback* that takes the result of the operation and hands it to the `ResultFuture` in Java API or await the result of the operation in Python API
   - Applying the async I/O operation on a DataStream as a transformation with or without retry
 
 The following code example illustrates the basic pattern:
 
+{{< tabs "6c8c009c-4c12-4338-9eeb-3be83cfa9e36" >}}
+{{< tab "Java" >}}
+
 ```java
 // This example implements the asynchronous request and callback with Futures that have the
 // interface of Java 8's futures (which is the same one followed by Flink's Future)
@@ -147,7 +150,74 @@ DataStream<Tuple2<String, String>> resultStream =
 	AsyncDataStream.unorderedWaitWithRetry(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100, asyncRetryStrategy);
 ```
 
-**Important note**: The `ResultFuture` is completed with the first call of `ResultFuture.complete`.
+{{< /tab >}}
+{{< tab "Python" >}}
+
+```python
+from typing import List
+
+from pyflink.common import Time, Types
+from pyflink.datastream import AsyncFunction, AsyncDataStream, async_retry_predicates
+from pyflink.datastream.functions import RuntimeContext, AsyncRetryStrategy
+
+
+class AsyncDatabaseRequest(AsyncFunction[str, (str, str)]):
+
+    def __init__(self, host, port, credentials):
+        self._host = host
+        self._port = port
+        self._credentials = credentials
+
+    def open(self, runtime_context: RuntimeContext):
+        # The database specific client that can issue concurrent requests with callbacks
+        self._client = DatabaseClient(self._host, self._port, self._credentials)
+
+    def close(self):
+        if self._client:
+            self._client.close()
+
+    async def async_invoke(self, value: str) -> List[(str, str)]:
+        try:
+            # issue the asynchronous request
+            result = await self._client.query(value)
+            return [(value, str(result))]
+        except Exception:
+            return [(value, None)]
+
+
+# create the original stream
+stream = ...
+
+# apply the async I/O transformation without retry
+result_stream = AsyncDataStream.unordered_wait(
+    data_stream=stream,
+    async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+    timeout=Time.seconds(10),
+    capacity=100,
+    output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+
+# or apply the async I/O transformation with retry
+# create an async retry strategy via utility class or a user defined strategy
+async_retry_strategy = AsyncRetryStrategy.fixed_delay(
+    max_attempts=3,
+    backoff_time_millis=100,
+    result_predicate=async_retry_predicates.empty_result_predicate,
+    exception_predicate=async_retry_predicates.has_exception_predicate)
+
+# apply the async I/O transformation with retry
+result_stream_with_retry = AsyncDataStream.unordered_wait_with_retry(
+    data_stream=stream,
+    async_function=AsyncDatabaseRequest("127.0.0.1", "1234", None),
+    timeout=Time.seconds(10),
+    async_retry_strategy=async_retry_strategy,
+    capacity=1000,
+    output_type=Types.TUPLE([Types.STRING(), Types.STRING()]))
+```
+
+{{< /tab >}}
+{{< /tabs >}}
+
+**Important note**: The `ResultFuture` is completed with the first call of `ResultFuture.complete` in the Java API.
 All subsequent `complete` calls will be ignored.
 
 The following three parameters control the asynchronous operations:
@@ -162,17 +232,21 @@ The following three parameters control the asynchronous operations:
     accumulate an ever-growing backlog of pending requests, but that it will trigger backpressure once the capacity
     is exhausted.
 
-  - **AsyncRetryStrategy**: The asyncRetryStrategy defines what conditions will trigger a delayed retry and the delay strategy,
+  - **AsyncRetryStrategy**: This parameter defines what conditions will trigger a delayed retry and the delay strategy,
     e.g., fixed-delay, exponential-backoff-delay, custom implementation, etc.
 
 ### Timeout Handling
 
 When an async I/O request times out, by default an exception is thrown and job is restarted.
 If you want to handle timeouts, you can override the `AsyncFunction#timeout` method.
-Make sure you call `ResultFuture.complete()` or `ResultFuture.completeExceptionally()` when overriding
+
+In the Java API, make sure you call `ResultFuture.complete()` or `ResultFuture.completeExceptionally()` when overriding
 in order to indicate to Flink that the processing of this input record has completed. You can call 
 `ResultFuture.complete(Collections.emptyList())` if you do not want to emit any record when timeouts happen.
 
+In the Python API, you can return a collection of results or raise an exception when overriding
+in order to indicate to Flink that the processing of this input record has completed. You can return
+empty list by calling `return []` if you do not want to emit any record when timeouts happen.
 
 ### Order of Results
 
@@ -182,14 +256,14 @@ To control in which order the resulting records are emitted, Flink offers two mo
   - **Unordered**: Result records are emitted as soon as the asynchronous request finishes.
     The order of the records in the stream is different after the async I/O operator than before.
     This mode has the lowest latency and lowest overhead, when used with *processing time* as the basic time characteristic.
-    Use `AsyncDataStream.unorderedWait(...)` for this mode.
+    Use `AsyncDataStream.unorderedWait(...)` or `AsyncDataStream.unordered_wait(...)` for this mode.
 
   - **Ordered**: In that case, the stream order is preserved. Result records are emitted in the same order as the asynchronous
     requests are triggered (the order of the operators input records). To achieve that, the operator buffers a result record
     until all its preceding records are emitted (or timed out).
     This usually introduces some amount of extra latency and some overhead in checkpointing, because records or results are maintained
     in the checkpointed state for a longer time, compared to the unordered mode.
-    Use `AsyncDataStream.orderedWait(...)` for this mode.
+    Use `AsyncDataStream.orderedWait(...)` or `AsyncDataStream.ordered_wait(...)` for this mode.
 
 
 ### Event Time
@@ -240,6 +314,7 @@ with the checkpoint bookkeeping happens in a dedicated thread-pool anyways.
 A `DirectExecutor` can be obtained via `org.apache.flink.util.concurrent.Executors.directExecutor()` or
 `com.google.common.util.concurrent.MoreExecutors.directExecutor()`.
 
+**NOTE:** This only applies for the Java API. In the Python API, you could just await the asynchronous result.
 
 ### Caveats
 

flink-python/docs/reference/pyflink.common/index.rst

@@ -28,4 +28,5 @@ This page gives an overview of all public PyFlink Common API.
     config
     typeinfo
     job_info
-    serializer
+    serializer
+    time

flink-python/docs/reference/pyflink.common/time.rst

@@ -0,0 +1,65 @@
+.. ################################################################################
+     Licensed to the Apache Software Foundation (ASF) under one
+     or more contributor license agreements.  See the NOTICE file
+     distributed with this work for additional information
+     regarding copyright ownership.  The ASF licenses this file
+     to you under the Apache License, Version 2.0 (the
+     "License"); you may not use this file except in compliance
+     with the License.  You may obtain a copy of the License at
+
+         http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing, software
+     distributed under the License is distributed on an "AS IS" BASIS,
+     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     See the License for the specific language governing permissions and
+    limitations under the License.
+   ################################################################################
+
+
+====
+Time
+====
+
+Duration
+--------
+
+.. currentmodule:: pyflink.common.time
+
+.. autosummary::
+    :toctree: api/
+
+    Duration.of_days
+    Duration.of_hours
+    Duration.of_minutes
+    Duration.of_millis
+    Duration.of_seconds
+    Duration.of_nanos
+
+
+Instant
+-------
+
+.. currentmodule:: pyflink.common.time
+
+.. autosummary::
+    :toctree: api/
+
+    Instant.of_epoch_milli
+    Instant.to_epoch_milli
+
+
+Time
+----
+
+.. currentmodule:: pyflink.common.time
+
+.. autosummary::
+    :toctree: api/
+
+    Time.milliseconds
+    Time.seconds
+    Time.minutes
+    Time.hours
+    Time.days
+    Time.to_milliseconds