[GitHub] [flink-statefun] igalshilman commented on a change in pull request #30: [FLINK-16226] Add Backpressure to HttpFunction

[GitBox <gi...@apache.org>]

igalshilman commented on a change in pull request #30: [FLINK-16226] Add Backpressure to HttpFunction
URL: https://github.com/apache/flink-statefun/pull/30#discussion_r382895705
 
 

 ##########
 File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunction.java
 ##########
 @@ -89,12 +101,28 @@ public void invoke(Context context, Object input) {
 
   private void onRequest(Context context, Any message) {
     Invocation.Builder invocationBuilder = singeInvocationBuilder(context, message);
-    if (hasInFlightRpc.getOrDefault(Boolean.FALSE)) {
-      batch.append(invocationBuilder.build());
+    int inflightOrBatched = requestState.getOrDefault(-1);
+    if (inflightOrBatched < 0) {
+      // no inflight requests, and nothing in the batch.
+      // so we let this request to go through, and change state to indicate that:
+      // a) there is a request in flight.
+      // b) there is nothing in the batch.
+      requestState.set(0);
+      sendToFunction(context, invocationBuilder);
       return;
     }
-    hasInFlightRpc.set(Boolean.TRUE);
-    sendToFunction(context, invocationBuilder);
+    // there is at least one request in flight (inflightOrBatched >= 0),
+    // so we add that request to the batch.
+    batch.append(invocationBuilder.build());
+    inflightOrBatched++;
+    requestState.set(inflightOrBatched);
+    if (isMaxBatchSizeExceeded(inflightOrBatched)) {
+      // we are at capacity, can't add anything to the batch.
+      // we need to signal to the runtime that we are unable to process any new input
+      // and we must wait for our in flight asynchronous operation to complete before
+      // we are able to process more input.
+      ((AsyncWaiter) context).awaitAsyncOperationComplete();
 
 Review comment:
   AsyncWaiter doesn’t extend from Context, and me need both of them in this method

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services