xds: Fix ext_proc client deadlock on unary and server-streaming RPCs when upstream is not ready

Outbound request body messages in ExternalProcessorClientInterceptor were gated on super.isReady() before being forwarded to the upstream ClientCall. On unary and server-streaming RPCs (method.getType().clientSendsOneMessage()), ClientCallImpl intentionally suppresses onReady() notifications, meaning queued messages and halfClose were never drained when the upstream transport stream connected, causing the RPC to hang indefinitely.

Treat the upstream call as ready for outbound message forwarding when method.getType().clientSendsOneMessage() is true via isUpstreamReady().

TAG=agy
CONV=581f86a6-097d-44a2-9d6e-6f9d2e83cbc0

Author: kannanjgithub

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

@@ -934,6 +934,10 @@ boolean isSidecarReady() {
       return observer != null && observer.isReady();
     }
 
+    private boolean isUpstreamReady() {
+      return method.getType().clientSendsOneMessage() || super.isReady();
+    }
+
     @Override
     public boolean isReady() {
       if (passThroughMode.get()) {
@@ -1193,7 +1197,7 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
             boolean sendImmediately = false;
             synchronized (streamLock) {
               sidestreamToUpstreamWindow -= body.size();
-              if (pendingUpstreamBodyMessages.isEmpty() && super.isReady()) {
+              if (pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) {
                 sendImmediately = true;
                 accumulatedWindowUpdateSidestreamToUpstream += body.size();
               } else {
@@ -1307,7 +1311,7 @@ void drainPendingUpstreamBodyMessages() {
         ByteString body = null;
         boolean triggerHalfClose = false;
         synchronized (streamLock) {
-          if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) {
+          if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) {
             body = pendingUpstreamBodyMessages.poll();
             accumulatedWindowUpdateSidestreamToUpstream += body.size();
             if (pendingUpstreamBodyMessages.isEmpty()
@@ -1366,16 +1370,16 @@ private void drainPendingDrainingMessages() {
         boolean triggerHalfClose = false;
 
         synchronized (streamLock) {
-          if (!pendingUpstreamBodyMessages.isEmpty() && super.isReady()) {
+          if (!pendingUpstreamBodyMessages.isEmpty() && isUpstreamReady()) {
             msg = pendingUpstreamBodyMessages.poll();
             isMutated = true;
           } else if (pendingUpstreamBodyMessages.isEmpty()
-              && !pendingRequestBodyMessages.isEmpty() && super.isReady()) {
+              && !pendingRequestBodyMessages.isEmpty() && isUpstreamReady()) {
             msg = pendingRequestBodyMessages.poll();
             isMutated = true;
           } else if (pendingUpstreamBodyMessages.isEmpty()
               && pendingRequestBodyMessages.isEmpty()
-              && !pendingDrainingMessages.isEmpty() && super.isReady()) {
+              && !pendingDrainingMessages.isEmpty() && isUpstreamReady()) {
             msg = pendingDrainingMessages.poll();
             isMutated = false;
           }