[flink-statefun] 05/07: [FLINK-19106] [datastream] Expose new timeout configs for DataStream API

[tz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git

commit 28aeee618dba74c9fb7e56fb5b5f9e828bf0a498
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Sep 1 17:31:34 2020 +0800

    [FLINK-19106] [datastream] Expose new timeout configs for DataStream API
    
    This closes #138.
---
 .../datastream/RequestReplyFunctionBuilder.java    | 36 +++++++++++++++++++++-
 1 file changed, 35 insertions(+), 1 deletion(-)

diff --git a/statefun-flink/statefun-flink-datastream/src/main/java/org/apache/flink/statefun/flink/datastream/RequestReplyFunctionBuilder.java b/statefun-flink/statefun-flink-datastream/src/main/java/org/apache/flink/statefun/flink/datastream/RequestReplyFunctionBuilder.java
index 07b0ce1..58c382c 100644
--- a/statefun-flink/statefun-flink-datastream/src/main/java/org/apache/flink/statefun/flink/datastream/RequestReplyFunctionBuilder.java
+++ b/statefun-flink/statefun-flink-datastream/src/main/java/org/apache/flink/statefun/flink/datastream/RequestReplyFunctionBuilder.java
@@ -71,7 +71,8 @@ public class RequestReplyFunctionBuilder {
   }
 
   /**
-   * Set a maximum request duration.
+   * Set a maximum request duration. This duration spans the complete call, including connecting to
+   * the function endpoint, writing the request, function processing, and reading the response.
    *
    * @param duration the duration after which the request is considered failed.
    * @return this builder.
@@ -82,6 +83,39 @@ public class RequestReplyFunctionBuilder {
   }
 
   /**
+   * Set a timeout for connecting to function endpoints.
+   *
+   * @param duration the duration after which a connect attempt is considered failed.
+   * @return this builder.
+   */
+  public RequestReplyFunctionBuilder withConnectTimeout(Duration duration) {
+    builder.withConnectTimeoutDuration(duration);
+    return this;
+  }
+
+  /**
+   * Set a timeout for individual read IO operations during a function invocation request.
+   *
+   * @param duration the duration after which a read IO operation is considered failed.
+   * @return this builder.
+   */
+  public RequestReplyFunctionBuilder withReadTimeout(Duration duration) {
+    builder.withReadTimeoutDuration(duration);
+    return this;
+  }
+
+  /**
+   * Set a timeout for individual write IO operations during a function invocation request.
+   *
+   * @param duration the duration after which a write IO operation is considered failed.
+   * @return this builder.
+   */
+  public RequestReplyFunctionBuilder withWriteTimeout(Duration duration) {
+    builder.withWriteTimeoutDuration(duration);
+    return this;
+  }
+
+  /**
    * Sets the max messages to batch together for a specific address.
    *
    * @param maxNumBatchRequests the maximum number of requests to batch for an address.