xds: Fix race condition in ext-proc fail-open client interceptor test

Resolve flakiness in ExternalProcessorClientInterceptorTest's
givenFailureModeAllowTrue_whenExtProcStreamFails_thenCallFailsOpen test.
The test was racing because the mock sidecar failed asynchronously
on headers, occasionally running after the client had already sent
a request message. If the message was sent first, fail-open was
disallowed because of data loss.

Fixed by intercepting the data plane server to block the client thread
until the data plane call actually starts (headers received). This
ensures the client interceptor has processed the ext-proc stream
failure and transitioned to fail-open pass-through mode before
the application sends the request body.

Author: kannanjgithub

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

@@ -8460,14 +8460,25 @@ public void onCompleted() {
         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());
+    final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
+    ServerInterceptor dataPlaneInterceptor = new ServerInterceptor() {
+      @Override
+      public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+          ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+        headersReceivedLatch.countDown();
+        return next.startCall(call, headers);
+      }
+    };
+    dataPlaneServiceRegistry.addService(ServerInterceptors.intercept(
+        ServerServiceDefinition.builder("test.TestService")
+            .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall(
+                (request, responseObserver) -> {
+                  responseObserver.onNext("Hello " + request);
+                  responseObserver.onCompleted();
+                  dataPlaneLatch.countDown();
+                }))
+            .build(),
+        dataPlaneInterceptor));
 
     ManagedChannel dataPlaneChannel = grpcCleanup.register(
         InProcessChannelBuilder.forName(dataPlaneServerName).directExecutor().build());
@@ -8487,6 +8498,7 @@ public void onClose(Status status, Metadata trailers) {
 
     // Send message and half-close to trigger unary call reaching server
     proxyCall.request(1);
+    assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue();
     proxyCall.sendMessage("test");
     proxyCall.halfClose();