The stream between the filter and the external processor was never being closed on the client side, causing the InProcessChannel and InProcessServer to hang during shutdown while waiting for the active RPC to terminate.

To fix this, I have updated ExternalProcessorFilter.java to ensure the control plane stream is gracefully closed when the data plane RPC completes or is cancelled.
Changes made:
1. Closing on Completion: In ExtProcClientCall.onNext, once the ResponseTrailers handshake is finished and the application has been notified via proceedWithClose(), I now call extProcClientCallRequestObserver.onCompleted().
2. Handling Cancellation: I overridden the cancel() method in ExtProcClientCall. If the data plane RPC is cancelled by the application, the filter now also cancels the external processor stream with an error, ensuring all resources are freed.
3. Observability Mode Fix: In observability mode, since we don't wait for a ResponseTrailers message from the server, I added logic to ExtProcListener.onClose() to close the external processor stream immediately after sending the final trailers.
These changes ensure proper lifecycle management of the side-channel RPC.

Author: kannanjgithub

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

@@ -441,6 +441,7 @@ else if (response.hasResponseBody()) {
               }
               // Finally notify the local app of the completion
               wrappedListener.proceedWithClose();
+              extProcClientCallRequestObserver.onCompleted();
             }
           }
 
@@ -544,6 +545,14 @@ public void halfClose() {
         super.halfClose();
       }
 
+      @Override
+      public void cancel(@Nullable String message, @Nullable Throwable cause) {
+        if (extProcClientCallRequestObserver != null) {
+          extProcClientCallRequestObserver.onError(Status.CANCELLED.withDescription(message).withCause(cause).asRuntimeException());
+        }
+        super.cancel(message, cause);
+      }
+
       private void handleRequestBodyResponse(io.envoyproxy.envoy.service.ext_proc.v3.BodyResponse bodyResponse) {
         if (bodyResponse.hasResponse() && bodyResponse.getResponse().hasBodyMutation()) {
           io.envoyproxy.envoy.service.ext_proc.v3.BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
@@ -691,6 +700,7 @@ public void onClose(io.grpc.Status status, Metadata trailers) {
 
         if (extProcClientCall.config.getObservabilityMode()) {
           super.onClose(status, trailers);
+          extProcClientCall.extProcClientCallRequestObserver.onCompleted();
         }
       }