Fix missing coordinated synchronizations between threads.

Author: kannanjgithub

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

@@ -361,10 +361,12 @@ protected ExtProcClientCall(ClientCall<InputStream, InputStream> delegate,
       }
 
       private void sendToDataPlane(Runnable action) {
-        if (headersSent) {
-          action.run();
-        } else {
-          pendingActions.add(action);
+        synchronized (lock) {
+          if (headersSent) {
+            action.run();
+          } else {
+            pendingActions.add(action);
+          }
         }
       }
 
@@ -400,9 +402,11 @@ public void onNext(io.envoyproxy.envoy.service.ext_proc.v3.ProcessingResponse re
               if (response.getRequestHeaders().hasResponse()) {
                 applyHeaderMutations(requestHeaders, response.getRequestHeaders().getResponse().getHeaderMutation());
               }
-              headersSent = true;
-              delegate().start(wrappedListener, requestHeaders);
-              drainQueue();
+              synchronized (lock) {
+                headersSent = true;
+                delegate().start(wrappedListener, requestHeaders);
+                drainQueue();
+              }
             }
             // 2. Client Message (Request Body)
             else if (response.hasRequestBody()) {
@@ -496,8 +500,10 @@ public void onCompleted() {
         }
 
         if (config.getObservabilityMode()) {
-          headersSent = true;
-          delegate().start(wrappedListener, headers);
+          synchronized (lock) {
+            headersSent = true;
+            delegate().start(wrappedListener, headers);
+          }
         }
       }
 
@@ -632,12 +638,14 @@ private void handleFailOpen(ExtProcListener listener) {
         if (extProcStreamCompleted.compareAndSet(false, true)) {
           // The ext_proc stream is gone. "Fail open" means we proceed with the RPC
           // without any more processing.
-          if (!headersSent) {
-            headersSent = true;
-            delegate().start(listener, requestHeaders);
+          synchronized (lock) {
+            if (!headersSent) {
+              headersSent = true;
+              delegate().start(listener, requestHeaders);
+            }
+            drainQueue();
           }
           listener.unblockAfterStreamComplete();
-          drainQueue();
         }
       }
     }