xds: remove flow control violation error handling in ext_proc client interceptor

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -1107,14 +1107,6 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
             com.google.protobuf.ByteString body = streamed.getBody();
             boolean sendImmediately = false;
             synchronized (streamLock) {
-              if (sidestreamToUpstreamWindow <= 0) {
-                internalOnError(Status.INTERNAL
-                    .withDescription(
-                        "Flow control violation: received client body from ext_proc "
-                        + "when window is closed")
-                    .asRuntimeException());
-                return;
-              }
               sidestreamToUpstreamWindow -= body.size();
               if (super.isReady() && pendingUpstreamBodyMessages.isEmpty()) {
                 sendImmediately = true;
@@ -1152,14 +1144,6 @@ private void handleResponseBodyResponse(
           com.google.protobuf.ByteString body = streamed.getBody();
           final int bodySize = body.size();
           synchronized (streamLock) {
-            if (sidestreamToDownstreamWindow <= 0) {
-              internalOnError(Status.INTERNAL
-                  .withDescription(
-                      "Flow control violation: received server body from ext_proc "
-                      + "when window is closed")
-                  .asRuntimeException());
-              return;
-            }
             sidestreamToDownstreamWindow -= bodySize;
           }
           deliverResponseBody(body, listener);

xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java

@@ -12014,364 +12014,7 @@ public void onMessage(String message) {
     channelManager.close();
   }
 
-  @Test
-  @SuppressWarnings("unchecked")
-  public void testSidestreamToDownstreamFlowControl_Violations() throws Exception {
-    ExternalProcessor proto = ExternalProcessor.newBuilder()
-        .setGrpcService(GrpcService.newBuilder()
-            .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
-                .setTargetUri("in-process:///" + extProcServerName)
-                .addChannelCredentialsPlugin(Any.newBuilder()
-                    .setTypeUrl(INSECURE_CREDENTIALS_TYPE_URL)
-                    .build())
-                .build())
-            .build())
-        .setProcessingMode(ProcessingMode.newBuilder()
-            .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
-            .setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
-            .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND)
-            .setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
-            .build())
-        .build();
-    ConfigOrError<ExternalProcessorFilterConfig> configOrError =
-        provider.parseFilterConfig(Any.pack(proto), filterContext);
-    assertThat(configOrError.errorDetail).isNull();
-    ExternalProcessorFilterConfig filterConfig = configOrError.config;
-
-    final String mutatedMessageTooLarge = new String(new char[70000]).replace('\0', 'd');
-    final CountDownLatch callClosedLatch = new CountDownLatch(1);
-    final AtomicReference<Status> capturedStatus = new AtomicReference<>();
-
-    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
-        new ExternalProcessorGrpc.ExternalProcessorImplBase() {
-          @Override
-          public StreamObserver<ProcessingRequest> process(
-              final StreamObserver<ProcessingResponse> responseObserver) {
-            ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
-            return new StreamObserver<ProcessingRequest>() {
-              @Override
-              public void onNext(ProcessingRequest request) {
-                if (request.hasRequestHeaders()) {
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setRequestHeaders(HeadersResponse.newBuilder().build())
-                      .build());
-                } else if (request.hasRequestBody()) {
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setRequestBody(BodyResponse.newBuilder()
-                          .setResponse(CommonResponse.newBuilder()
-                              .setBodyMutation(BodyMutation.newBuilder()
-                                  .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                      .setBody(request.getRequestBody().getBody())
-                                      .build())
-                                  .build())
-                              .build())
-                          .build())
-                      .build());
-                } else if (request.hasResponseHeaders()) {
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setResponseHeaders(HeadersResponse.newBuilder().build())
-                      .build());
-                } else if (request.hasResponseBody()) {
-                  // Send Chunk 1 (70k) -> Consumed by client app's pending request count of 1.
-                  // Replenishes window.
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setResponseBody(BodyResponse.newBuilder()
-                          .setResponse(CommonResponse.newBuilder()
-                              .setBodyMutation(BodyMutation.newBuilder()
-                                  .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                      .setBody(ByteString.copyFromUtf8(mutatedMessageTooLarge))
-                                      .build())
-                                  .build())
-                              .build())
-                          .build())
-                      .build());
-
-                  // Send Chunk 2 (70k) -> Excess chunk. Stored in queue. Window stays negative.
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setResponseBody(BodyResponse.newBuilder()
-                          .setResponse(CommonResponse.newBuilder()
-                              .setBodyMutation(BodyMutation.newBuilder()
-                                  .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                      .setBody(ByteString.copyFromUtf8(mutatedMessageTooLarge))
-                                      .build())
-                                  .build())
-                              .build())
-                          .build())
-                      .build());
-
-                  // Send Chunk 3 (1 byte) -> Arrives when window is negative, triggering violation.
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setResponseBody(BodyResponse.newBuilder()
-                          .setResponse(CommonResponse.newBuilder()
-                              .setBodyMutation(BodyMutation.newBuilder()
-                                  .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                      .setBody(ByteString.copyFromUtf8("a"))
-                                      .build())
-                                  .build())
-                              .build())
-                          .build())
-                      .build());
-                }
-              }
-
-              @Override
-              public void onError(Throwable t) {}
-
-              @Override
-              public void onCompleted() {
-                responseObserver.onCompleted();
-              }
-            };
-          }
-        };
-
-    String uniqueExtProcServerName = InProcessServerBuilder.generateName();
-    grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
-        .addService(extProcImpl)
-        .directExecutor()
-        .build().start());
-
-    CachedChannelManager channelManager = new CachedChannelManager(config -> {
-      return grpcCleanup.register(
-          InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
-    });
-
-    ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
-        filterConfig, channelManager, scheduler, FAKE_CONTEXT);
-
-    final AtomicReference<StreamObserver<String>> dataPlaneResponseObserverRef =
-        new AtomicReference<>();
-    dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
-        .addMethod(METHOD_BIDI_STREAMING, ServerCalls.asyncBidiStreamingCall(
-            new ServerCalls.BidiStreamingMethod<String, String>() {
-              @Override
-              public StreamObserver<String> invoke(StreamObserver<String> responseObserver) {
-                dataPlaneResponseObserverRef.set(responseObserver);
-                return new StreamObserver<String>() {
-                  @Override
-                  public void onNext(String value) {}
-
-                  @Override
-                  public void onError(Throwable t) {}
-
-                  @Override
-                  public void onCompleted() {}
-                };
-              }
-            }))
-        .build());
-
-    ManagedChannel dataPlaneChannel = grpcCleanup.register(
-        InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
-
-    ClientCall<String, String> proxyCall =
-        interceptCall(interceptor, METHOD_BIDI_STREAMING,
-            DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()),
-            dataPlaneChannel);
-
-    proxyCall.start(new ClientCall.Listener<String>() {
-      @Override
-      public void onClose(Status status, Metadata trailers) {
-        capturedStatus.set(status);
-        callClosedLatch.countDown();
-      }
-    }, new Metadata());
-    proxyCall.request(1);
-
-    proxyCall.sendMessage("Client Msg");
-
-    // Send a response from upstream to trigger headers and then the body response
-    StreamObserver<String> upstreamResponseObserver = dataPlaneResponseObserverRef.get();
-    upstreamResponseObserver.onNext("Response Msg");
-
-    assertThat(callClosedLatch.await(5, TimeUnit.SECONDS)).isTrue();
-
-    // The call should fail immediately with INTERNAL error code due to flow control violation
-    assertThat(capturedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
-    assertThat(capturedStatus.get().getDescription())
-
-        .isEqualTo("External processor stream failed");
-    assertThat(capturedStatus.get().getCause()).isInstanceOf(io.grpc.StatusRuntimeException.class);
-    assertThat(capturedStatus.get().getCause().getMessage())
-        .contains("Flow control violation: received server body from ext_proc "
-            + "when window is closed");
-
-    channelManager.close();
-  }
-
-  @Test
-  @SuppressWarnings("unchecked")
-  public void testSidestreamToUpstreamFlowControl_Violations() throws Exception {
-    ExternalProcessor proto = ExternalProcessor.newBuilder()
-        .setGrpcService(GrpcService.newBuilder()
-            .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
-                .setTargetUri("in-process:///" + extProcServerName)
-                .addChannelCredentialsPlugin(Any.newBuilder()
-                    .setTypeUrl(INSECURE_CREDENTIALS_TYPE_URL)
-                    .build())
-                .build())
-            .build())
-        .setProcessingMode(ProcessingMode.newBuilder()
-            .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
-            .setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
-            .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND)
-            .setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
-            .build())
-        .build();
-    ConfigOrError<ExternalProcessorFilterConfig> configOrError =
-        provider.parseFilterConfig(Any.pack(proto), filterContext);
-    assertThat(configOrError.errorDetail).isNull();
-    ExternalProcessorFilterConfig filterConfig = configOrError.config;
-
-    final String mutatedMessageTooLarge = new String(new char[70000]).replace('\0', 'c');
-    final CountDownLatch callClosedLatch = new CountDownLatch(1);
-    final AtomicReference<Status> capturedStatus = new AtomicReference<>();
-
-    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
-        new ExternalProcessorGrpc.ExternalProcessorImplBase() {
-          @Override
-          public StreamObserver<ProcessingRequest> process(
-              final StreamObserver<ProcessingResponse> responseObserver) {
-            ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
-            return new StreamObserver<ProcessingRequest>() {
-              @Override
-              public void onNext(ProcessingRequest request) {
-                if (request.hasRequestHeaders()) {
-                  responseObserver.onNext(ProcessingResponse.newBuilder()
-                      .setRequestHeaders(HeadersResponse.newBuilder().build())
-                      .build());
-                } else if (request.hasRequestBody()) {
-                  ByteString original = request.getRequestBody().getBody();
-                  if (original.toStringUtf8().equals("Message 1")) {
-                    responseObserver.onNext(ProcessingResponse.newBuilder()
-                        .setRequestBody(BodyResponse.newBuilder()
-                            .setResponse(CommonResponse.newBuilder()
-                                .setBodyMutation(BodyMutation.newBuilder()
-                                    .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                        .setBody(ByteString.copyFromUtf8(mutatedMessageTooLarge))
-                                        .build())
-                                    .build())
-                                .build())
-                            .build())
-                        .build());
-                  } else {
-                    responseObserver.onNext(ProcessingResponse.newBuilder()
-                        .setRequestBody(BodyResponse.newBuilder()
-                            .setResponse(CommonResponse.newBuilder()
-                                .setBodyMutation(BodyMutation.newBuilder()
-                                    .setStreamedResponse(StreamedBodyResponse.newBuilder()
-                                        .setBody(ByteString.copyFromUtf8("a"))
-                                        .build())
-                                    .build())
-                                .build())
-                            .build())
-                        .build());
-                  }
-                }
-              }
-
-              @Override
-              public void onError(Throwable t) {}
 
-              @Override
-              public void onCompleted() {
-                responseObserver.onCompleted();
-              }
-            };
-          }
-        };
-
-    String uniqueExtProcServerName = InProcessServerBuilder.generateName();
-    grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
-        .addService(extProcImpl)
-        .directExecutor()
-        .build().start());
-
-    CachedChannelManager channelManager = new CachedChannelManager(config -> {
-      return grpcCleanup.register(
-          InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
-    });
-
-    ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
-        filterConfig, channelManager, scheduler, FAKE_CONTEXT);
-
-    dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
-        .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall(
-            new ServerCalls.ClientStreamingMethod<String, String>() {
-              @Override
-              public StreamObserver<String> invoke(StreamObserver<String> responseObserver) {
-                return new StreamObserver<String>() {
-                  @Override
-                  public void onNext(String value) {}
-
-                  @Override
-                  public void onError(Throwable t) {}
-
-                  @Override
-                  public void onCompleted() {
-                    responseObserver.onNext("Response");
-                    responseObserver.onCompleted();
-                  }
-                };
-              }
-            }))
-        .build());
-
-    // Build the data plane channel with a custom ClientInterceptor.
-    // This interceptor overrides isReady() to always return false.
-    // This simulates a blocked backend server
-    // (transport flow control buffer is full) and blocks the filter's upstream
-    // window replenishment.
-    ManagedChannel dataPlaneChannel = grpcCleanup.register(
-        InProcessChannelBuilder.forName(dataPlaneServerName)
-            .intercept(new ClientInterceptor() {
-              @Override
-              public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
-                  MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
-                return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
-                    next.newCall(method, callOptions)) {
-                  @Override
-                  public boolean isReady() {
-                    return false;
-                  }
-                };
-              }
-            })
-            .directExecutor()
-            .build());
-
-    ClientCall<String, String> proxyCall =
-        interceptCall(interceptor, METHOD_CLIENT_STREAMING,
-            DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), dataPlaneChannel);
-
-    proxyCall.start(new ClientCall.Listener<String>() {
-      @Override
-      public void onClose(Status status, Metadata trailers) {
-        capturedStatus.set(status);
-        callClosedLatch.countDown();
-      }
-    }, new Metadata());
-
-    // Send first message. Window becomes negative.
-    proxyCall.sendMessage("Message 1");
-
-    // Send second message. Window is still negative, triggering flow control violation.
-    proxyCall.sendMessage("Message 2");
-
-    assertThat(callClosedLatch.await(5, TimeUnit.SECONDS)).isTrue();
-
-    // The call should fail immediately with INTERNAL error code due to flow control violation
-    assertThat(capturedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
-    assertThat(capturedStatus.get().getDescription())
-
-        .isEqualTo("External processor stream failed");
-    assertThat(capturedStatus.get().getCause()).isInstanceOf(io.grpc.StatusRuntimeException.class);
-    assertThat(capturedStatus.get().getCause().getMessage())
-        .contains("Flow control violation: received client body from ext_proc "
-            + "when window is closed");
-
-    channelManager.close();
-  }
 
   @Test
   @SuppressWarnings("unchecked")