tzulitai commented on a change in pull request #30: [FLINK-16226] Add
Backpressure to HttpFunction
URL: https://github.com/apache/flink-statefun/pull/30#discussion_r382901358
##########
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:
Yes, so I was thinking about something like
`onRequest(context, (AsyncWaiter) context, Any message)`.
It doesn't matter though, to keep it as is.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
