Style fixes.

Author: kannanjgithub

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

@@ -197,15 +197,7 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
       MethodDescriptor<ReqT, RespT> method,
       CallOptions callOptions,
       Channel next) {
-    java.util.concurrent.Executor callExecutor = callOptions.getExecutor();
-    if (callExecutor == null) {
-      callExecutor = new java.util.concurrent.Executor() {
-        @Override
-        public void execute(Runnable command) {
-          command.run();
-        }
-      };
-    }
+    Executor callExecutor = callOptions.getExecutor();
     SerializingExecutor serializingExecutor = new SerializingExecutor(callExecutor);
 
     ExternalProcessorGrpc.ExternalProcessorStub extProcStub = ExternalProcessorGrpc.newStub(
@@ -314,7 +306,7 @@ private static class DataPlaneClientCall
     private final Queue<ByteString> pendingRequestBodyMessages =
         new java.util.concurrent.ConcurrentLinkedQueue<>();
 
-    // Buffered mutated response bodies from sidecar
+    // Buffered mutated response bodies from ext_proc server
     private int downstreamRequestsPending = 0;
     private final Queue<ByteString> pendingMutatedResponseBodies =
         new java.util.concurrent.ConcurrentLinkedQueue<>();
@@ -488,10 +480,12 @@ public void onNext(ProcessingResponse response) {
                 upstreamToSidestreamWindow += update.getWindowIncrementUpstreamToSidestream();
                 drainPendingRequestBodyMessages();
                 drainPendingRequests();
+                if (wrappedListener != null) {
+                  wrappedListener.drainSavedMessages();
+                }
               }
-              if (wrappedListener != null) {
-                wrappedListener.drainSavedMessages();
-              }
+              // If isReady() becomes true (depends on updated downstreamToSidestreamWindow),
+              // notify the client application via onReadyNotify() (runs unlocked).
               if (!wasReady && isReady()) {
                 onReadyNotify();
               }
@@ -811,7 +805,7 @@ void drainPendingRequests() {
         }
 
         // Normal mode flow control: pull 1 message at a time
-        if (isSidecarReady() && upstreamToSidestreamWindow > 0 && pendingRequests.get() > 0) {
+        if (isExtProcReady() && upstreamToSidestreamWindow > 0 && pendingRequests.get() > 0) {
           super.request(1);
           pendingRequests.decrementAndGet();
         }
@@ -863,7 +857,7 @@ private void onReadyNotify() {
       wrappedListener.onReadyNotify();
     }
 
-    boolean isSidecarReady() {
+    boolean isExtProcReady() {
       ExtProcStreamState state = extProcStreamState.get();
       if (state.isCompleted()) {
         return true;
@@ -889,11 +883,11 @@ public boolean isReady() {
         return false;
       }
       synchronized (streamLock) {
-        boolean sidecarReady = isSidecarReady();
+        boolean extProcReady = isExtProcReady();
         if (config.getObservabilityMode()) {
-          return super.isReady() && sidecarReady;
+          return super.isReady() && extProcReady;
         }
-        return downstreamToSidestreamWindow > 0 && sidecarReady
+        return downstreamToSidestreamWindow > 0 && extProcReady
             && pendingRequestBodyMessages.isEmpty();
       }
     }
@@ -904,22 +898,34 @@ public void request(int numMessages) {
         super.request(numMessages);
         return;
       }
-      if (!config.getObservabilityMode() 
-          && currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
-        super.request(numMessages);
-        return;
-      }
-      if (config.getObservabilityMode() 
-          || currentProcessingMode.getResponseBodyMode() != ProcessingMode.BodySendMode.GRPC) {
-        super.request(numMessages);
-        return;
-      }
       synchronized (streamLock) {
-        pendingRequests.addAndGet(numMessages);
-        downstreamRequestsPending += numMessages;
-        if (isSidecarReady()) {
-          drainPendingMutatedResponseBodies();
-          drainPendingRequests();
+        boolean sendResponseBodiesToExtProc = config.getObservabilityMode()
+            || currentProcessingMode.getResponseBodyMode() == ProcessingMode.BodySendMode.GRPC;
+
+        if (!sendResponseBodiesToExtProc) {
+          // We do not send response bodies to ext_proc server at all. Bypassed.
+          super.request(numMessages);
+          return;
+        }
+
+        // We send response bodies to ext_proc server (either in normal GRPC mode or observability mode).
+        // Gated by ext_proc server readiness.
+        boolean normalFlowControl = !config.getObservabilityMode(); // i.e. normal GRPC response body mode
+
+        if (normalFlowControl) {
+          pendingRequests.addAndGet(numMessages);
+          downstreamRequestsPending += numMessages;
+          if (isExtProcReady()) {
+            drainPendingMutatedResponseBodies();
+            drainPendingRequests();
+          }
+        } else {
+          // Observability mode: gate on readiness but pull all at once
+          if (isExtProcReady()) {
+            super.request(numMessages);
+          } else {
+            pendingRequests.addAndGet(numMessages);
+          }
         }
       }
     }
@@ -1443,7 +1449,7 @@ public void onMessage(InputStream message) {
 
     void drainSavedMessages() {
       synchronized (dataPlaneClientCall.getStreamLock()) {
-        while (dataPlaneClientCall.isSidecarReady()
+        while (dataPlaneClientCall.isExtProcReady()
             && dataPlaneClientCall.upstreamToSidestreamWindow > 0
             && !savedMessages.isEmpty()) {
           InputStream msg = savedMessages.poll();

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

@@ -230,12 +230,6 @@ protected ClientCall<ReqT, RespT> delegate() {
           @Override
           public void start(Listener<RespT> listener, Metadata headers) {
             Executor callExecutor = callOptions.getExecutor();
-            if (callExecutor == null) { // This should never happen in practice because
-              // ManagedChannelImpl.ConfigSelectingClientCall always provides CallOptions with
-              // a callExecutor.
-              // TODO(https://github.com/grpc/grpc-java/issues/7868)
-              callExecutor = MoreExecutors.directExecutor();
-            }
 
             Long delayNanos;
             Status abortStatus = null;