Fix onReady notification flow control and verification.

1. Fix onReadyNotify in DataPlaneListener to only forward onReady if dataPlaneClientCall.isReady() is true.
2. Call onReadyNotify in handleRequestDrainComplete to notify the app when request draining completes and the path becomes ready again.
3. Update givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained to verify that onReady is NOT called during request drain but IS called after the drain completes.

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

Author: kannanjgithub

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

@@ -1064,6 +1064,9 @@ private void handleRequestDrainComplete() {
       if (requestDraining.compareAndSet(true, false)) {
         requestDrainComplete.set(true);
         drainPendingDrainingMessages();
+        if (wrappedListener != null) {
+          wrappedListener.onReadyNotify();
+        }
       }
     }
 
@@ -1327,7 +1330,9 @@ public void onClose(Status status, Metadata trailers) {
     }
 
     void onReadyNotify() {
-      dataPlaneClientCall.getCallContext().run(() -> delegate().onReady());
+      if (dataPlaneClientCall.isReady()) {
+        dataPlaneClientCall.getCallContext().run(() -> delegate().onReady());
+      }
     }
 
     void proceedWithHeaders() {

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

@@ -9185,13 +9185,16 @@ public void givenRequestDrainActive_whenAppRequestsMessages_thenRequestsDrained(
     final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
     final CountDownLatch sendDrainLatch = new CountDownLatch(1);
     final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1);
+    final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
+        new AtomicReference<>();
     // External Processor Server
     ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl;
     extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() {
       @Override
       @SuppressWarnings("unchecked")
       public StreamObserver<ProcessingRequest> process(
           final StreamObserver<ProcessingResponse> responseObserver) {
+        responseObserverRef.set(responseObserver);
         ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
         return new StreamObserver<ProcessingRequest>() {
           @Override
@@ -9203,6 +9206,7 @@ public void onNext(ProcessingRequest request) {
                   sendDrainLatch.await();
                   synchronized (responseObserver) {
                     responseObserver.onNext(ProcessingResponse.newBuilder()
+                        .setRequestHeaders(HeadersResponse.newBuilder().build())
                         .setRequestDrainRequests(true)
                         .build());
                   }
@@ -9270,7 +9274,8 @@ public void request(int numMessages) {
     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>() {}, new Metadata());
+    ClientCall.Listener<String> mockListener = Mockito.mock(ClientCall.Listener.class);
+    proxyCall.start(mockListener, new Metadata());
 
     // Wait for headers to reach mock server
     assertThat(headersReceivedLatch.await(5, TimeUnit.SECONDS)).isTrue();
@@ -9289,10 +9294,31 @@ public void request(int numMessages) {
 
     // Verify requests are now drained to data plane (request drain does not block response path)
     assertThat(dataPlaneRequestCount.get()).isEqualTo(3);
+
+    // Verify onReady was NOT called because request path is draining
+    Mockito.verify(mockListener, Mockito.never()).onReady();
 
     // Wait for filter to send drain_complete to mock server
     assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue();
 
+    // Echo drain_complete back to filter to complete handshake
+    synchronized (responseObserverRef.get()) {
+      responseObserverRef.get().onNext(ProcessingResponse.newBuilder()
+          .setRequestBody(BodyResponse.newBuilder()
+              .setResponse(CommonResponse.newBuilder()
+                  .setBodyMutation(BodyMutation.newBuilder()
+                      .setStreamedResponse(StreamedBodyResponse.newBuilder()
+                          .setDrainComplete(true)
+                          .build())
+                      .build())
+                  .build())
+              .build())
+          .build());
+    }
+
+    // Verify onReady IS now called after request drain completes
+    Mockito.verify(mockListener, Mockito.timeout(5000)).onReady();
+
     proxyCall.cancel("Cleanup", null);
     channelManager.close();
   }