From aefc6d0c005c79707795a73b69b81160f575cee5 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Thu, 27 Aug 2026 05:47:48 +0000 Subject: [PATCH 1/9] Implement new draining logic as per updates in grfc. --- .../ExternalProcessorClientInterceptor.java | 158 +++- ...xternalProcessorClientInterceptorTest.java | 756 ++++++++++++++---- .../ext_proc/v3/external_processor.proto | 592 +++++++++----- 3 files changed, 1147 insertions(+), 359 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 0bc79e5ec5d..d83b3a39326 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -316,6 +316,14 @@ private static class DataPlaneClientCall final AtomicBoolean pendingHalfClose = new AtomicBoolean(false); final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false); private final AtomicBoolean downstreamCancelled = new AtomicBoolean(false); + final AtomicBoolean requestDraining = new AtomicBoolean(false); + final AtomicBoolean responseDraining = new AtomicBoolean(false); + final AtomicBoolean requestDrainComplete = new AtomicBoolean(false); + final AtomicBoolean responseDrainComplete = new AtomicBoolean(false); + final AtomicBoolean requestEosSent = new AtomicBoolean(false); + final AtomicBoolean responseTrailersSent = new AtomicBoolean(false); + final AtomicBoolean requestBodySentToExtProc = new AtomicBoolean(false); + final AtomicBoolean responseBodySentToExtProc = new AtomicBoolean(false); protected DataPlaneClientCall( DataPlaneDelayedCall delayedCall, @@ -502,10 +510,25 @@ public void onNext(ProcessingResponse response) { } } - if (response.getRequestDrain()) { - extProcStreamState.set(ExtProcStreamState.DRAINING); - halfCloseExtProcStream(); - activateCall(); + if (response.getRequestDrainRequests()) { + if (!requestEosSent.get() && requestDraining.compareAndSet(false, true)) { + activateCall(); + sendToExtProc(ProcessingRequest.newBuilder() + .setRequestBody(HttpBody.newBuilder() + .setDrainComplete(true) + .build()) + .build()); + } + } + if (response.getRequestDrainResponses()) { + if (!responseTrailersSent.get() && responseDraining.compareAndSet(false, true)) { + activateCall(); + sendToExtProc(ProcessingRequest.newBuilder() + .setResponseBody(HttpBody.newBuilder() + .setDrainComplete(true) + .build()) + .build()); + } } // 1. Client Headers @@ -599,6 +622,29 @@ public void onError(Throwable t) { @Override public void onCompleted() { + boolean requestDrainRequired = currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.GRPC + && requestBodySentToExtProc.get() + && !requestDrainComplete.get() + && !requestEosSent.get(); + + boolean responseDrainRequired = currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC + && responseBodySentToExtProc.get() + && !responseDrainComplete.get() + && !responseTrailersSent.get(); + + if (!config.getObservabilityMode() && (requestDrainRequired || responseDrainRequired)) { + if (markExtProcStreamFailed(extProcStreamState)) { + synchronized (streamLock) { + extProcClientCallRequestObserver = null; + } + cancelDownstream("External processor stream completed without drain", + Status.INTERNAL.withDescription("External processor stream completed without drain") + .asRuntimeException()); + wrappedListener.proceedWithClose(); + } + return; + } + if (markExtProcStreamCompleted(extProcStreamState)) { handleFailOpen(wrappedListener); } @@ -719,24 +765,32 @@ private void internalOnError(Throwable t) { } } - private void halfCloseExtProcStream() { - synchronized (streamLock) { - if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) { - extProcClientCallRequestObserver.onCompleted(); - } - } - } + private void onReadyNotify() { wrappedListener.onReadyNotify(); } - private boolean isSidecarReady() { + private boolean isRequestSidecarReady() { + ExtProcStreamState state = extProcStreamState.get(); + if (state.isCompleted()) { + return true; + } + if (requestDraining.get()) { + return false; + } + synchronized (streamLock) { + ClientCallStreamObserver observer = extProcClientCallRequestObserver; + return observer != null && observer.isReady(); + } + } + + private boolean isResponseSidecarReady() { ExtProcStreamState state = extProcStreamState.get(); if (state.isCompleted()) { return true; } - if (state.isDraining()) { + if (responseDraining.get()) { return false; } synchronized (streamLock) { @@ -756,7 +810,7 @@ public boolean isReady() { if (dataPlaneCallState.get() == DataPlaneCallState.IDLE && !config.getObservabilityMode()) { return false; } - boolean sidecarReady = isSidecarReady(); + boolean sidecarReady = isRequestSidecarReady(); if (config.getObservabilityMode()) { return super.isReady() && sidecarReady; } @@ -774,7 +828,7 @@ public void request(int numMessages) { super.request(numMessages); return; } - if (!isSidecarReady()) { + if (!isResponseSidecarReady()) { pendingRequests.addAndGet(numMessages); return; } @@ -788,19 +842,19 @@ public void sendMessage(InputStream message) { return; } - if (passThroughMode.get()) { + if (passThroughMode.get() || requestDrainComplete.get()) { super.sendMessage(message); return; } synchronized (streamLock) { - if (passThroughMode.get()) { + if (passThroughMode.get() || requestDrainComplete.get()) { super.sendMessage(message); return; } ExtProcStreamState state = extProcStreamState.get(); - if (state.isDraining() || state.isCompleted()) { + if (requestDraining.get() || state.isCompleted()) { if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { super.sendMessage(message); return; @@ -829,6 +883,7 @@ public void sendMessage(InputStream message) { .setEndOfStream(false) .build()) .build()); + requestBodySentToExtProc.set(true); bodyMessageSentToExtProc.set(true); if (config.getObservabilityMode()) { @@ -851,7 +906,7 @@ private void proceedWithHalfClose() { @Override public void halfClose() { clientHalfCloseStartNanos = System.nanoTime(); - if (passThroughMode.get()) { + if (passThroughMode.get() || requestDrainComplete.get()) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } @@ -869,7 +924,7 @@ public void halfClose() { return; } - if (extProcStreamState.get().isDraining()) { + if (requestDraining.get()) { boolean canProceed = false; synchronized (streamLock) { if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE @@ -898,6 +953,7 @@ public void halfClose() { .setEndOfStreamWithoutMessage(true) .build()) .build()); + requestEosSent.set(true); } private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) { @@ -925,7 +981,9 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) { BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - if (!streamed.getEndOfStreamWithoutMessage()) { + if (streamed.getDrainComplete()) { + handleRequestDrainComplete(); + } else if (!streamed.getEndOfStreamWithoutMessage()) { super.sendMessage(new KnownLengthInputStream(streamed.getBody())); } if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) { @@ -943,7 +1001,11 @@ private void handleResponseBodyResponse( BodyMutation mutation = bodyResponse.getResponse().getBodyMutation(); if (mutation.hasStreamedResponse()) { StreamedBodyResponse streamed = mutation.getStreamedResponse(); - listener.onExternalBody(streamed.getBody()); + if (streamed.getDrainComplete()) { + handleResponseDrainComplete(); + } else { + listener.onExternalBody(streamed.getBody()); + } } } } @@ -991,6 +1053,20 @@ private void drainPendingDrainingMessages() { } } + private void handleRequestDrainComplete() { + if (requestDraining.compareAndSet(true, false)) { + requestDrainComplete.set(true); + drainPendingDrainingMessages(); + } + } + + private void handleResponseDrainComplete() { + if (responseDraining.compareAndSet(true, false)) { + responseDrainComplete.set(true); + wrappedListener.unblockAfterResponseDrain(); + } + } + private void handleFailOpen(DataPlaneListener listener) { activateCall(); drainPendingRequests(); @@ -1111,7 +1187,8 @@ public void onHeaders(Metadata headers) { || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() == ProcessingMode.HeaderSendMode.DEFAULT; - if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseHeaders) { + + if (dataPlaneClientCall.responseDraining.get() && sendResponseHeaders) { this.savedHeaders = headers; return; } @@ -1144,7 +1221,7 @@ public void onMessage(InputStream message) { return; } - boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining() + boolean checkDrain = dataPlaneClientCall.responseDraining.get() && dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC; @@ -1174,6 +1251,7 @@ public void onMessage(InputStream message) { try { ByteString bodyByteString = ByteString.readFrom(message); sendResponseBodyToExtProc(bodyByteString, false); + dataPlaneClientCall.responseBodySentToExtProc.set(true); dataPlaneClientCall.bodyMessageSentToExtProc.set(true); if (dataPlaneClientCall.getConfig().getObservabilityMode()) { @@ -1198,8 +1276,13 @@ public void onClose(Status status, Metadata trailers) { && (!dataPlaneClientCall.getConfig().getFailureModeAllow() || dataPlaneClientCall.bodyMessageSentToExtProc.get())) { if (markDataPlaneCallClosed(dataPlaneClientCall.dataPlaneCallState)) { - proceedWithClose(Status.INTERNAL.withDescription("External processor stream failed") - .withCause(status.getCause()), new Metadata()); + Status finalStatus = Status.INTERNAL.withCause(status.getCause()); + if (status.getDescription() != null) { + finalStatus = finalStatus.withDescription(status.getDescription()); + } else { + finalStatus = finalStatus.withDescription("External processor stream failed"); + } + proceedWithClose(finalStatus, new Metadata()); } return; } @@ -1225,7 +1308,7 @@ public void onClose(Status status, Metadata trailers) { dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() == ProcessingMode.HeaderSendMode.SEND; - if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseTrailers) { + if (dataPlaneClientCall.responseDraining.get() && sendResponseTrailers) { return; } @@ -1245,7 +1328,7 @@ void proceedWithHeaders() { proceedWithHeaders(savedHeaders); synchronized (savedMessages) { savedHeaders = null; - if (!dataPlaneClientCall.getExtProcStreamState().get().isDraining()) { + if (!dataPlaneClientCall.responseDraining.get()) { InputStream msg; while ((msg = savedMessages.poll()) != null) { onMessage(msg); @@ -1302,6 +1385,15 @@ void unblockAfterStreamComplete() { proceedWithClose(); } + void unblockAfterResponseDrain() { + synchronized (savedMessages) { + inboundPassThrough = true; + } + proceedWithHeaders(); + proceedWithSavedMessages(); + proceedWithClose(); + } + private void proceedWithSavedMessages() { synchronized (savedMessages) { InputStream msg; @@ -1319,6 +1411,14 @@ private void triggerCloseHandshake() { return; } + if (dataPlaneClientCall.responseDrainComplete.get()) { + proceedWithClose(); + if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { + dataPlaneClientCall.closeExtProcStream(); + } + return; + } + boolean sendResponseHeaders = dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() == ProcessingMode.HeaderSendMode.SEND @@ -1340,6 +1440,7 @@ private void triggerCloseHandshake() { .setEndOfStream(true) .build()) .build()); + dataPlaneClientCall.responseTrailersSent.set(true); } else { proceedWithClose(); if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { @@ -1356,6 +1457,7 @@ private void triggerCloseHandshake() { dataPlaneClientCall.getConfig().getForwardRulesConfig())) .build()) .build()); + dataPlaneClientCall.responseTrailersSent.set(true); } else { proceedWithClose(); if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index cd6de138a48..8447ea150d7 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -6159,7 +6159,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainRequests(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6343,7 +6343,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainSentLatch.countDown(); } @@ -6456,7 +6456,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainSentLatch.countDown(); } @@ -6555,6 +6555,7 @@ public void testHalfCloseDeferredWhenDrainingAndMessagesSent() throws Exception final CountDownLatch headersReceivedLatch = new CountDownLatch(1); final CountDownLatch bodyReceivedLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final AtomicReference> extProcResponseObserverRef = new AtomicReference<>(); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = @@ -6568,17 +6569,24 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestHeaders(HeadersResponse.newBuilder().build()) - .build()); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } headersReceivedLatch.countDown(); } else if (request.hasRequestBody()) { - // Respond to body with request_drain = true - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder().build()) - .setRequestDrain(true) - .build()); - bodyReceivedLatch.countDown(); + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder().build()) + .setRequestDrainRequests(true) + .build()); + } + bodyReceivedLatch.countDown(); + } } } @@ -6649,14 +6657,34 @@ public void onCompleted() { // Verify downstream server has NOT received half-close yet. assertThat(serverHalfClosedLatch.await(1, TimeUnit.SECONDS)).isFalse(); - // Now complete the ext_proc stream. - if (extProcResponseObserverRef.get() != null) { - extProcResponseObserverRef.get().onCompleted(); + // Wait for filter to send drain_complete + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now complete the draining handshake by echoing drain_complete + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); } // Verify downstream server now receives half-close. assertThat(serverHalfClosedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // Cleanup ext_proc stream + synchronized (extProcRespObserver) { + extProcRespObserver.onCompleted(); + } + channelManager.close(); } @@ -6697,7 +6725,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6824,7 +6852,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -6945,7 +6973,7 @@ public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) // Trigger Request Drain + .setRequestDrainResponses(true) // Trigger Request Drain .build()); sidecarActionLatch.countDown(); } @@ -7063,7 +7091,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainLatch.countDown(); } @@ -7169,7 +7197,7 @@ public void onNext(ProcessingRequest request) { new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } sidecarOnNextLatch.countDown(); @@ -7328,7 +7356,7 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); drainLatch.countDown(); } @@ -7437,7 +7465,7 @@ public void onNext(ProcessingRequest request) { new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } try { @@ -7582,8 +7610,9 @@ public void onMessage(String message) { ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); - final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final AtomicInteger extProcReceivedBodyCount = new AtomicInteger(0); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -7591,6 +7620,7 @@ public void onMessage(String message) { @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -7600,21 +7630,16 @@ public void onNext(ProcessingRequest request) { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainRequests(true) .build()); } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } }).start(); } else if (request.hasRequestBody()) { - extProcReceivedBodyCount.incrementAndGet(); + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + extProcReceivedBodyCount.incrementAndGet(); + } } } @@ -7624,7 +7649,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -7690,7 +7714,7 @@ public void onMessage(String message) { proxyCall.start(appListener, new Metadata()); // Wait for drain to be processed - assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(proxyCall.isReady()).isFalse(); // Send message and half-close concurrently during drain state @@ -7707,7 +7731,23 @@ public void onMessage(String message) { assertThat(extProcReceivedBodyCount.get()).isEqualTo(0); // Now let sidecar complete - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Request response from data plane proxyCall.request(1); @@ -7753,7 +7793,9 @@ public void drainingStartsAfterRequestHeaders_whenAppSendsAndHalfCloses_thenBuff ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); final CountDownLatch drainCompletedLatch = new CountDownLatch(1); final CountDownLatch extProcReceivedBodyLatch = new CountDownLatch(1); final CountDownLatch mutatedBodyDeliveredLatch = new CountDownLatch(1); @@ -7764,6 +7806,7 @@ public void drainingStartsAfterRequestHeaders_whenAppSendsAndHalfCloses_thenBuff @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -7775,32 +7818,27 @@ public void onNext(ProcessingRequest request) { .build()); } } else if (request.hasRequestBody()) { - extProcReceivedBodyLatch.countDown(); - new Thread(() -> { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8("Mutated Message 1")) - .build()) - .build()) - .build()) - .build()) - .setRequestDrain(true) - .build()); - } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + extProcReceivedBodyLatch.countDown(); + new Thread(() -> { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8("Mutated Message 1")) + .build()) + .build()) + .build()) + .build()) + .setRequestDrainRequests(true) + .build()); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); + }).start(); + } } } @@ -7810,7 +7848,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -7887,6 +7924,7 @@ public void onMessage(String message) { assertThat(mutatedBodyDeliveredLatch.await(5, TimeUnit.SECONDS)).isTrue(); // Verify proxyCall is in draining (i.e. isReady is false) + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(proxyCall.isReady()).isFalse(); // Send message and half-close concurrently during drain state @@ -7903,7 +7941,24 @@ public void onMessage(String message) { assertThat(dataPlaneReceivedMessages).containsExactly("Mutated Message 1"); // Now let sidecar complete - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + drainCompletedLatch.countDown(); + } // Wait for the control stream drain to be fully completed assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -7952,36 +8007,36 @@ public void drainingStartsBeforeResponseHeaders_whenUpstreamResponds_thenBuffere ExternalProcessorFilterConfig filterConfig = configOrError.config; // External Processor Server - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); - final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @Override @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock: Received request: " + request); if (request.hasRequestHeaders()) { + System.out.println("JetskiMock: Received request headers, triggering drain"); new Thread(() -> { synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() .setRequestHeaders(HeadersResponse.newBuilder().build()) - .setRequestDrain(true) + .setRequestDrainResponses(true) .build()); } - try { - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); - } - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } }).start(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + System.out.println("JetskiMock: Received drain_complete from filter"); + filterSentDrainLatch.countDown(); + } } } @@ -7991,7 +8046,6 @@ public void onError(Throwable t) { @Override public void onCompleted() { - drainCompletedLatch.countDown(); } }; } @@ -8071,27 +8125,49 @@ public void onClose(Status status, Metadata trailers) { // Request messages from server proxyCall.request(10); - // Wait for drain to be processed and sidecar's client stream to finish - assertThat(drainCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue(); - assertThat(proxyCall.isReady()).isFalse(); + // Wait for drain to be processed + System.out.println("JetskiTest: waiting for filterSentDrainLatch"); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); + System.out.println("JetskiTest: filterSentDrainLatch counted down"); + assertThat(proxyCall.isReady()).isTrue(); // Verify the data plane call has started + System.out.println("JetskiTest: waiting for dataPlaneCallStartedLatch"); assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + System.out.println("JetskiTest: dataPlaneCallStartedLatch counted down"); StreamObserver dataPlaneResponseObserver = dataPlaneResponseObserverRef.get(); assertThat(dataPlaneResponseObserver).isNotNull(); // Upstream server sends response headers, response message, and closes call during drain + System.out.println("JetskiTest: sending response headers/message"); dataPlaneResponseObserver.onNext("Response Message During Drain"); dataPlaneResponseObserver.onCompleted(); // Verify app listener has NOT received headers, messages, or close yet because the drain // is active + System.out.println("JetskiTest: verifying app hasn't received anything"); assertThat(appReceivedHeaders.get()).isNull(); assertThat(appReceivedMessages).isEmpty(); assertThat(appReceivedStatus.get()).isNull(); // Now let sidecar complete the drain - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Wait for the call to close on application side assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -8136,8 +8212,10 @@ public void drainingStartsAfterResponseHeaders_whenUpstreamResponds_thenBuffered final CountDownLatch respBody1Latch = new CountDownLatch(1); final CountDownLatch respBody2Latch = new CountDownLatch(1); final CountDownLatch m3SentLatch = new CountDownLatch(1); - final CountDownLatch sidecarFinishLatch = new CountDownLatch(1); final CountDownLatch drainCompletedLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = @@ -8146,6 +8224,7 @@ public void drainingStartsAfterResponseHeaders_whenUpstreamResponds_thenBuffered @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -8165,64 +8244,63 @@ public void onNext(ProcessingRequest request) { } respHeadersLatch.countDown(); } else if (request.hasResponseBody()) { - String msgStr = request.getResponseBody().getBody().toStringUtf8(); - if ("Original Message 1".equals(msgStr)) { - new Thread(() -> { - try { - // Wait until M2 is received by sidecar so both M1 and M2 are in flight - if (m2ReceivedLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setResponseBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8( - "Mutated Message 1")) - .build()) - .build()) - .build()) - .build()) - .setRequestDrain(true) - .build()); - } - respBody1Latch.countDown(); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); - } else if ("Original Message 2".equals(msgStr)) { - m2ReceivedLatch.countDown(); - new Thread(() -> { - try { - // Wait until M3 is sent by upstream concurrently during drain - if (m3SentLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setResponseBody(BodyResponse.newBuilder() - .setResponse(CommonResponse.newBuilder() - .setBodyMutation(BodyMutation.newBuilder() - .setStreamedResponse(StreamedBodyResponse.newBuilder() - .setBody(ByteString.copyFromUtf8( - "Mutated Message 2")) - .build()) - .build()) - .build()) - .build()) - .build()); + if (request.getResponseBody().getDrainComplete()) { + filterSentDrainLatch.countDown(); + } else { + String msgStr = request.getResponseBody().getBody().toStringUtf8(); + if ("Original Message 1".equals(msgStr)) { + new Thread(() -> { + try { + // Wait until M2 is received by sidecar so both M1 and M2 are in flight + if (m2ReceivedLatch.await(5, TimeUnit.SECONDS)) { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8( + "Mutated Message 1")) + .build()) + .build()) + .build()) + .build()) + .setRequestDrainResponses(true) + .build()); + } + respBody1Latch.countDown(); } - respBody2Latch.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - if (sidecarFinishLatch.await(5, TimeUnit.SECONDS)) { - synchronized (responseObserver) { - responseObserver.onCompleted(); + }).start(); + } else if ("Original Message 2".equals(msgStr)) { + m2ReceivedLatch.countDown(); + new Thread(() -> { + try { + // Wait until M3 is sent by upstream concurrently during drain + if (m3SentLatch.await(5, TimeUnit.SECONDS)) { + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(ByteString.copyFromUtf8( + "Mutated Message 2")) + .build()) + .build()) + .build()) + .build()) + .build()); + } + respBody2Latch.countDown(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }).start(); + }).start(); + } } } } @@ -8339,6 +8417,7 @@ public void onClose(Status status, Metadata trailers) { // Verify the stream is currently in DRAINING state and app has only received Mutated // Message 1 so far assertThat(appReceivedMessages).containsExactly("Mutated Message 1"); + assertThat(filterSentDrainLatch.await(5, TimeUnit.SECONDS)).isTrue(); // 3. Upstream concurrently sends M3 and completes the call during draining dataPlaneResponseObserver.onNext("Original Message 3"); @@ -8360,7 +8439,23 @@ public void onClose(Status status, Metadata trailers) { assertThat(appReceivedMessages).containsExactly("Mutated Message 1", "Mutated Message 2"); // 5. Complete sidecar stream to finish the drain - sidecarFinishLatch.countDown(); + System.out.println("JetskiTest: completing drain from test"); + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + extProcRespObserver.onCompleted(); + } // Wait for the call to close on application side assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -8378,6 +8473,369 @@ public void onClose(Status status, Metadata trailers) { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void drainIgnored_whenRequestEosSent() throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .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 configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final CountDownLatch reqEosReceivedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock1: Received request: " + request); + if (request.hasRequestHeaders()) { + System.out.println("JetskiMock1: Received request headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasRequestBody()) { + if (request.getRequestBody().getEndOfStreamWithoutMessage() + || request.getRequestBody().getEndOfStream()) { + System.out.println("JetskiMock1: Received request EOS"); + reqEosReceivedLatch.countDown(); + } + if (request.getRequestBody().getDrainComplete()) { + System.out.println("JetskiMock1: Received drain_complete"); + filterSentDrainLatch.countDown(); + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final CountDownLatch dataPlaneCallStartedLatch = new CountDownLatch(1); + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_CLIENT_STREAMING, ServerCalls.asyncClientStreamingCall( + new ServerCalls.ClientStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + dataPlaneCallStartedLatch.countDown(); + return new StreamObserver() { + @Override + public void onNext(String value) {} + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() { + responseObserver.onNext("Direct Response"); + responseObserver.onCompleted(); + } + }; + } + })) + .build()); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build()); + + final AtomicReference appReceivedMessage = new AtomicReference<>(); + final CountDownLatch appLatch = new CountDownLatch(1); + ClientCall.Listener appListener = new ClientCall.Listener() { + @Override + public void onMessage(String message) { + appReceivedMessage.set(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + appLatch.countDown(); + } + }; + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_CLIENT_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(appListener, new Metadata()); + + // Send request headers by requesting message (starts call) + proxyCall.request(1); + + // Wait for the data plane call to start + assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Send app message and half-close (sends EOS) + proxyCall.sendMessage("App Message"); + proxyCall.halfClose(); + + // Wait for mock server to receive EOS + assertThat(reqEosReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now, mock server sends drain request AFTER EOS was already sent by client + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainRequests(true) + .build()); + } + + // Verify that filter does NOT send drain_complete (latch should timeout) + assertThat(filterSentDrainLatch.await(2, TimeUnit.SECONDS)).isFalse(); + + // Complete the call + synchronized (extProcRespObserver) { + extProcRespObserver.onCompleted(); + } + + assertThat(appLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(appReceivedMessage.get()).isEqualTo("Direct Response"); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void drainIgnored_whenResponseTrailersSent() throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .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 configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch filterSentDrainLatch = new CountDownLatch(1); + final CountDownLatch trailersReceivedLatch = new CountDownLatch(1); + final AtomicReference> extProcResponseObserverRef = + new AtomicReference<>(); + + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl = + new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + extProcResponseObserverRef.set(responseObserver); + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + System.out.println("JetskiMock2: Received request: " + request); + if (request.hasRequestHeaders()) { + System.out.println("JetskiMock2: Received request headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasResponseHeaders()) { + System.out.println("JetskiMock2: Received response headers"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseHeaders(HeadersResponse.newBuilder().build()) + .build()); + } + } else if (request.hasResponseTrailers()) { + System.out.println("JetskiMock2: Received response trailers"); + trailersReceivedLatch.countDown(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + System.out.println("JetskiMock2: Received drain_complete"); + filterSentDrainLatch.countDown(); + } else { + System.out.println("JetskiMock2: Received response body chunk, echoing"); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setBody(request.getResponseBody().getBody()) + .build()) + .build()) + .build()) + .build()) + .build()); + } + } + } + } + + @Override + public void onError(Throwable t) {} + + @Override + public void onCompleted() {} + }; + } + }; + + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); + }); + + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + final CountDownLatch dataPlaneCallStartedLatch = new CountDownLatch(1); + final AtomicReference> dataPlaneResponseObserverRef = + new AtomicReference<>(); + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_BIDI_STREAMING, ServerCalls.asyncBidiStreamingCall( + new ServerCalls.BidiStreamingMethod() { + @Override + public StreamObserver invoke(StreamObserver responseObserver) { + dataPlaneResponseObserverRef.set(responseObserver); + dataPlaneCallStartedLatch.countDown(); + return new StreamObserver() { + @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()); + + final List appReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>(); + final CountDownLatch appCloseLatch = new CountDownLatch(1); + ClientCall.Listener appListener = new ClientCall.Listener() { + @Override + public void onHeaders(Metadata headers) {} + + @Override + public void onMessage(String message) { + appReceivedMessages.add(message); + } + + @Override + public void onClose(Status status, Metadata trailers) { + appCloseLatch.countDown(); + } + }; + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_BIDI_STREAMING, callOptions, dataPlaneChannel); + proxyCall.start(appListener, new Metadata()); + + proxyCall.request(10); + + // Wait for the data plane call to start + assertThat(dataPlaneCallStartedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + StreamObserver dataPlaneResponseObserver = dataPlaneResponseObserverRef.get(); + assertThat(dataPlaneResponseObserver).isNotNull(); + + // Upstream server sends response and closes call (sends trailers) + dataPlaneResponseObserver.onNext("Response Message"); + dataPlaneResponseObserver.onCompleted(); + + // Wait for mock ext_proc to receive and process response trailers + assertThat(trailersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Now, mock server sends drain request AFTER trailers were already sent/processed + StreamObserver extProcRespObserver = extProcResponseObserverRef.get(); + assertThat(extProcRespObserver).isNotNull(); + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainResponses(true) + .build()); + } + + // Verify that filter does NOT send drain_complete (latch should timeout) + assertThat(filterSentDrainLatch.await(2, TimeUnit.SECONDS)).isFalse(); + + // Complete the call + synchronized (extProcRespObserver) { + extProcRespObserver.onNext(ProcessingResponse.newBuilder() + .setResponseTrailers(TrailersResponse.newBuilder().build()) + .build()); + extProcRespObserver.onCompleted(); + } + + assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(appReceivedMessages).containsExactly("Response Message"); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + // --- Category 15: Inbound Backpressure (request(n) / pendingRequests) --- @Test @@ -8539,7 +8997,7 @@ public void request(int numMessages) { @Test @SuppressWarnings("unchecked") - public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsBuffered() + public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffered() throws Exception { ExternalProcessor proto = ExternalProcessor.newBuilder() .setGrpcService(GrpcService.newBuilder() @@ -8561,6 +9019,7 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsBuffered assertThat(configOrError.errorDetail).isNull(); ExternalProcessorFilterConfig filterConfig = configOrError.config; + final CountDownLatch drainSentLatch = new CountDownLatch(1); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -8574,8 +9033,9 @@ public StreamObserver process( public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrain(true) + .setRequestDrainResponses(true) .build()); + drainSentLatch.countDown(); } } @@ -8636,19 +9096,15 @@ public void request(int numMessages) { proxyCall.start(new ClientCall.Listener() {}, new Metadata()); // Wait for drain to be processed - long startTime = System.currentTimeMillis(); - while (proxyCall.isReady() && System.currentTimeMillis() - startTime < 5000) { - Thread.sleep(10); - } - assertThat(proxyCall.isReady()).isFalse(); + assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // proxyCall.isReady() should remain true during response drain (request path is NOT draining) + assertThat(proxyCall.isReady()).isTrue(); // App requests more messages proxyCall.request(3); // Verify requests are buffered and not sent to data plane assertThat(dataPlaneRequestCount.get()).isEqualTo(0); - // proxyCall.isReady() should remain false during drain - assertThat(proxyCall.isReady()).isFalse(); proxyCall.cancel("Cleanup", null); channelManager.close(); @@ -9775,7 +10231,7 @@ public void onClose(Status status, Metadata trailers) { } assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("External processor stream failed"); + assertThat(closedStatus.get().getDescription()).contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close(); @@ -9917,7 +10373,7 @@ public void onCompleted() { // Verify application receives INTERNAL with correct description assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("External processor stream failed"); + assertThat(closedStatus.get().getDescription()).contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close(); diff --git a/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto b/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto index 1c033c08d26..1ac424b6358 100644 --- a/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto +++ b/xds/third_party/envoy/src/main/proto/envoy/service/ext_proc/v3/external_processor.proto @@ -6,6 +6,7 @@ import "envoy/config/core/v3/base.proto"; import "envoy/extensions/filters/http/ext_proc/v3/processing_mode.proto"; import "envoy/type/v3/http_status.proto"; +import "google/protobuf/any.proto"; import "google/protobuf/duration.proto"; import "google/protobuf/struct.proto"; @@ -23,37 +24,31 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE; // [#protodoc-title: External processing service] -// A service that can access and modify HTTP requests and responses -// as part of a filter chain. +// A service that can access and modify HTTP requests and responses as part of a filter chain. // The overall external processing protocol works like this: // // 1. The data plane sends to the service information about the HTTP request. -// 2. The service sends back a ProcessingResponse message that directs -// the data plane to either stop processing, continue without it, or send -// it the next chunk of the message body. -// 3. If so requested, the data plane sends the server the message body in -// chunks, or the entire body at once. In either case, the server may send -// back a ProcessingResponse for each message it receives, or wait for -// a certain amount of body chunks received before streaming back the -// ProcessingResponse messages. -// 4. If so requested, the data plane sends the server the HTTP trailers, -// and the server sends back a ProcessingResponse. -// 5. At this point, request processing is done, and we pick up again -// at step 1 when the data plane receives a response from the upstream -// server. -// 6. At any point above, if the server closes the gRPC stream cleanly, -// then the data plane proceeds without consulting the server. -// 7. At any point above, if the server closes the gRPC stream with an error, -// then the data plane returns a 500 error to the client, unless the filter -// was configured to ignore errors. +// 2. The service sends back a ``ProcessingResponse`` message that directs the data plane to either +// stop processing, continue without it, or send it the next chunk of the message body. +// 3. If so requested, the data plane sends the server the message body in chunks, or the entire +// body at once. In either case, the server may send back a ``ProcessingResponse`` for each +// message it receives, or wait for a certain amount of body chunks to be received before +// streaming back the ``ProcessingResponse`` messages. +// 4. If so requested, the data plane sends the server the HTTP trailers, and the server sends back +// a ``ProcessingResponse``. +// 5. At this point, request processing is done, and we pick up again at step 1 when the data plane +// receives a response from the upstream server. +// 6. At any point above, if the server closes the gRPC stream cleanly, then the data plane +// proceeds without consulting the server. +// 7. At any point above, if the server closes the gRPC stream with an error, then the data plane +// returns a ``500`` error to the client, unless the filter was configured to ignore errors. // -// In other words, the process is a request/response conversation, but -// using a gRPC stream to make it easier for the server to -// maintain state. +// In other words, the process is a request/response conversation, but using a gRPC stream to make +// it easier for the server to maintain state. service ExternalProcessor { // This begins the bidirectional stream that the data plane will use to // give the server control over what the filter does. The actual - // protocol is described by the ProcessingRequest and ProcessingResponse + // protocol is described by the ``ProcessingRequest`` and ``ProcessingResponse`` // messages below. rpc Process(stream ProcessingRequest) returns (stream ProcessingResponse) { } @@ -61,30 +56,90 @@ service ExternalProcessor { // This message specifies the filter protocol configurations which will be sent to the ext_proc // server in a :ref:`ProcessingRequest `. -// If the server does not support these protocol configurations, it may choose to close the gRPC stream. -// If the server supports these protocol configurations, it should respond based on the API specifications. +// If the server does not support these protocol configurations, it may choose to close the gRPC +// stream. If the server supports these protocol configurations, it should respond based on the +// API specifications. message ProtocolConfiguration { - // Specify the filter configuration :ref:`request_body_mode - // ` + // Specifies the filter configuration + // :ref:`request_body_mode `. envoy.extensions.filters.http.ext_proc.v3.ProcessingMode.BodySendMode request_body_mode = 1 - [(validate.rules).enum = {defined_only: true}]; + [(validate.rules).enum = {defined_only: true}]; - // Specify the filter configuration :ref:`response_body_mode - // ` + // Specifies the filter configuration + // :ref:`response_body_mode `. envoy.extensions.filters.http.ext_proc.v3.ProcessingMode.BodySendMode response_body_mode = 2 - [(validate.rules).enum = {defined_only: true}]; + [(validate.rules).enum = {defined_only: true}]; - // Specify the filter configuration :ref:`send_body_without_waiting_for_header_response - // ` - // If the client is waiting for a header response from the server, setting ``true`` means the client will send body to the server - // as they arrive. Setting ``false`` means the client will buffer the arrived data and not send it to the server immediately. + // Specifies the filter configuration + // :ref:`send_body_without_waiting_for_header_response `. + // If the client is waiting for a header response from the server, setting to ``true`` means the + // client will send the body to the server as it arrives. Setting to ``false`` means the client + // will buffer the arrived data and not send it to the server immediately. bool send_body_without_waiting_for_header_response = 3; } // This represents the different types of messages that the data plane can send // to an external processing server. -// [#next-free-field: 12] +// [#next-free-field: 14] message ProcessingRequest { + // Initial flow control window sizes for ``FULL_DUPLEX_STREAMED`` and + // ``GRPC`` body send modes. + // + // A sender starts with this amount of flow control window. Whenever + // it sends body data, it must decrement its flow control window by + // the number of bytes that it has sent. When its flow control + // window is less than or equal to the amount of body data it wishes + // to send, it may not send until it receives a window update causing + // its flow control window to be large enough. + // + // However, note that in ``GRPC`` body send mode, whenever the flow + // control window is greater than zero, a sender may send a single + // message, even if the size of that message exceeds the available flow + // control window. At that point, the flow control window will be negative + // and the sender must not send the next message until it becomes positive. + // + // Note that the initial size for the to-sidestream windows are set by + // the sender, not the receiver. This is because each sidestream may be + // routed to a different ext_proc server instance, but there is no + // connection-level handshake to set a default for that server + // instance, so the only alternative here would be to have the + // ext_proc server instance set this on a per-stream basis, which + // would require an additional round-trip and therefore hurt latency. + // This unfortunately means that the ext_proc server instance has a + // bit less control: as soon as it receives these initial values, it can + // immediately send a window update that reduces the window, but it + // must be prepared to handle any data that the sender has already sent. + // The initial sizes for the to-sidestream windows are generally + // expected to be in the range of 32K to 64K. + // + // [#not-implemented-hide:] + message FlowControlInit { + // Downstream-to-sidestream initial window size. + int64 initial_window_downstream_to_sidestream = 1; + + // Sidestream-to-upstream initial window size. + int64 initial_window_sidestream_to_upstream = 2; + + // Upstream-to-sidestream initial window size. + int64 initial_window_upstream_to_sidestream = 3; + + // Sidestream-to-downstream initial window size. + int64 initial_window_sidestream_to_downstream = 4; + } + + // Flow control window update. Values may be positive or negative. The + // sender must immediately add these values to its flow control window, + // which governs how much data can be sent. + // + // [#not-implemented-hide:] + message ClientWindowUpdate { + // Window update for sidestream-to-upstream. + int64 window_increment_sidestream_to_upstream = 1; + + // Window update for sidestream-to-downstream. + int64 window_increment_sidestream_to_downstream = 2; + } + reserved 1; reserved "async_mode"; @@ -93,35 +148,33 @@ message ProcessingRequest { // ones are set for a particular HTTP request/response depend on the // processing mode. oneof request { - option (validate.required) = true; - // Information about the HTTP request headers, as well as peer info and additional // properties. Unless ``observability_mode`` is ``true``, the server must send back a - // HeaderResponse message, an ImmediateResponse message, or close the stream. + // ``HeaderResponse`` message, an ``ImmediateResponse`` message, or close the stream. HttpHeaders request_headers = 2; // Information about the HTTP response headers, as well as peer info and additional // properties. Unless ``observability_mode`` is ``true``, the server must send back a - // HeaderResponse message or close the stream. + // ``HeaderResponse`` message or close the stream. HttpHeaders response_headers = 3; - // A chunk of the HTTP request body. Unless ``observability_mode`` is true, the server must send back - // a BodyResponse message, an ImmediateResponse message, or close the stream. + // A chunk of the HTTP request body. Unless ``observability_mode`` is ``true``, the server must + // send back a ``BodyResponse`` message, an ``ImmediateResponse`` message, or close the stream. HttpBody request_body = 4; - // A chunk of the HTTP response body. Unless ``observability_mode`` is ``true``, the server must send back - // a BodyResponse message or close the stream. + // A chunk of the HTTP response body. Unless ``observability_mode`` is ``true``, the server must + // send back a ``BodyResponse`` message or close the stream. HttpBody response_body = 5; // The HTTP trailers for the request path. Unless ``observability_mode`` is ``true``, the server - // must send back a TrailerResponse message or close the stream. + // must send back a ``TrailerResponse`` message or close the stream. // // This message is only sent if the trailers processing mode is set to ``SEND`` and // the original downstream request has trailers. HttpTrailers request_trailers = 6; // The HTTP trailers for the response path. Unless ``observability_mode`` is ``true``, the server - // must send back a TrailerResponse message or close the stream. + // must send back a ``TrailerResponse`` message or close the stream. // // This message is only sent if the trailers processing mode is set to ``SEND`` and // the original upstream response has trailers. @@ -137,39 +190,75 @@ message ProcessingRequest { // :ref:`attributes ` supported in the data plane. map attributes = 9; - // Specify whether the filter that sent this request is running in :ref:`observability_mode - // ` - // and defaults to false. + // Specifies whether the filter that sent this request is running in + // :ref:`observability_mode `. // - // * A value of ``false`` indicates that the server must respond - // to this message by either sending back a matching ProcessingResponse message, - // or by closing the stream. + // * A value of ``false`` indicates that the server must respond to this message by either + // sending back a matching ``ProcessingResponse`` message, or by closing the stream. // * A value of ``true`` indicates that the server should not respond to this message, as any - // responses will be ignored. However, it may still close the stream to indicate that no more messages - // are needed. + // responses will be ignored. However, it may still close the stream to indicate that no more + // messages are needed. // + // Defaults to ``false``. bool observability_mode = 10; // Specify the filter protocol configurations to be sent to the server. // ``protocol_config`` is only encoded in the first ``ProcessingRequest`` message from the client to the server. ProtocolConfiguration protocol_config = 11; + + // Flow control initialization for ``FULL_DUPLEX_STREAMED`` and + // ``GRPC`` body send modes. + // + // Must be set in the initial message on the stream. Not used in + // subsequent messages. + // + // [#not-implemented-hide:] + FlowControlInit flow_control_init = 12; + + // Flow control updates for ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body + // send modes. + // + // This message may be included in a request message that also + // populates one of the fields in the ``request`` oneof above, or it + // may be sent in a request message that does not set the + // ``request`` oneof. + // + // In ``FULL_DUPLEX_STREAMED`` body send mode, for backward + // compatibility with data planes that do not yet support flow control, + // the data plane must not send a message containing only this field + // (i.e., not setting the ``request`` oneof) unless the ext_proc server + // has sent a window update, thus indicating that it supports flow control. + // + // [#not-implemented-hide:] + ClientWindowUpdate client_window_update = 13; } // This represents the different types of messages the server may send back to the data plane -// when the ``observability_mode`` field in the received ProcessingRequest is set to false. +// when the ``observability_mode`` field in the received ``ProcessingRequest`` is set to ``false``. // // * If the corresponding ``BodySendMode`` in the // :ref:`processing_mode ` -// is not set to ``FULL_DUPLEX_STREAMED``, then for every received ProcessingRequest, -// the server must send back exactly one ProcessingResponse message. +// is not set to ``FULL_DUPLEX_STREAMED``, then for every received ``ProcessingRequest``, +// the server must send back exactly one ``ProcessingResponse`` message. // * If it is set to ``FULL_DUPLEX_STREAMED``, the server must follow the API defined -// for this mode to send the ProcessingResponse messages. -// [#next-free-field: 13] +// for this mode to send the ``ProcessingResponse`` messages. +// [#next-free-field: 17] message ProcessingResponse { + // Flow control window update. Values may be positive or negative. The + // sender must immediately add these values to its flow control window, + // which governs how much data can be sent. + // + // [#not-implemented-hide:] + message ServerWindowUpdate { + // Window update for downstream-to-sidestream. + int64 window_increment_downstream_to_sidestream = 1; + + // Window update for upstream-to-sidestream. + int64 window_increment_upstream_to_sidestream = 2; + } + // The response type that is sent by the server. oneof response { - option (validate.required) = true; - // The server must send back this message in response to a message with the // ``request_headers`` field set. HeadersResponse request_headers = 1; @@ -204,17 +293,19 @@ message ProcessingResponse { ImmediateResponse immediate_response = 7; // The server sends back this message to initiate or continue local response streaming. - // The server must initiate local response streaming with the ``headers_response`` in response to a ProcessingRequest - // with the ``request_headers`` only. - // The server may follow up with multiple messages containing ``body_response``. The server must indicate - // end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` + // The server must initiate local response streaming with the ``headers_response`` in response + // to a ``ProcessingRequest`` with the ``request_headers`` only. + // The server may follow up with multiple messages containing ``body_response``. The server must + // indicate end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` // or ``body_response`` message or by sending a ``trailers_response`` message. - // The client may send a ``request_body`` or ``request_trailers`` to the server depending on configuration. + // The client may send a ``request_body`` or ``request_trailers`` to the server depending on + // configuration. // The streaming local response can only be sent when the ``request_header_mode`` in the filter // :ref:`processing_mode ` - // is set to ``SEND``. The ext_proc server should not send StreamedImmediateResponse if it did not observe request headers, - // as it will result in the race with the upstream server response and reset of the client request. - // Presently only the FULL_DUPLEX_STREAMED or NONE body modes are supported. + // is set to ``SEND``. The ext_proc server should not send ``StreamedImmediateResponse`` if it + // did not observe request headers, as it will result in a race with the upstream server + // response and reset of the client request. + // Presently only the ``FULL_DUPLEX_STREAMED`` or ``NONE`` body modes are supported. StreamedImmediateResponse streamed_immediate_response = 11; } @@ -223,99 +314,229 @@ message ProcessingResponse { // field name(s) of the struct. google.protobuf.Struct dynamic_metadata = 8; - // Override how parts of the HTTP request and response are processed - // for the duration of this particular request/response only. Servers - // may use this to intelligently control how requests are processed - // based on the headers and other metadata that they see. - // This field is only applicable when servers responding to the header requests. - // If it is set in the response to the body or trailer requests, it will be ignored by the data plane. + // Optional typed metadata that will be emitted as dynamic metadata to be consumed by + // following filters. This metadata will be placed in the namespace(s) specified by the + // keys of the map. + // + // Typed dynamic metadata should be preferred over untyped dynamic metadata (``dynamic_metadata``) + // because it is more efficient and more type-safe. + map typed_dynamic_metadata = 13; + + // Override how parts of the HTTP request and response are processed for the duration of this + // particular request/response only. Servers may use this to intelligently control how requests + // are processed based on the headers and other metadata that they see. + // + // This field is applicable when servers are responding to the header requests. If it is set + // in the response to the body or trailer requests, it will be ignored by the data plane. // It is also ignored by the data plane when the ext_proc filter config - // :ref:`allow_mode_override - // ` - // is set to false, or - // :ref:`send_body_without_waiting_for_header_response - // ` - // is set to true. + // :ref:`allow_mode_override ` + // is set to ``false``, or + // :ref:`send_body_without_waiting_for_header_response ` + // is set to ``true``. + // + // The external processing server can override the processing mode by returning a standalone ``mode_override`` + // in the ``ProcessingResponse`` after receiving a ``ProcessingRequest`` for request headers. This standalone + // override must be sent before the request headers response is sent. + // Subsequent messages will adhere to this new mode. + // + // Constraints: + // + // 1. Request Path: A standalone ``mode_override`` only supports transitioning the body processing mode + // from other mode to ``FULL_DUPLEX_STREAMED``, and the trailer mode must be set to ``SEND``. + // 2. Response Path: There are no restrictions on processing mode changes in this case. + // envoy.extensions.filters.http.ext_proc.v3.ProcessingMode mode_override = 9; // [#not-implemented-hide:] - // Used only in ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. - // Instructs the data plane to stop sending body data and to send a - // half-close on the ext_proc stream. The ext_proc server should then echo - // back all subsequent body contents as-is until it sees the client's - // half-close, at which point the ext_proc server can terminate the stream - // with an OK status. This provides a safe way for the ext_proc server - // to indicate that it does not need to see the rest of the stream; - // without this, the ext_proc server could not terminate the stream - // early, because it would wind up dropping any body contents that the - // client had already sent before it saw the ext_proc stream termination. - bool request_drain = 12; - - // When ext_proc server receives a request message, in case it needs more - // time to process the message, it sends back a ProcessingResponse message - // with a new timeout value. When the data plane receives this response - // message, it ignores other fields in the response, just stop the original - // timer, which has the timeout value specified in - // :ref:`message_timeout - // ` - // and start a new timer with this ``override_message_timeout`` value and keep the - // data plane ext_proc filter state machine intact. - // Has to be >= 1ms and <= - // :ref:`max_message_timeout ` - // Such message can be sent at most once in a particular data plane ext_proc filter processing state. - // To enable this API, one has to set ``max_message_timeout`` to a number >= 1ms. + // Deprecated and not implemented. This field has been replaced with + // the request_drain_requests and request_drain_responses fields. + bool request_drain = 12 + [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; + + // [#not-implemented-hide:] + // Initiates a drain of request body data. Used only in + // ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. + // + // The expected sequence for a drain is as follows: + // + // 1. The ext_proc server sends a message to the data plane with this + // field set to true. + // 2. The data plane pauses reading from the downstream client, applying + // any necessary flow control push-back. + // 3. The data plane sends a + // :ref:`request_body ` + // with the + // :ref:`drain_complete ` + // field set to true. + // 4. The ext_proc server continues processing all subsequent request body + // chunks until it sees the body chunk with the + // :ref:`drain_complete ` + // field set to true. When sending back its response to that last body + // chunk, the ext_proc server will set the + // :ref:`drain_complete ` + // field to true to let the data plane know that it has finished draining. + // 5. When the data plane sees the response with the + // :ref:`drain_complete ` + // field set to true, it will resume reading from the downstream + // client, passing all data directly to the upstream server, + // without going through the ext_proc sidestream. + // + // This procedure provides a safe way for the ext_proc server to indicate + // that it does not need to see the rest of the request body. + // + // Note that if the data plane is sending request body data and the + // ext_proc server wants to terminate the stream with an OK status, it + // must perform this drain before doing so. + // + // Note that the data plane may have either sent a body chunk with + // :ref:`end_of_stream ` + // set to true or may have sent trailers before it received the drain + // request from the ext_proc server. In these cases, the data plane will + // ignore the drain request, and the ext_proc will consider the drain + // complete when it sees the end-of-stream or trailers. + bool request_drain_requests = 15; + + // [#not-implemented-hide:] + // Initiates a drain of response body data. Used only in + // ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body send modes. + // + // The expected sequence for a drain is as follows: + // + // 1. The ext_proc server sends a message to the data plane with this + // field set to true. + // 2. The data plane pauses reading from the upstream server, applying + // any necessary flow control push-back. + // 3. The data plane sends a + // :ref:`response_body ` + // with the + // :ref:`drain_complete ` + // field set to true. + // 4. The ext_proc server continues processing all subsequent response body + // chunks until it sees the body chunk with the + // :ref:`drain_complete ` + // field set to true. When sending back its response to that last body + // chunk, the ext_proc server will set the + // :ref:`drain_complete ` + // field to true to let the data plane know that it has finished draining. + // 5. When the data plane sees the response with the + // :ref:`drain_complete ` + // field set to true, it will resume reading from the upstream + // server, passing all data directly to the downstream client, + // without going through the ext_proc sidestream. + // + // This procedure provides a safe way for the ext_proc server to indicate + // that it does not need to see the rest of the response body. + // + // Note that if the data plane is sending response body data and the + // ext_proc server wants to terminate the stream with an OK status, it + // must perform this drain before doing so. + // + // Note that the data plane may have either sent a body chunk with + // :ref:`end_of_stream ` + // set to true or may have sent trailers before it received the drain + // request from the ext_proc server. In these cases, the data plane will + // ignore the drain request, and the ext_proc will consider the drain + // complete when it sees the end-of-stream or trailers. + bool request_drain_responses = 16; + + // When the ext_proc server receives a request message and needs more time to process it, it + // sends back a ``ProcessingResponse`` message with a new timeout value. When the data plane + // receives this response message, it ignores other fields in the response, stops the original + // timer (which has the timeout value specified in + // :ref:`message_timeout `), + // and starts a new timer with this ``override_message_timeout`` value while keeping the data + // plane ext_proc filter state machine intact. + // + // The value must be >= 1ms and <= + // :ref:`max_message_timeout `. + // Such a message can be sent at most once in a particular data plane ext_proc filter processing + // state. To enable this API, ``max_message_timeout`` must be set to a value >= 1ms. google.protobuf.Duration override_message_timeout = 10; + + // Flow control updates for ``FULL_DUPLEX_STREAMED`` and ``GRPC`` body + // send modes. + // + // This message may be included in a response message that also + // populates one of the fields in the ``response`` oneof above, or it + // may be sent in a response message that does not set the + // ``response`` oneof. + // + // In ``FULL_DUPLEX_STREAMED`` body send mode, for backward + // compatibility with data planes that do not yet support flow control, + // the ext_proc server must not set this field unless the data plane + // sent initial window sizes in its initial message on the stream. + // Conversely, if the data plane did send initial window sizes in its + // initial message on the stream, the ext_proc server must send a + // window update immediately to let the data plane know that it also + // supports flow control. If the ext_proc server is sending a message + // immediately anyway (e.g., for a header or body chunk), it can include + // this field in that same message; otherwise, the ext_proc server must + // send a message containing only this field. + // + // [#not-implemented-hide:] + ServerWindowUpdate server_window_update = 14; } // The following are messages that are sent to the server. -// This message is sent to the external server when the HTTP request and responses +// This message is sent to the external server when the HTTP request and response headers // are first received. message HttpHeaders { - // The HTTP request headers. All header keys will be - // lower-cased, because HTTP header keys are case-insensitive. - // The header value is encoded in the + // The HTTP request headers. All header keys will be lower-cased, because HTTP header keys are + // case-insensitive. The header value is encoded in the // :ref:`raw_value ` field. config.core.v3.HeaderMap headers = 1; // [#not-implemented-hide:] - // This field is deprecated and not implemented. Attributes will be sent in - // the top-level :ref:`attributes ` field. map attributes = 2 - [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; + [deprecated = true, (envoy.annotations.deprecated_at_minor_version) = "3.0"]; - // If ``true``, then there is no message body associated with this - // request or response. + // If ``true``, then there is no message body associated with this request or response. bool end_of_stream = 3; } -// This message is sent to the external server when the HTTP request and -// response bodies are received. +// This message is sent to the external server when the HTTP request and response bodies are +// received. +// [#next-free-field: 6] message HttpBody { - // The contents of the body in the HTTP request/response. Note that in - // streaming mode multiple ``HttpBody`` messages may be sent. + // The contents of the body in the HTTP request/response. Note that in streaming mode multiple + // ``HttpBody`` messages may be sent. // - // In ``GRPC`` body send mode, a separate ``HttpBody`` message will be - // sent for each message in the gRPC stream. + // In ``GRPC`` body send mode, a separate ``HttpBody`` message will be sent for each message in + // the gRPC stream. bytes body = 1; - // If ``true``, this will be the last ``HttpBody`` message that will be sent and no - // trailers will be sent for the current request/response. + // If ``true``, this will be the last ``HttpBody`` message that will be sent and no trailers + // will be sent for the current request/response. bool end_of_stream = 2; - // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is - // true and ``body`` is empty. Those values would normally indicate an - // empty message on the stream with the end-of-stream bit set. - // However, if the half-close happens after the last message on the - // stream was already sent, then this field will be true to indicate an - // end-of-stream with *no* message (as opposed to an empty message). + // This field is used only in ``GRPC`` body send mode. It is not used in any other body send + // mode. + // + // This field is used only when ``end_of_stream`` is true and ``body`` is empty. + // Normally, in ``GRPC`` body send mode, an empty ``body`` field indicates an empty message on + // the gRPC stream. However, it is possible that the gRPC client sends a half-close without + // actually sending a message on the stream, so we need a way to differentiate between + // an empty message being sent and no message being sent. If this field is true, then it + // indicates that no message has been sent; if it is false, then it indicates that an empty + // message has been sent. + // [#not-implemented-hide:] bool end_of_stream_without_message = 3; - // This field is used in ``GRPC`` body send mode to indicate whether - // the message is compressed. This will never be set to true by gRPC - // but may be set to true by a proxy like Envoy. + // This field is used in ``GRPC`` body send mode to indicate whether the message is compressed. + // This will never be set to ``true`` by gRPC but may be set to ``true`` by a proxy like Envoy. bool grpc_message_compressed = 4; + + // [#not-implemented-hide:] + // In ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body send modes, if the + // data plane has seen the + // :ref:`request_drain_requests ` + // or :ref:`request_drain_responses ` + // field, it will populate this field to indicate that it has finished + // sending data to the ext_proc server. + bool drain_complete = 5; } // This message is sent to the external server when the HTTP request and @@ -352,13 +573,14 @@ message TrailersResponse { HeaderMutation header_mutation = 1; } -// This message is sent by the external server to the data plane after ``HttpHeaders`` -// to initiate local response streaming. The server may follow up with multiple messages containing ``body_response``. -// The server must indicate end of stream by setting ``end_of_stream`` to ``true`` in the ``headers_response`` -// or ``body_response`` message or by sending a ``trailers_response`` message. +// This message is sent by the external server to the data plane after ``HttpHeaders`` to initiate +// local response streaming. The server may follow up with multiple messages containing +// ``body_response``. The server must indicate end of stream by setting ``end_of_stream`` to +// ``true`` in the ``headers_response`` or ``body_response`` message or by sending a +// ``trailers_response`` message. message StreamedImmediateResponse { oneof response { - // Response headers to be sent downstream. The ":status" header must be set. + // Response headers to be sent downstream. The ``:status`` header must be set. HttpHeaders headers_response = 1; // Response body to be sent downstream. @@ -384,7 +606,7 @@ message CommonResponse { // further messages for this request or response even if the processing // mode is configured to do so. // - // When used in response to a request_headers or response_headers message, + // When used in response to a ``request_headers`` or ``response_headers`` message, // this status makes it possible to either completely replace the body // while discarding the original body, or to add a body to a message that // formerly did not have one. @@ -401,23 +623,22 @@ message CommonResponse { ResponseStatus status = 1 [(validate.rules).enum = {defined_only: true}]; // Instructions on how to manipulate the headers. When responding to an - // HttpBody request, header mutations will only take effect if - // the current processing mode for the body is BUFFERED. + // ``HttpBody`` request, header mutations will only take effect if the current processing mode + // for the body is ``BUFFERED``. HeaderMutation header_mutation = 2; - // Replace the body of the last message sent to the remote server on this - // stream. If responding to an HttpBody request, simply replace or clear - // the body chunk that was sent with that request. Body mutations may take - // effect in response either to ``header`` or ``body`` messages. When it is - // in response to ``header`` messages, it only take effect if the + // Replace the body of the last message sent to the remote server on this stream. If responding + // to an ``HttpBody`` request, simply replace or clear the body chunk that was sent with that + // request. Body mutations may take effect in response either to ``header`` or ``body`` messages. + // When it is in response to ``header`` messages, it only takes effect if the // :ref:`status ` - // is set to CONTINUE_AND_REPLACE. + // is set to ``CONTINUE_AND_REPLACE``. BodyMutation body_mutation = 3; // [#not-implemented-hide:] - // Add new trailers to the message. This may be used when responding to either a - // HttpHeaders or HttpBody message, but only if this message is returned - // along with the CONTINUE_AND_REPLACE status. + // Add new trailers to the message. This may be used when responding to either an + // ``HttpHeaders`` or ``HttpBody`` message, but only if this message is returned + // along with the ``CONTINUE_AND_REPLACE`` status. // The header value is encoded in the // :ref:`raw_value ` field. config.core.v3.HeaderMap trailers = 4; @@ -429,34 +650,32 @@ message CommonResponse { bool clear_route_cache = 5; } -// This message causes the filter to attempt to create a locally -// generated response, send it downstream, stop processing -// additional filters, and ignore any additional messages received -// from the remote server for this request or response. If a response -// has already started, then this will either ship the reply directly -// to the downstream codec, or reset the stream. +// This message causes the filter to attempt to create a locally generated response, send it +// downstream, stop processing additional filters, and ignore any additional messages received +// from the remote server for this request or response. If a response has already started, then +// this will either ship the reply directly to the downstream codec, or reset the stream. // [#next-free-field: 6] message ImmediateResponse { // The response code to return. type.v3.HttpStatus status = 1 [(validate.rules).message = {required: true}]; - // Apply changes to the default headers, which will include content-type. + // Apply changes to the default headers, which will include ``content-type``. HeaderMutation headers = 2; // The message body to return with the response which is sent using the - // text/plain content type, or encoded in the grpc-message header. + // ``text/plain`` content type, or encoded in the ``grpc-message`` header. bytes body = 3; // If set, then include a gRPC status trailer. GrpcStatus grpc_status = 4; // A string detailing why this local reply was sent, which may be included - // in log and debug output (e.g. this populates the %RESPONSE_CODE_DETAILS% + // in log and debug output (e.g., this populates the ``%RESPONSE_CODE_DETAILS%`` // command operator field for use in access logging). string details = 5; } -// This message specifies a gRPC status for an ImmediateResponse message. +// This message specifies a gRPC status for an ``ImmediateResponse`` message. message GrpcStatus { // The actual gRPC status. uint32 status = 1; @@ -478,33 +697,45 @@ message HeaderMutation { } // The body response message corresponding to ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body modes. +// [#next-free-field: 6] message StreamedBodyResponse { // In ``FULL_DUPLEX_STREAMED`` body send mode, contains the body response chunk that will be // passed to the upstream/downstream by the data plane. In ``GRPC`` body send mode, contains // a serialized gRPC message to be passed to the upstream/downstream by the data plane. bytes body = 1; - // The server sets this flag to true if it has received a body request with - // :ref:`end_of_stream ` set to true, - // and this is the last chunk of body responses. - // Note that in ``GRPC`` body send mode, this allows the ext_proc - // server to tell the data plane to send a half close after a client - // message, which will result in discarding any other messages sent by - // the client application. + // The server sets this flag to ``true`` if it has received a body request with + // :ref:`end_of_stream ` set to + // ``true``, and this is the last chunk of body responses. + // + // Note that in ``GRPC`` body send mode, this allows the ext_proc server to tell the data plane + // to send a half close after a client message, which will result in discarding any other + // messages sent by the client application. bool end_of_stream = 2; - // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is - // true and ``body`` is empty. Those values would normally indicate an - // empty message on the stream with the end-of-stream bit set. - // However, if the half-close happens after the last message on the - // stream was already sent, then this field will be true to indicate an - // end-of-stream with *no* message (as opposed to an empty message). + // This field is used in ``GRPC`` body send mode when ``end_of_stream`` is ``true`` and ``body`` + // is empty. Those values would normally indicate an empty message on the stream with the + // end-of-stream bit set. However, if the half-close happens after the last message on the stream + // was already sent, then this field will be ``true`` to indicate an end-of-stream with *no* + // message (as opposed to an empty message). bool end_of_stream_without_message = 3; - // This field is used in ``GRPC`` body send mode to indicate whether - // the message is compressed. This will never be set to true by gRPC - // but may be set to true by a proxy like Envoy. + // This field is used in ``GRPC`` body send mode to indicate whether the message is compressed. + // This will never be set to ``true`` by gRPC but may be set to ``true`` by a proxy like Envoy. bool grpc_message_compressed = 4; + + // [#not-implemented-hide:] + // In ``FULL_DUPLEX_STREAMED`` or ``GRPC`` body send modes, if the + // ext_proc server has seen the + // :ref:`drain_complete ` + // field, it will populate this field to indicate that it has finished + // sending data back to the data plane. + // + // If the data plane receives a message with this field set to true + // before it has sent a drain-complete message to the ext_proc server, + // the data plane will treat that as if the ext_proc stream has failed + // with a non-OK status. + bool drain_complete = 5; } // This message specifies the body mutation the server sends to the data plane. @@ -517,17 +748,16 @@ message BodyMutation { // is not set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. bytes body = 1; - // Clear the corresponding body chunk. - // Should only be used when the corresponding ``BodySendMode`` in the + // Clear the corresponding body chunk. Should only be used when the corresponding + // ``BodySendMode`` in the // :ref:`processing_mode ` // is not set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. - // Clear the corresponding body chunk. bool clear_body = 2; // Must be used when the corresponding ``BodySendMode`` in the // :ref:`processing_mode ` // is set to ``FULL_DUPLEX_STREAMED`` or ``GRPC``. StreamedBodyResponse streamed_response = 3 - [(xds.annotations.v3.field_status).work_in_progress = true]; + [(xds.annotations.v3.field_status).work_in_progress = true]; } } From 89dd6bf9177670c462d36b8c1bc7c1a58208eacc Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 07:42:32 +0000 Subject: [PATCH 2/9] Fix flow control bypass during client response draining in ext_proc. When activateCall is called during draining (initiated by response drain), it flushes requests buffered in DelayedClientCall directly to the rawCall, bypassing flow control checks. To fix this: 1. Buffer request(n) calls in pendingRequests when the call is IDLE, instead of letting them go to DelayedClientCall. 2. Guard drainPendingRequests to check isResponseSidecarReady and abort if it is false (keeping requests buffered). 3. Trigger drainPendingRequests when response drain completes. Also updated tests to verify early request buffering during response drain. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../ExternalProcessorClientInterceptor.java | 8 +++++ ...xternalProcessorClientInterceptorTest.java | 32 +++++++++++++++---- 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index d83b3a39326..5988dd347d9 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -726,6 +726,9 @@ private void onExtProcStreamReady() { } private void drainPendingRequests() { + if (!isResponseSidecarReady()) { + return; + } int toRequest = pendingRequests.getAndSet(0); if (toRequest > 0) { super.request(toRequest); @@ -828,6 +831,10 @@ public void request(int numMessages) { super.request(numMessages); return; } + if (dataPlaneCallState.get() == DataPlaneCallState.IDLE) { + pendingRequests.addAndGet(numMessages); + return; + } if (!isResponseSidecarReady()) { pendingRequests.addAndGet(numMessages); return; @@ -1391,6 +1398,7 @@ void unblockAfterResponseDrain() { } proceedWithHeaders(); proceedWithSavedMessages(); + dataPlaneClientCall.drainPendingRequests(); proceedWithClose(); } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 8447ea150d7..604a93c66b3 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -9020,6 +9020,8 @@ public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffere ExternalProcessorFilterConfig filterConfig = configOrError.config; final CountDownLatch drainSentLatch = new CountDownLatch(1); + final CountDownLatch headersReceivedLatch = new CountDownLatch(1); + final CountDownLatch sendDrainLatch = new CountDownLatch(1); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -9032,10 +9034,20 @@ public StreamObserver process( @Override public void onNext(ProcessingRequest request) { if (request.hasRequestHeaders()) { - responseObserver.onNext(ProcessingResponse.newBuilder() - .setRequestDrainResponses(true) - .build()); - drainSentLatch.countDown(); + headersReceivedLatch.countDown(); + new Thread(() -> { + try { + sendDrainLatch.await(); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainResponses(true) + .build()); + } + drainSentLatch.countDown(); + } catch (Exception e) { + System.out.println("JetskiDebug: error in mock server: " + e); + } + }).start(); } } @@ -9094,15 +9106,21 @@ public void request(int numMessages) { ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + // Wait for headers to reach mock server + assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // App requests messages early (before drain is received) + proxyCall.request(3); + + // Now trigger drain + sendDrainLatch.countDown(); // Wait for drain to be processed assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue(); // proxyCall.isReady() should remain true during response drain (request path is NOT draining) assertThat(proxyCall.isReady()).isTrue(); - // App requests more messages - proxyCall.request(3); - // Verify requests are buffered and not sent to data plane assertThat(dataPlaneRequestCount.get()).isEqualTo(0); From f270d8fd1f2556209d62d3e7886903ed02c0d90b Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 08:27:17 +0000 Subject: [PATCH 3/9] Extend response drain test to verify handshake completion. Extend givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffered to complete the response draining handshake and verify that buffered requests are then successfully sent to the data plane. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- ...xternalProcessorClientInterceptorTest.java | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 604a93c66b3..e5f9bf4a494 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -9022,6 +9022,9 @@ public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffere final CountDownLatch drainSentLatch = new CountDownLatch(1); final CountDownLatch headersReceivedLatch = new CountDownLatch(1); final CountDownLatch sendDrainLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1); + final AtomicReference> responseObserverRef = + new AtomicReference<>(); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -9029,6 +9032,7 @@ public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffere @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + responseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -9048,6 +9052,10 @@ public void onNext(ProcessingRequest request) { System.out.println("JetskiDebug: error in mock server: " + e); } }).start(); + } else if (request.hasResponseBody()) { + if (request.getResponseBody().getDrainComplete()) { + filterSentDrainCompleteLatch.countDown(); + } } } @@ -9123,6 +9131,27 @@ public void request(int numMessages) { // Verify requests are buffered and not sent to data plane assertThat(dataPlaneRequestCount.get()).isEqualTo(0); + + // Wait for filter to send drain_complete to mock server + assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Echo drain_complete back to filter to complete handshake + synchronized (responseObserverRef.get()) { + responseObserverRef.get().onNext(ProcessingResponse.newBuilder() + .setResponseBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + + // Verify requests are now drained to data plane + assertThat(dataPlaneRequestCount.get()).isEqualTo(3); proxyCall.cancel("Cleanup", null); channelManager.close(); From 71a3964170792b5ab92c48aa84c9c4a4d01dc3f3 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 08:31:14 +0000 Subject: [PATCH 4/9] Add request drain flow control verification test. Add givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained to verify that early request(n) calls are buffered when IDLE, and successfully drained to the data plane server upon activation if only request draining is active (which should not block the response/read path). TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- ...xternalProcessorClientInterceptorTest.java | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index e5f9bf4a494..89cef5abeb5 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -9157,6 +9157,146 @@ public void request(int numMessages) { channelManager.close(); } + @Test + @SuppressWarnings("unchecked") + public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained() + throws Exception { + ExternalProcessor proto = ExternalProcessor.newBuilder() + .setGrpcService(GrpcService.newBuilder() + .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder() + .setTargetUri("in-process:///" + extProcServerName) + .addChannelCredentialsPlugin(Any.newBuilder() + .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service." + + "channel_credentials.insecure.v3.InsecureCredentials") + .build()) + .build()) + .build()) + .setProcessingMode(ProcessingMode.newBuilder() + .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC) + .setRequestTrailerMode(ProcessingMode.HeaderSendMode.SEND) + .build()) + .build(); + ConfigOrError configOrError = + provider.parseFilterConfig(Any.pack(proto), filterContext); + assertThat(configOrError.errorDetail).isNull(); + ExternalProcessorFilterConfig filterConfig = configOrError.config; + + final CountDownLatch drainSentLatch = new CountDownLatch(1); + final CountDownLatch headersReceivedLatch = new CountDownLatch(1); + final CountDownLatch sendDrainLatch = new CountDownLatch(1); + final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1); + // External Processor Server + ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; + extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { + @Override + @SuppressWarnings("unchecked") + public StreamObserver process( + final StreamObserver responseObserver) { + ((ServerCallStreamObserver) responseObserver).request(100); + return new StreamObserver() { + @Override + public void onNext(ProcessingRequest request) { + if (request.hasRequestHeaders()) { + headersReceivedLatch.countDown(); + new Thread(() -> { + try { + sendDrainLatch.await(); + synchronized (responseObserver) { + responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestDrainRequests(true) + .build()); + } + drainSentLatch.countDown(); + } catch (Exception e) {} + }).start(); + } else if (request.hasRequestBody()) { + if (request.getRequestBody().getDrainComplete()) { + filterSentDrainCompleteLatch.countDown(); + } + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + }; + grpcCleanup.register(InProcessServerBuilder.forName(extProcServerName) + .addService(extProcImpl) + .directExecutor() + .build().start()); + + CachedChannelManager channelManager = new CachedChannelManager(config -> { + return grpcCleanup.register( + InProcessChannelBuilder.forName(extProcServerName).directExecutor().build()); + }); + + dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService") + .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall( + (request, responseObserver) -> { + responseObserver.onNext("Hello " + request); + responseObserver.onCompleted(); + })) + .build()); + + final AtomicInteger dataPlaneRequestCount = new AtomicInteger(); + ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor( + filterConfig, channelManager, scheduler, FAKE_CONTEXT); + + ManagedChannel dataPlaneChannel = grpcCleanup.register( + InProcessChannelBuilder.forName(dataPlaneServerName) + .directExecutor() + .intercept(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void request(int numMessages) { + dataPlaneRequestCount.addAndGet(numMessages); + super.request(numMessages); + } + }; + } + }) + .build()); + + CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); + ClientCall proxyCall = + interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); + proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + + // Wait for headers to reach mock server + assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // App requests messages early (before drain is received) + proxyCall.request(3); + + // Verify requests are buffered and not sent to data plane yet (call is IDLE) + assertThat(dataPlaneRequestCount.get()).isEqualTo(0); + + // Now trigger drain (which also activates the call) + sendDrainLatch.countDown(); + + // Wait for drain to be processed + assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + // Verify requests are now drained to data plane (request drain does not block response path) + assertThat(dataPlaneRequestCount.get()).isEqualTo(3); + + // Wait for filter to send drain_complete to mock server + assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + proxyCall.cancel("Cleanup", null); + channelManager.close(); + } + @Test @SuppressWarnings("unchecked") public void givenBufferedRequests_whenExtProcStreamBecomesReady_thenDataPlaneDrained() From c9e02faed9979479289699e2b5224643ff786646 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 10:52:19 +0000 Subject: [PATCH 5/9] Fix onReady notification flow control and verification. 1. Fix onReadyNotify in DataPlaneListener to only forward onReady if dataPlaneClientCall.isReady() is true. 2. Call onReadyNotify in handleRequestDrainComplete to notify the app when request draining completes and the path becomes ready again. 3. Update givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained to verify that onReady is NOT called during request drain but IS called after the drain completes. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../ExternalProcessorClientInterceptor.java | 7 ++++- ...xternalProcessorClientInterceptorTest.java | 28 ++++++++++++++++++- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index 5988dd347d9..d8732314570 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -1064,6 +1064,9 @@ private void handleRequestDrainComplete() { if (requestDraining.compareAndSet(true, false)) { requestDrainComplete.set(true); drainPendingDrainingMessages(); + if (wrappedListener != null) { + wrappedListener.onReadyNotify(); + } } } @@ -1327,7 +1330,9 @@ public void onClose(Status status, Metadata trailers) { } void onReadyNotify() { - dataPlaneClientCall.getCallContext().run(() -> delegate().onReady()); + if (dataPlaneClientCall.isReady()) { + dataPlaneClientCall.getCallContext().run(() -> delegate().onReady()); + } } void proceedWithHeaders() { diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 89cef5abeb5..57c2fdb3fe7 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -9185,6 +9185,8 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained( final CountDownLatch headersReceivedLatch = new CountDownLatch(1); final CountDownLatch sendDrainLatch = new CountDownLatch(1); final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1); + final AtomicReference> responseObserverRef = + new AtomicReference<>(); // External Processor Server ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl; extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() { @@ -9192,6 +9194,7 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained( @SuppressWarnings("unchecked") public StreamObserver process( final StreamObserver responseObserver) { + responseObserverRef.set(responseObserver); ((ServerCallStreamObserver) responseObserver).request(100); return new StreamObserver() { @Override @@ -9203,6 +9206,7 @@ public void onNext(ProcessingRequest request) { sendDrainLatch.await(); synchronized (responseObserver) { responseObserver.onNext(ProcessingResponse.newBuilder() + .setRequestHeaders(HeadersResponse.newBuilder().build()) .setRequestDrainRequests(true) .build()); } @@ -9270,7 +9274,8 @@ public void request(int numMessages) { CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()); ClientCall proxyCall = interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel); - proxyCall.start(new ClientCall.Listener() {}, new Metadata()); + ClientCall.Listener mockListener = Mockito.mock(ClientCall.Listener.class); + proxyCall.start(mockListener, new Metadata()); // Wait for headers to reach mock server assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue(); @@ -9289,10 +9294,31 @@ public void request(int numMessages) { // Verify requests are now drained to data plane (request drain does not block response path) assertThat(dataPlaneRequestCount.get()).isEqualTo(3); + + // Verify onReady was NOT called because request path is draining + Mockito.verify(mockListener, Mockito.never()).onReady(); // Wait for filter to send drain_complete to mock server assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // Echo drain_complete back to filter to complete handshake + synchronized (responseObserverRef.get()) { + responseObserverRef.get().onNext(ProcessingResponse.newBuilder() + .setRequestBody(BodyResponse.newBuilder() + .setResponse(CommonResponse.newBuilder() + .setBodyMutation(BodyMutation.newBuilder() + .setStreamedResponse(StreamedBodyResponse.newBuilder() + .setDrainComplete(true) + .build()) + .build()) + .build()) + .build()) + .build()); + } + + // Verify onReady IS now called after request drain completes + Mockito.verify(mockListener, Mockito.timeout(5000)).onReady(); + proxyCall.cancel("Cleanup", null); channelManager.close(); } From 33f3b12d49f968046383d93e77b0e35851f14061 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 11:29:49 +0000 Subject: [PATCH 6/9] Fix sendMessage race condition and test expectations. 1. Remove requestDrainComplete check from sendMessage to prevent out-of-order message delivery when app calls sendMessage concurrently with drain completion. Now we only bypass queue when passThroughMode is true. 2. Update givenExtProcStreamCompleted_whenIsReadyCalled_thenDelegatesToSuper test to expect the correct number of downstream isReady calls, which increased by 1 because we now correctly query isReady inside onReadyNotify on stream completion. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../java/io/grpc/xds/ExternalProcessorClientInterceptor.java | 5 +++-- .../io/grpc/xds/ExternalProcessorClientInterceptorTest.java | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index d8732314570..b579d865363 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -849,13 +849,13 @@ public void sendMessage(InputStream message) { return; } - if (passThroughMode.get() || requestDrainComplete.get()) { + if (passThroughMode.get()) { super.sendMessage(message); return; } synchronized (streamLock) { - if (passThroughMode.get() || requestDrainComplete.get()) { + if (passThroughMode.get()) { super.sendMessage(message); return; } @@ -1394,6 +1394,7 @@ void unblockAfterStreamComplete() { proceedWithHeaders(); proceedWithSavedMessages(); dataPlaneClientCall.drainPendingDrainingMessages(); + onReadyNotify(); proceedWithClose(); } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index 57c2fdb3fe7..e18a4a33e1e 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -6070,11 +6070,11 @@ public boolean isReady() { // 4. Assert that proxyCall.isReady() delegates directly to the downstream call downstreamReady.set(true); assertThat(proxyCall.isReady()).isTrue(); - assertThat(downstreamIsReadyCallCount.get()).isEqualTo(1); + assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2); downstreamReady.set(false); assertThat(proxyCall.isReady()).isFalse(); - assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2); + assertThat(downstreamIsReadyCallCount.get()).isEqualTo(3); proxyCall.cancel("cleanup", null); channelManager.close(); From 2d3c77b56770d2e5e446171dfcd92c07c644abb1 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 12:03:16 +0000 Subject: [PATCH 7/9] Fix halfClose race condition during request draining. Remove requestDrainComplete check from immediate halfClose path, and safely restore early halfClose optimization under lock: allow immediate halfClose during requestDraining/requestDrainComplete only if body mode is NONE or if the buffer queue is empty and we either completed the handshake or haven't sent any body messages yet. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../ExternalProcessorClientInterceptor.java | 60 ++++++++++--------- 1 file changed, 33 insertions(+), 27 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index b579d865363..bfa25a66d58 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -913,54 +913,60 @@ private void proceedWithHalfClose() { @Override public void halfClose() { clientHalfCloseStartNanos = System.nanoTime(); - if (passThroughMode.get() || requestDrainComplete.get()) { + if (passThroughMode.get()) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } return; } - pendingHalfClose.set(true); - - if (extProcStreamState.get().isCompleted()) { + synchronized (streamLock) { if (passThroughMode.get()) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } + return; } - return; - } - if (requestDraining.get()) { - boolean canProceed = false; - synchronized (streamLock) { - if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE - || (!bodyMessageSentToExtProc.get() && pendingDrainingMessages.isEmpty())) { + pendingHalfClose.set(true); + + if (extProcStreamState.get().isCompleted()) { + return; + } + + if (requestDraining.get() || requestDrainComplete.get()) { + boolean canProceed = false; + if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { canProceed = true; + } else if (pendingDrainingMessages.isEmpty()) { + if (requestDrainComplete.get() || !bodyMessageSentToExtProc.get()) { + canProceed = true; + } + } + + if (canProceed) { + if (requestSideClosed.compareAndSet(false, true)) { + proceedWithHalfClose(); + } } + return; } - if (canProceed) { + + if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { if (requestSideClosed.compareAndSet(false, true)) { proceedWithHalfClose(); } + return; } - return; - } - if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) { - if (requestSideClosed.compareAndSet(false, true)) { - proceedWithHalfClose(); - } - return; + // Mode is GRPC + sendToExtProc(ProcessingRequest.newBuilder() + .setRequestBody(HttpBody.newBuilder() + .setEndOfStreamWithoutMessage(true) + .build()) + .build()); + requestEosSent.set(true); } - - // Mode is GRPC - sendToExtProc(ProcessingRequest.newBuilder() - .setRequestBody(HttpBody.newBuilder() - .setEndOfStreamWithoutMessage(true) - .build()) - .build()); - requestEosSent.set(true); } private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) { From fdaaa9a66d86a80ba0fec2ef54996496a0cdaa2b Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 12:12:39 +0000 Subject: [PATCH 8/9] Fix triggerCloseHandshake for observability mode. Initialize closeNow to observabilityMode to ensure the call is closed immediately when closing in observability mode, preventing hangs and avoiding duplicate proceedWithClose calls. Update givenObservabilityMode_whenDataPlaneClosed_thenSidecarCloseIsDeferred to assert that onClose is called exactly once, preventing double onClose notifications from passing silently. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../ExternalProcessorClientInterceptor.java | 92 +++++++++---------- ...xternalProcessorClientInterceptorTest.java | 3 + 2 files changed, 47 insertions(+), 48 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index bfa25a66d58..df09e40e735 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -1431,67 +1431,63 @@ private void triggerCloseHandshake() { return; } - if (dataPlaneClientCall.responseDrainComplete.get()) { - proceedWithClose(); - if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { - dataPlaneClientCall.closeExtProcStream(); - } - return; - } - - boolean sendResponseHeaders = - dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() - == ProcessingMode.HeaderSendMode.SEND - || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() - == ProcessingMode.HeaderSendMode.DEFAULT; - - boolean sendResponseTrailers = - dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() - == ProcessingMode.HeaderSendMode.SEND; + boolean closeNow = dataPlaneClientCall.getConfig().getObservabilityMode(); - if (trailersOnly.get()) { - if (sendResponseHeaders) { + if (dataPlaneClientCall.responseDrainComplete.get()) { + closeNow = true; + } else { + boolean sendResponseHeaders = + dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() + == ProcessingMode.HeaderSendMode.SEND + || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode() + == ProcessingMode.HeaderSendMode.DEFAULT; + + boolean sendResponseTrailers = + dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode() + == ProcessingMode.HeaderSendMode.SEND; + + if (trailersOnly.get()) { + if (sendResponseHeaders) { + dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() + .setResponseHeaders(HttpHeaders.newBuilder() + .setHeaders( + toHeaderMap( + savedTrailers, + dataPlaneClientCall.getConfig().getForwardRulesConfig())) + .setEndOfStream(true) + .build()) + .build()); + dataPlaneClientCall.responseTrailersSent.set(true); + } else { + closeNow = true; + } + } else if (sendResponseTrailers) { + dataPlaneClientCall.getIsProcessingTrailers().set(true); dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() - .setResponseHeaders(HttpHeaders.newBuilder() - .setHeaders( + .setResponseTrailers(HttpTrailers.newBuilder() + .setTrailers( toHeaderMap( savedTrailers, dataPlaneClientCall.getConfig().getForwardRulesConfig())) - .setEndOfStream(true) .build()) .build()); dataPlaneClientCall.responseTrailersSent.set(true); } else { - proceedWithClose(); - if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { - dataPlaneClientCall.closeExtProcStream(); - } - } - } else if (sendResponseTrailers) { - dataPlaneClientCall.getIsProcessingTrailers().set(true); - dataPlaneClientCall.sendToExtProc(ProcessingRequest.newBuilder() - .setResponseTrailers(HttpTrailers.newBuilder() - .setTrailers( - toHeaderMap( - savedTrailers, - dataPlaneClientCall.getConfig().getForwardRulesConfig())) - .build()) - .build()); - dataPlaneClientCall.responseTrailersSent.set(true); - } else { - proceedWithClose(); - if (!dataPlaneClientCall.getConfig().getObservabilityMode()) { - dataPlaneClientCall.closeExtProcStream(); + closeNow = true; } } - if (dataPlaneClientCall.getConfig().getObservabilityMode()) { + if (closeNow) { proceedWithClose(); - @SuppressWarnings("unused") - ScheduledFuture unused = dataPlaneClientCall.getScheduler().schedule( - dataPlaneClientCall::closeExtProcStream, - dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(), - TimeUnit.NANOSECONDS); + if (dataPlaneClientCall.getConfig().getObservabilityMode()) { + @SuppressWarnings("unused") + ScheduledFuture unused = dataPlaneClientCall.getScheduler().schedule( + dataPlaneClientCall::closeExtProcStream, + dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(), + TimeUnit.NANOSECONDS); + } else { + dataPlaneClientCall.closeExtProcStream(); + } } } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index e18a4a33e1e..bd9178a8de9 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -10236,9 +10236,11 @@ public void onCompleted() { .build()); try { + final AtomicInteger onCloseCallCount = new AtomicInteger(0); final CountDownLatch appCloseLatch = new CountDownLatch(1); ClientCall.Listener appListener = new ClientCall.Listener() { @Override public void onClose(Status status, Metadata trailers) { + onCloseCallCount.incrementAndGet(); appCloseLatch.countDown(); } }; @@ -10265,6 +10267,7 @@ public void onCompleted() { fakeClock.forwardTime(1, TimeUnit.SECONDS); } assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(onCloseCallCount.get()).isEqualTo(1); // At this point, app received onClose, but sidecar should NOT be completed yet assertThat(sidecarCompletedLatch.getCount()).isEqualTo(1); From 80227c9443025f4ae6d2f03f23f973a2c225e800 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Fri, 28 Aug 2026 12:39:48 +0000 Subject: [PATCH 9/9] Fix checkstyle violations in main and test. Wrap long lines and fix empty/whitespace catch block in test to conform to style guidelines. TAG=agy CONV=2c1e4760-c239-4698-810a-162bf10fccc4 --- .../ExternalProcessorClientInterceptor.java | 26 +++++++++++-------- ...xternalProcessorClientInterceptorTest.java | 10 ++++--- 2 files changed, 22 insertions(+), 14 deletions(-) diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java index df09e40e735..ef417ed349e 100644 --- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java +++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java @@ -622,23 +622,27 @@ public void onError(Throwable t) { @Override public void onCompleted() { - boolean requestDrainRequired = currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.GRPC - && requestBodySentToExtProc.get() - && !requestDrainComplete.get() - && !requestEosSent.get(); - - boolean responseDrainRequired = currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC - && responseBodySentToExtProc.get() - && !responseDrainComplete.get() - && !responseTrailersSent.get(); + boolean requestDrainRequired = + currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.GRPC + && requestBodySentToExtProc.get() + && !requestDrainComplete.get() + && !requestEosSent.get(); + + boolean responseDrainRequired = + currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC + && responseBodySentToExtProc.get() + && !responseDrainComplete.get() + && !responseTrailersSent.get(); if (!config.getObservabilityMode() && (requestDrainRequired || responseDrainRequired)) { if (markExtProcStreamFailed(extProcStreamState)) { synchronized (streamLock) { extProcClientCallRequestObserver = null; } - cancelDownstream("External processor stream completed without drain", - Status.INTERNAL.withDescription("External processor stream completed without drain") + cancelDownstream( + "External processor stream completed without drain", + Status.INTERNAL + .withDescription("External processor stream completed without drain") .asRuntimeException()); wrappedListener.proceedWithClose(); } diff --git a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java index bd9178a8de9..61d0b647cfc 100644 --- a/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java +++ b/xds/src/test/java/io/grpc/xds/ExternalProcessorClientInterceptorTest.java @@ -9211,7 +9211,9 @@ public void onNext(ProcessingRequest request) { .build()); } drainSentLatch.countDown(); - } catch (Exception e) {} + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } }).start(); } else if (request.hasRequestBody()) { if (request.getRequestBody().getDrainComplete()) { @@ -10447,7 +10449,8 @@ public void onClose(Status status, Metadata trailers) { } assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("gRPC message compression not supported in ext_proc"); + assertThat(closedStatus.get().getDescription()) + .contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close(); @@ -10589,7 +10592,8 @@ public void onCompleted() { // Verify application receives INTERNAL with correct description assertThat(closedLatch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL); - assertThat(closedStatus.get().getDescription()).contains("gRPC message compression not supported in ext_proc"); + assertThat(closedStatus.get().getDescription()) + .contains("gRPC message compression not supported in ext_proc"); proxyCall.cancel("Cleanup", null); channelManager.close();