lincoln-lil commented on code in PR #19983:
URL: https://github.com/apache/flink/pull/19983#discussion_r908419613
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AsyncDataStream.java:
##########
@@ -163,4 +193,134 @@ public static <IN, OUT> SingleOutputStreamOperator<OUT>
orderedWait(
return addOperator(
in, func, timeUnit.toMillis(timeout), DEFAULT_QUEUE_CAPACITY,
OutputMode.ORDERED);
}
+
+ // ======= retryable ========
+
+ /**
+ * Add an AsyncWaitOperator with an AsyncRetryStrategy to support retry of
AsyncFunction. The
+ * order of output stream records may be reordered.
+ *
+ * @param in Input {@link DataStream}
+ * @param func {@link AsyncFunction}
+ * @param timeout from first invoke to final completion of asynchronous
operation, may include
+ * multiple retries, and will be reset in case of restart
+ * @param timeUnit of the given timeout
+ * @param asyncRetryStrategy The strategy of reattempt async i/o operation
that can be triggered
+ * @param <IN> Type of input record
+ * @param <OUT> Type of output record
+ * @return A new {@link SingleOutputStreamOperator}.
+ */
+ public static <IN, OUT> SingleOutputStreamOperator<OUT>
unorderedWaitWithRetry(
+ DataStream<IN> in,
+ AsyncFunction<IN, OUT> func,
+ long timeout,
+ TimeUnit timeUnit,
+ AsyncRetryStrategy asyncRetryStrategy) {
+ Preconditions.checkArgument(
Review Comment:
The 'timeout' parameter does not necessarily need to be greater than 0 when
retry is not enabled, so move this check into the addOperator and add a
condition 'asyncRetryStrategy != NO_RETRY_STRATEGY'.
--
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]
