Flow control backpush should only apply in observability mode (not normal mode) and in ext-proc request draining time period. Fixed deviances from this expectation.

Author: kannanjgithub

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

@@ -189,10 +189,9 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
           .withExecutor(callOptions.getExecutor());
 
       if (filterConfig.grpcServiceConfig.timeout() != null && filterConfig.grpcServiceConfig.timeout().isPresent()) {
-        long timeoutSeconds = filterConfig.grpcServiceConfig.timeout().get().getSeconds();
-        int timeoutNanos = filterConfig.grpcServiceConfig.timeout().get().getNano();
-        if (timeoutSeconds > 0 || timeoutNanos > 0) {
-          stub = stub.withDeadlineAfter(timeoutSeconds * 1_000_000_000L + timeoutNanos, TimeUnit.NANOSECONDS);
+        long timeoutNanos = filterConfig.grpcServiceConfig.timeout().get().toNanos();
+        if (timeoutNanos > 0) {
+          stub = stub.withDeadlineAfter(timeoutNanos, TimeUnit.NANOSECONDS);
         }
       }
 
@@ -452,7 +451,7 @@ public void onNext(ProcessingResponse response) {
 
               if (response.getRequestDrain()) {
                 drainingExtProcStream.set(true);
-                closeExtProcStream();
+                halfCloseExtProcStream();
                 return;
               }
 
@@ -589,9 +588,7 @@ private void sendToExtProc(ProcessingRequest request) {
 
       private void onExtProcStreamReady() {
         drainPendingRequests();
-        if (isReady()) {
-          wrappedListener.onReadyNotify();
-        }
+        onReadyNotify();
       }
 
       private void drainPendingRequests() {
@@ -617,6 +614,24 @@ private void closeExtProcStream() {
         }
       }
 
+      private void halfCloseExtProcStream() {
+        synchronized (streamLock) {
+          if (!extProcStreamCompleted.get() && extProcClientCallRequestObserver != null) {
+            try {
+              extProcClientCallRequestObserver.onCompleted();
+            } catch (IllegalStateException | io.grpc.StatusRuntimeException e) {
+              // Ignore
+            }
+          }
+        }
+      }
+
+      private void onReadyNotify() {
+        if (isReady()) {
+          wrappedListener.onReadyNotify();
+        }
+      }
+
       @Override
       public boolean isReady() {
         if (extProcStreamCompleted.get()) {
@@ -625,9 +640,7 @@ public boolean isReady() {
         if (drainingExtProcStream.get()) {
           return false;
         }
-        boolean sendingBody = config.getProcessingMode().getRequestBodyMode()
-            == ProcessingMode.BodySendMode.GRPC;
-        if (config.getObservabilityMode() || sendingBody) {
+        if (config.getObservabilityMode()) {
           synchronized (streamLock) {
             return super.isReady() && extProcClientCallRequestObserver != null
                 && extProcClientCallRequestObserver.isReady();
@@ -638,14 +651,26 @@ public boolean isReady() {
 
       @Override
       public void request(int numMessages) {
+        if (extProcStreamCompleted.get()) {
+          super.request(numMessages);
+          return;
+        }
         // If the external processor is backed up with flow control, we need to stop requesting
         // messages from the remote side.
-        synchronized (streamLock) {
-          if (!isReady()) {
+        if (drainingExtProcStream.get()) {
+          synchronized (streamLock) {
             pendingRequests += numMessages;
             return;
           }
         }
+        if (config.getObservabilityMode()) {
+          synchronized (streamLock) {
+            if (!isReady()) {
+              pendingRequests += numMessages;
+              return;
+            }
+          }
+        }
         super.request(numMessages);
       }
 
@@ -779,9 +804,6 @@ public void onReady() {
       }
 
       void onReadyNotify() {
-        if (extProcClientCall.drainingExtProcStream.get()) {
-          return;
-        }
         if (extProcClientCall.isReady()) {
           super.onReady();
         }
@@ -901,6 +923,7 @@ void onExternalBody(com.google.protobuf.ByteString body) {
 
       void unblockAfterStreamComplete() {
         proceedWithHeaders();
+        onReadyNotify();
         proceedWithClose();
       }
     }