xds: treat ext_proc stream OK termination without drain as non-OK status

Update ext_proc client interceptor to treat OK termination without drain as non-OK status regardless of whether messages have started being sent to the ext_proc server, in accordance with the updated gRFC specification.

TAG=agy
CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0

Author: kannanjgithub

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

@@ -669,11 +669,18 @@ public void onError(Throwable t) {
 
         @Override
         public void onCompleted() {
-          if (markExtProcStreamCompleted(extProcStreamState)) {
-            synchronized (streamLock) {
-              extProcClientCallRequestObserver = null;
+          ExtProcStreamState state = extProcStreamState.get();
+          if (state == ExtProcStreamState.DRAINING || config.getObservabilityMode()) {
+            if (markExtProcStreamCompleted(extProcStreamState)) {
+              synchronized (streamLock) {
+                extProcClientCallRequestObserver = null;
+              }
+              handleFailOpen(wrappedListener);
             }
-            handleFailOpen(wrappedListener);
+          } else if (state == ExtProcStreamState.ACTIVE) {
+            internalOnError(Status.UNAVAILABLE
+                .withDescription("External processor stream completed without drain")
+                .asRuntimeException());
           }
         }
       });
@@ -849,7 +856,11 @@ private void internalOnError(Throwable t) {
       if (markExtProcStreamFailed(extProcStreamState)) {
         synchronized (streamLock) {
           if (extProcClientCallRequestObserver != null) {
-            extProcClientCallRequestObserver.onError(t);
+            try {
+              extProcClientCallRequestObserver.onError(t);
+            } catch (Throwable ignored) {
+              // Ignore exception if observer is already closed/completed
+            }
             extProcClientCallRequestObserver = null;
           }
         }

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

@@ -6554,6 +6554,7 @@ public void onNext(ProcessingRequest request) {
                 if (request.hasRequestHeaders()) {
                   responseObserver.onNext(ProcessingResponse.newBuilder()
                       .setRequestHeaders(HeadersResponse.newBuilder().build())
+                      .setRequestDrain(true)
                       .build());
                   responseObserver.onCompleted();
                   sidecarResponseLatch.countDown();
@@ -9925,9 +9926,10 @@ public void request(int numMessages) {
 
   @Test
   @SuppressWarnings("unchecked")
-  public void givenExtProcStreamCompleted_whenAppRequestsMessages_thenRequestsForwarded()
+  public void givenNoRequestDrain_whenExtProcStreamCompletesNormally_thenTreatsAsNonOkAndCallFails()
       throws Exception {
     ExternalProcessor proto = ExternalProcessor.newBuilder()
+        .setFailureModeAllow(false) // Fail closed
         .setGrpcService(GrpcService.newBuilder()
             .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
                 .setTargetUri("in-process:///" + extProcServerName)
@@ -9943,9 +9945,9 @@ public void givenExtProcStreamCompleted_whenAppRequestsMessages_thenRequestsForw
     assertThat(configOrError.errorDetail).isNull();
     ExternalProcessorFilterConfig filterConfig = configOrError.config;
 
-    // External Processor Server
-    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl;
-    extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() {
+    // External Processor Server completes normally without sending request_drain = true
+    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
+        new ExternalProcessorGrpc.ExternalProcessorImplBase() {
       @Override
       @SuppressWarnings("unchecked")
       public StreamObserver<ProcessingRequest> process(
@@ -9955,7 +9957,7 @@ public StreamObserver<ProcessingRequest> process(
           @Override
           public void onNext(ProcessingRequest request) {
             if (request.hasRequestHeaders()) {
-              // Immediately complete the stream from server side
+              // Immediately complete the stream from server side without drain
               responseObserver.onCompleted();
             }
           }
@@ -9991,49 +9993,32 @@ public void onCompleted() {
             }))
         .build());
 
-    final AtomicInteger dataPlaneRequestCount = new AtomicInteger(0);
     ManagedChannel dataPlaneChannel = grpcCleanup.register(
         InProcessChannelBuilder.forName(dataPlaneServerName)
             .directExecutor()
-            .intercept(new ClientInterceptor() {
-                @Override
-                public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
-                    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
-                  return new io.grpc.ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
-                      next.newCall(method, callOptions)) {
-                    @Override
-                    public void request(int numMessages) {
-                      dataPlaneRequestCount.addAndGet(numMessages);
-                      super.request(numMessages);
-                    }
-                  };
-                }
-            })
             .build());
 
-    final CountDownLatch readyLatch = new CountDownLatch(1);
+    final CountDownLatch callCompletedLatch = new CountDownLatch(1);
+    final AtomicReference<Status> closedStatus = new AtomicReference<>();
     CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor());
     ClientCall<String, String> proxyCall =
         interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
     proxyCall.start(new ClientCall.Listener<String>() {
       @Override
-      public void onReady() {
-        readyLatch.countDown();
+      public void onClose(Status status, Metadata trailers) {
+        closedStatus.set(status);
+        callCompletedLatch.countDown();
       }
     }, new Metadata());
 
-    // Wait for sidecar stream completion
-    assertThat(readyLatch.await(5, TimeUnit.SECONDS)).isTrue();
-    assertThat(proxyCall.isReady()).isTrue();
+    proxyCall.request(1);
+    proxyCall.sendMessage("hello");
+    proxyCall.halfClose();
 
-    proxyCall.request(7);
+    assertThat(callCompletedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(closedStatus.get().getCode()).isEqualTo(Status.Code.INTERNAL);
+    assertThat(closedStatus.get().getDescription()).contains("External processor stream failed");
 
-    // Verify request forwarded immediately
-    assertThat(dataPlaneRequestCount.get()).isEqualTo(7);
-    // proxyCall.isReady() should remain true as sidecar is gone
-    assertThat(proxyCall.isReady()).isTrue();
-    
-    proxyCall.cancel("Cleanup", null);
     channelManager.close();
   }