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

Author: kannanjgithub

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();
+        }
       }
     }
 

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<String> appListener = new ClientCall.Listener<String>() {
         @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);