xds: bypass sidecar readiness for non-GRPC responses & fix draining conformance in ext_proc filter

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

@@ -768,6 +768,11 @@ public void request(int numMessages) {
         super.request(numMessages);
         return;
       }
+      if (!config.getObservabilityMode() 
+          && currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
+        super.request(numMessages);
+        return;
+      }
       if (!isSidecarReady()) {
         pendingRequests.addAndGet(numMessages);
         return;
@@ -795,6 +800,10 @@ public void sendMessage(InputStream message) {
 
         ExtProcStreamState state = extProcStreamState.get();
         if (state.isDraining() || state.isCompleted()) {
+          if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE) {
+            super.sendMessage(message);
+            return;
+          }
           try {
             ByteString copiedBody = ByteString.readFrom(message);
             pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
@@ -855,6 +864,18 @@ public void halfClose() {
       }
 
       if (extProcStreamState.get().isDraining()) {
+        boolean canProceed = false;
+        synchronized (streamLock) {
+          if (currentProcessingMode.getRequestBodyMode() == ProcessingMode.BodySendMode.NONE
+              || (!bodyMessageSentToExtProc.get() && pendingDrainingMessages.isEmpty())) {
+            canProceed = true;
+          }
+        }
+        if (canProceed) {
+          if (requestSideClosed.compareAndSet(false, true)) {
+            proceedWithHalfClose();
+          }
+        }
         return;
       }
 
@@ -1072,16 +1093,17 @@ public void onReady() {
     public void onHeaders(Metadata headers) {
       dataPlaneClientCall.setServerHeadersStartNanos(System.nanoTime());
       responseHeadersSent.set(true);
-      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining()) {
-        this.savedHeaders = headers;
-        return;
-      }
       boolean sendResponseHeaders =
           dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
               == ProcessingMode.HeaderSendMode.SEND
           || dataPlaneClientCall.getCurrentProcessingMode().getResponseHeaderMode()
               == ProcessingMode.HeaderSendMode.DEFAULT;
 
+      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseHeaders) {
+        this.savedHeaders = headers;
+        return;
+      }
+
       if (dataPlaneClientCall.getPassThroughMode().get() 
           || dataPlaneClientCall.getExtProcStreamState().get().isCompleted() 
           || !sendResponseHeaders) {
@@ -1110,8 +1132,11 @@ public void onMessage(InputStream message) {
           return;
         }
 
-        if (savedHeaders != null
-            || dataPlaneClientCall.getExtProcStreamState().get().isDraining()) {
+        boolean checkDrain = dataPlaneClientCall.getExtProcStreamState().get().isDraining()
+            && dataPlaneClientCall.getCurrentProcessingMode().getResponseBodyMode()
+                == ProcessingMode.BodySendMode.GRPC;
+
+        if (savedHeaders != null || checkDrain) {
           try {
             ByteString copiedBody = ByteString.readFrom(message);
             savedMessages.add(new KnownLengthInputStream(copiedBody));
@@ -1184,7 +1209,11 @@ public void onClose(Status status, Metadata trailers) {
         return;
       }
 
-      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining()) {
+      boolean sendResponseTrailers =
+          dataPlaneClientCall.getCurrentProcessingMode().getResponseTrailerMode()
+              == ProcessingMode.HeaderSendMode.SEND;
+
+      if (dataPlaneClientCall.getExtProcStreamState().get().isDraining() && sendResponseTrailers) {
         return;
       }
 
@@ -1193,15 +1222,6 @@ public void onClose(Status status, Metadata trailers) {
       }
 
       triggerCloseHandshake();
-
-      if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
-        proceedWithClose();
-        @SuppressWarnings("unused")
-        ScheduledFuture<?> unused = dataPlaneClientCall.getScheduler().schedule(
-            dataPlaneClientCall::closeExtProcStream,
-            dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(),
-            TimeUnit.NANOSECONDS);
-      }
     }
 
     void onReadyNotify() {
@@ -1330,6 +1350,15 @@ private void triggerCloseHandshake() {
           dataPlaneClientCall.closeExtProcStream();
         }
       }
+
+      if (dataPlaneClientCall.getConfig().getObservabilityMode()) {
+        proceedWithClose();
+        @SuppressWarnings("unused")
+        ScheduledFuture<?> unused = dataPlaneClientCall.getScheduler().schedule(
+            dataPlaneClientCall::closeExtProcStream,
+            dataPlaneClientCall.getConfig().getDeferredCloseTimeoutNanos(),
+            TimeUnit.NANOSECONDS);
+      }
     }
 
     private void sendResponseBodyToExtProc(