From 55a8b5770c147d69cc0bb2ce3d9babc8a3c27169 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Thu, 29 Oct 2026 08:18:54 +0000 Subject: [PATCH] xds: Fix ext_proc client deadlock on unary and server-streaming RPCs when upstream is not ready Outbound request body messages in ExternalProcessorClientInterceptor were gated on super.isReady() before being forwarded to the upstream ClientCall. On unary and server-streaming RPCs (method.getType().clientSendsOneMessage()), ClientCallImpl intentionally suppresses onReady() notifications, meaning queued messages and halfClose were never drained when the upstream transport stream connected, causing the RPC to hang indefinitely. Treat the upstream call as ready for outbound message forwarding when method.getType().clientSendsOneMessage() is true via isUpstreamReady(). TAG=agy CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0 --- .../ExternalProcessorClientInterceptor.java | 14 +- ...xternalProcessorClientInterceptorTest.java | 824 ++++++++++++++++++ 2 files changed, 833 insertions(+), 5 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 93a2230b9a4..857e08ecd96 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -934,6 +934,10 @@ boolean isSidecarReady() { return observer != null && observer.isReady(); } + private boolean isUpstreamReady() { + return method.getType().clientSendsOneMessage() || super.isReady(); + } + @Override public boolean isReady() { if (passThroughMode.get()) { @@ -1193,7 +1197,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { boolean sendImmediately = false; synchronized (streamLock) { sidestreamToUpstreamWindow -= body.size(); - if (pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { sendImmediately = true; accumulatedWindowUpdateSidestreamToUpstream += body.size(); } else { @@ -1307,7 +1311,7 @@ void drainPendingUpstreamBodyMessages() { ByteString body = null; boolean triggerHalfClose = false; synchronized (streamLock) { - if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { body = pendingUpstreamBodyMessages.poll(); accumulatedWindowUpdateSidestreamToUpstream += body.size(); if (pendingUpstreamBodyMessages.isEmpty() @@ -1366,16 +1370,16 @@ private void drainPendingDrainingMessages() { boolean triggerHalfClose = false; synchronized (streamLock) { - if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) { + if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) { msg = pendingUpstreamBodyMessages.poll(); isMutated = true; } else if (pendingUpstreamBodyMessages.isEmpty() - && !pendingRequestBodyMessages.isEmpty() && super.isReady()) { + && !pendingRequestBodyMessages.isEmpty() && isUpstreamReady()) { msg = pendingRequestBodyMessages.poll(); isMutated = true; } else if (pendingUpstreamBodyMessages.isEmpty() && pendingRequestBodyMessages.isEmpty() - && !pendingDrainingMessages.isEmpty() && super.isReady()) { + && !pendingDrainingMessages.isEmpty() && isUpstreamReady()) { msg = pendingDrainingMessages.poll(); isMutated = false; } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 48f0931cbc5..19ae0940bd7 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -50,6 +50,7 @@ import io.grpc.Context; import io.grpc.Deadline; import io.grpc.ForwardingClientCall.SimpleForwardingClientCall; +import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener; import io.grpc.ManagedChannel; import io.grpc.Metadata; import io.grpc.MethodDescriptor; @@ -154,6 +155,14 @@ public class ExternalProcessorClientInterceptorTest { .setResponseMarshaller(new StringMarshaller()) .build(); + private static final MethodDescriptor METHOD_SERVER_STREAMING = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("test.TestService/ServerStreaming") + .setRequestMarshaller(new StringMarshaller()) + .setResponseMarshaller(new StringMarshaller()) + .build(); + private static final MethodDescriptor METHOD_BIDI_STREAMING = MethodDescriptor.newBuilder() .setType(MethodDescriptor.MethodType.BIDI_STREAMING) @@ -22397,6 +22406,821 @@ public void cancel(String message, Throwable cause) { shutdownAndAwaitTermination(executor); } + // --- Category 30: Unary and Server-Streaming Outbound Flow Control --- + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenUpstreamNotReady_messageAndHalfCloseForwardedImmediately() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @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(ByteString.copyFromUtf8("mutated-ping")) + .setEndOfStream(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + 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 receivedServerRequest = new AtomicReference<>(); + final AtomicBoolean serverReceivedHalfClose = new AtomicBoolean(false); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequest.set(request); + serverReceivedHalfClose.set(true); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + // Interceptor simulating DelayedStream / stream allocation delay where isReady() returns false + // and onReady() is suppressed (matching ClientCallImpl behavior for clientSendsOneMessage()). + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() { + // Suppress onReady matching ClientCallImpl behavior for clientSendsOneMessage() + } + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientReceivedResponse = new AtomicReference<>(); + final AtomicReference clientReceivedStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientReceivedResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientReceivedStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(receivedServerRequest.get()).isEqualTo("mutated-ping"); + assertThat(serverReceivedHalfClose.get()).isTrue(); + assertThat(clientReceivedResponse.get()).isEqualTo("Hello mutated-ping"); + assertThat(clientReceivedStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenExtProcSendsEndOfStreamWithoutMessage_whenUpstreamNotReady_succeeds() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + // Response 1: body message without end_of_stream + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("chunk-ping")) + .setEndOfStream(false) + .build()) + .build()) + .build()) + .build()) + .build()); + // Response 2: end_of_stream without message + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setEndOfStream(true) + .setEndOfStreamWithoutMessage(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + 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 List receivedServerRequests = new ArrayList<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequests.add(request); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + final AtomicReference clientStatus = new AtomicReference<>(); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequests).containsExactly("chunk-ping"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void unaryCall_whenExtProcDraining_whenUpstreamNotReady_drainingMessagesFlushed() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .setRequestDrain(true) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + 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 receivedServerRequest = new AtomicReference<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> { + receivedServerRequest.set(request); + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SAY_HELLO, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientResponse = new AtomicReference<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + proxyCall.sendMessage("unmutated-ping"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequest.get()).isEqualTo("unmutated-ping"); + assertThat(clientResponse.get()).isEqualTo("Hello unmutated-ping"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void serverStreamingCall_whenUpstreamNotReady_messageAndHalfCloseForwarded() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @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(ByteString.copyFromUtf8("stream-query")) + .setEndOfStream(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + 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 receivedServerQuery = new AtomicReference<>(); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SERVER_STREAMING, + ServerCalls.asyncServerStreamingCall((request, responseObserver) -> { + receivedServerQuery.set(request); + responseObserver.onNext("Item 1 for " + request); + responseObserver.onNext("Item 2 for " + request); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + ClientInterceptor unreadyUpstreamInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return false; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + super.start(new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onReady() {} + }, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(unreadyUpstreamInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_SERVER_STREAMING, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final List clientItems = new ArrayList<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientItems.add(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(10); + proxyCall.sendMessage("query"); + proxyCall.halfClose(); + + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerQuery.get()).isEqualTo("stream-query"); + assertThat(clientItems) + .containsExactly("Item 1 for stream-query", "Item 2 for stream-query"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void clientStreamingCall_whenUpstreamNotReady_messageGatedUntilOnReady() + throws Exception { + String uniqueExtProcServerName = InProcessServerBuilder.generateName(); + String uniqueDataPlaneServerName = InProcessServerBuilder.generateName(); + + ExternalProcessor proto = createBaseProto(uniqueExtProcServerName) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND) + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SKIP) + .setResponseBodyMode(ProcessingMode.BodySendMode.NONE) + .build()) + .build(); + ExternalProcessorFilterConfig filterConfig = + provider.parseFilterConfig(Any.pack(proto), filterContext).config; + + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + public StreamObserver process( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } else if (request.hasRequestBody()) { + boolean eosWithoutMsg = + request.getRequestBody().getEndOfStreamWithoutMessage(); + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(request.getRequestBody().getBody()) + .setEndOfStream(request.getRequestBody().getEndOfStream()) + .setEndOfStreamWithoutMessage(eosWithoutMsg) + .build()) + .build()) + .build()) + .build()) + .build()); + } else if (request.hasResponseHeaders()) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + 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 List receivedServerRequests = new ArrayList<>(); + final CountDownLatch serverReceivedFirstMessageLatch = new CountDownLatch(1); + final CountDownLatch serverDoneLatch = new CountDownLatch(1); + + MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry(); + dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, + ServerCalls.asyncClientStreamingCall(responseObserver -> { + return new StreamObserver() { + @Override + public void onNext(String value) { + receivedServerRequests.add(value); + serverReceivedFirstMessageLatch.countDown(); + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Total: " + receivedServerRequests.size()); + responseObserver.onCompleted(); + serverDoneLatch.countDown(); + } + }; + })) + .build()); + + grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName) + .fallbackHandlerRegistry(dataPlaneRegistry) + .directExecutor() + .build().start()); + + final AtomicBoolean upstreamIsReady = new AtomicBoolean(false); + final AtomicReference> capturedListener = new AtomicReference<>(); + + ClientInterceptor controllableInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public boolean isReady() { + return upstreamIsReady.get(); + } + + @Override + public void start(Listener responseListener, Metadata headers) { + capturedListener.set(responseListener); + super.start(responseListener, headers); + } + }; + } + }; + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(uniqueDataPlaneServerName) + .intercept(controllableInterceptor) + .directExecutor() + .build()); + + ClientCall proxyCall = interceptCall( + interceptor, + METHOD_CLIENT_STREAMING, + DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()), + dataPlaneChannel); + + final AtomicReference clientResponse = new AtomicReference<>(); + final AtomicReference clientStatus = new AtomicReference<>(); + final CountDownLatch clientDoneLatch = new CountDownLatch(1); + + proxyCall.start(new ClientCall.Listener() { + @Override + public void onMessage(String message) { + clientResponse.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + clientStatus.set(status); + clientDoneLatch.countDown(); + } + }, new Metadata()); + + proxyCall.request(1); + // Send message while upstream is NOT ready + proxyCall.sendMessage("msg1"); + + // Give ext-proc time to process and return BodyResponse + // Upstream should NOT have received it yet because isReady is false for client streaming + assertThat(serverReceivedFirstMessageLatch.await(200, TimeUnit.MILLISECONDS)).isFalse(); + assertThat(receivedServerRequests).isEmpty(); + + // Now make upstream ready and trigger onReady() + upstreamIsReady.set(true); + capturedListener.get().onReady(); + + // Now msg1 should be flushed to server + assertThat(serverReceivedFirstMessageLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(receivedServerRequests).containsExactly("msg1"); + + proxyCall.halfClose(); + assertThat(serverDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientDoneLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(clientResponse.get()).isEqualTo("Total: 1"); + assertThat(clientStatus.get().getCode()).isEqualTo(Status.Code.OK); + + channelManager.close(); + } + private static List filterClientRequests(List requests) { List clientRequests = new ArrayList<>(); for (ProcessingRequest r : requests) {