Fix flow control bypass during client response draining in ext_proc.

When activateCall is called during draining (initiated by response drain), it flushes requests buffered in DelayedClientCall directly to the rawCall, bypassing flow control checks.

To fix this:
1. Buffer request(n) calls in pendingRequests when the call is IDLE, instead of letting them go to DelayedClientCall.
2. Guard drainPendingRequests to check isResponseSidecarReady and abort if it is false (keeping requests buffered).
3. Trigger drainPendingRequests when response drain completes.

Also updated tests to verify early request buffering during response drain.

TAG=agy
CONV=2c1e4760-c239-4698-810a-162bf10fccc4

Author: kannanjgithub

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

@@ -726,6 +726,9 @@ private void onExtProcStreamReady() {
     }
 
     private void drainPendingRequests() {
+      if (!isResponseSidecarReady()) {
+        return;
+      }
       int toRequest = pendingRequests.getAndSet(0);
       if (toRequest > 0) {
         super.request(toRequest);
@@ -828,6 +831,10 @@ public void request(int numMessages) {
         super.request(numMessages);
         return;
       }
+      if (dataPlaneCallState.get() == DataPlaneCallState.IDLE) {
+        pendingRequests.addAndGet(numMessages);
+        return;
+      }
       if (!isResponseSidecarReady()) {
         pendingRequests.addAndGet(numMessages);
         return;
@@ -1391,6 +1398,7 @@ void unblockAfterResponseDrain() {
       }
       proceedWithHeaders();
       proceedWithSavedMessages();
+      dataPlaneClientCall.drainPendingRequests();
       proceedWithClose();
     }
 

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

@@ -9020,6 +9020,8 @@ public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffere
     ExternalProcessorFilterConfig filterConfig = configOrError.config;
 
     final CountDownLatch drainSentLatch = new CountDownLatch(1);
+    final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
+    final CountDownLatch sendDrainLatch = new CountDownLatch(1);
     // External Processor Server
     ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl;
     extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() {
@@ -9032,10 +9034,20 @@ public StreamObserver<ProcessingRequest> process(
           @Override
           public void onNext(ProcessingRequest request) {
             if (request.hasRequestHeaders()) {
-              responseObserver.onNext(ProcessingResponse.newBuilder()
-                  .setRequestDrainResponses(true)
-                  .build());
-              drainSentLatch.countDown();
+              headersReceivedLatch.countDown();
+              new Thread(() -> {
+                try {
+                  sendDrainLatch.await();
+                  synchronized (responseObserver) {
+                    responseObserver.onNext(ProcessingResponse.newBuilder()
+                        .setRequestDrainResponses(true)
+                        .build());
+                  }
+                  drainSentLatch.countDown();
+                } catch (Exception e) {
+                  System.out.println("JetskiDebug: error in mock server: " + e);
+                }
+              }).start();
             }
           }
 
@@ -9094,15 +9106,21 @@ public void request(int numMessages) {
     ClientCall<String, String> proxyCall =
         interceptCall(interceptor, METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
     proxyCall.start(new ClientCall.Listener<String>() {}, new Metadata());
+    
+    // Wait for headers to reach mock server
+    assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // App requests messages early (before drain is received)
+    proxyCall.request(3);
+
+    // Now trigger drain
+    sendDrainLatch.countDown();
 
     // Wait for drain to be processed
     assertThat(drainSentLatch.await(5, TimeUnit.SECONDS)).isTrue();
     // proxyCall.isReady() should remain true during response drain (request path is NOT draining)
     assertThat(proxyCall.isReady()).isTrue();
 
-    // App requests more messages
-    proxyCall.request(3);
-
     // Verify requests are buffered and not sent to data plane
     assertThat(dataPlaneRequestCount.get()).isEqualTo(0);