xds: fix halfClose in ext_proc observability mode when request body mode is GRPC and add test

TAG=agy
CONV=56c87717-243d-4f12-af9e-c532e509892a

Author: kannanjgithub

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

@@ -1151,6 +1151,11 @@ public void halfClose() {
         mergeAccumulatedWindowUpdates(builder);
         sendToExtProc(builder.build());
       }
+      if (config.getObservabilityMode()) {
+        if (requestSideClosed.compareAndSet(false, true)) {
+          proceedWithHalfClose();
+        }
+      }
     }
 
     void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {

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

@@ -11298,6 +11298,138 @@ public void onClose(Status status, Metadata trailers) {
     channelManager.close();
   }
 
+  @Test
+  public void testObservabilityMode_SendsRequestAndResponseBodiesWhenBodyModeGrpc()
+      throws Exception {
+    String uniqueExtProcServerName = InProcessServerBuilder.generateName();
+    ExternalProcessor proto = ExternalProcessor.newBuilder()
+        .setGrpcService(GrpcService.newBuilder()
+            .setGoogleGrpc(GrpcService.GoogleGrpc.newBuilder()
+                .setTargetUri("in-process:///" + uniqueExtProcServerName)
+                .addChannelCredentialsPlugin(Any.newBuilder()
+                    .setTypeUrl("type.googleapis.com/envoy.extensions.grpc_service."
+                        + "channel_credentials.insecure.v3.InsecureCredentials")
+                    .build())
+                .build())
+            .build())
+        .setObservabilityMode(true)
+        .setProcessingMode(ProcessingMode.newBuilder()
+            .setRequestBodyMode(ProcessingMode.BodySendMode.GRPC)
+            .setResponseBodyMode(ProcessingMode.BodySendMode.GRPC)
+            .setResponseTrailerMode(ProcessingMode.HeaderSendMode.SEND)
+            .build())
+        .build();
+    ConfigOrError<ExternalProcessorFilterConfig> configOrError =
+        provider.parseFilterConfig(Any.pack(proto), filterContext);
+    assertThat(configOrError.errorDetail).isNull();
+    ExternalProcessorFilterConfig filterConfig = configOrError.config;
+
+    final CountDownLatch extProcReceivedReqBodyLatch = new CountDownLatch(1);
+    final CountDownLatch extProcReceivedRespBodyLatch = new CountDownLatch(1);
+    final AtomicReference<ProcessingRequest> extProcReqBodyRequest = new AtomicReference<>();
+    final AtomicReference<ProcessingRequest> extProcRespBodyRequest = new AtomicReference<>();
+    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
+        new ExternalProcessorGrpc.ExternalProcessorImplBase() {
+          @Override
+          @SuppressWarnings("unchecked")
+          public StreamObserver<ProcessingRequest> process(
+              final StreamObserver<ProcessingResponse> responseObserver) {
+            ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
+            return new StreamObserver<ProcessingRequest>() {
+              @Override
+              public void onNext(ProcessingRequest request) {
+                if (request.hasRequestBody() && !request.getRequestBody().getEndOfStream()) {
+                  extProcReqBodyRequest.set(request);
+                  extProcReceivedReqBodyLatch.countDown();
+                } else if (request.hasResponseBody()
+                    && !request.getResponseBody().getEndOfStream()) {
+                  extProcRespBodyRequest.set(request);
+                  extProcReceivedRespBodyLatch.countDown();
+                }
+              }
+
+              @Override
+              public void onError(Throwable t) {}
+
+              @Override
+              public void onCompleted() {}
+            };
+          }
+        };
+    grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
+        .addService(extProcImpl)
+        .directExecutor()
+        .build().start());
+
+    CachedChannelManager channelManager = new CachedChannelManager(config -> {
+      return grpcCleanup.register(
+          InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
+    });
+
+    ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
+        filterConfig, channelManager, scheduler, FAKE_CONTEXT);
+
+    final CountDownLatch dataPlaneLatch = new CountDownLatch(1);
+    dataPlaneServiceRegistry.addService(ServerServiceDefinition.builder("test.TestService")
+        .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
+            (request, responseObserver) -> {
+              responseObserver.onNext("Hello " + request);
+              responseObserver.onCompleted();
+              dataPlaneLatch.countDown();
+            }))
+        .build());
+
+    ManagedChannel dataPlaneChannel = grpcCleanup.register(
+        InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
+
+    final List<String> appReceivedMessages = new java.util.concurrent.CopyOnWriteArrayList<>();
+    final CountDownLatch appMessageLatch = new CountDownLatch(1);
+    final CountDownLatch appCloseLatch = new CountDownLatch(1);
+    final AtomicReference<Status> appReceivedStatus = new AtomicReference<>();
+
+    ClientCall.Listener<String> appListener = new ClientCall.Listener<String>() {
+      @Override
+      public void onMessage(String message) {
+        appReceivedMessages.add(message);
+        appMessageLatch.countDown();
+      }
+
+      @Override
+      public void onClose(Status status, Metadata trailers) {
+        appReceivedStatus.set(status);
+        appCloseLatch.countDown();
+      }
+    };
+
+    CallOptions callOptions = DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor());
+    ClientCall<String, String> proxyCall =
+        interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
+    proxyCall.start(appListener, new Metadata());
+    proxyCall.request(1);
+
+    proxyCall.sendMessage("test");
+    proxyCall.halfClose();
+
+    assertThat(dataPlaneLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(appMessageLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(appReceivedMessages).containsExactly("Hello test");
+    assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(appReceivedStatus.get().isOk()).isTrue();
+
+    assertThat(extProcReceivedReqBodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(extProcReqBodyRequest.get().getObservabilityMode()).isTrue();
+    assertThat(extProcReqBodyRequest.get().getRequestBody().getBody().toStringUtf8())
+        .isEqualTo("test");
+
+    assertThat(extProcReceivedRespBodyLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(extProcRespBodyRequest.get().getObservabilityMode()).isTrue();
+    assertThat(extProcRespBodyRequest.get().getResponseBody().getBody().toStringUtf8())
+        .isEqualTo("Hello test");
+
+    proxyCall.cancel("Cleanup", null);
+    channelManager.close();
+  }
+
   // --- Category 18: Immediate Response Handling ---
 
   @Test