Fix sendMessage race condition and test expectations.

1. Remove requestDrainComplete check from sendMessage to prevent out-of-order message delivery when app calls sendMessage concurrently with drain completion. Now we only bypass queue when passThroughMode is true.
2. Update givenExtProcStreamCompleted_whenIsReadyCalled_thenDelegatesToSuper test to expect the correct number of downstream isReady calls, which increased by 1 because we now correctly query isReady inside onReadyNotify on stream completion.

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

Author: kannanjgithub

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

@@ -849,13 +849,13 @@ public void sendMessage(InputStream message) {
         return;
       }
 
-      if (passThroughMode.get() || requestDrainComplete.get()) {
+      if (passThroughMode.get()) {
         super.sendMessage(message);
         return;
       }
 
       synchronized (streamLock) {
-        if (passThroughMode.get() || requestDrainComplete.get()) {
+        if (passThroughMode.get()) {
           super.sendMessage(message);
           return;
         }
@@ -1394,6 +1394,7 @@ void unblockAfterStreamComplete() {
       proceedWithHeaders();
       proceedWithSavedMessages();
       dataPlaneClientCall.drainPendingDrainingMessages();
+      onReadyNotify();
       proceedWithClose();
     }
 

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

@@ -6070,11 +6070,11 @@ public boolean isReady() {
     // 4. Assert that proxyCall.isReady() delegates directly to the downstream call
     downstreamReady.set(true);
     assertThat(proxyCall.isReady()).isTrue();
-    assertThat(downstreamIsReadyCallCount.get()).isEqualTo(1);
+    assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2);
 
     downstreamReady.set(false);
     assertThat(proxyCall.isReady()).isFalse();
-    assertThat(downstreamIsReadyCallCount.get()).isEqualTo(2);
+    assertThat(downstreamIsReadyCallCount.get()).isEqualTo(3);
 
     proxyCall.cancel("cleanup", null);
     channelManager.close();