(beam) branch master updated: [Dataflow Streaming] Avoid LinkedBlockingQueue.removeIf data race in BoundedQueueExecutor (#40812)

[damccorm]

This is an automated email from the ASF dual-hosted git repository.

damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new bd682df6697 [Dataflow Streaming] Avoid LinkedBlockingQueue.removeIf 
data race in BoundedQueueExecutor (#40812)
bd682df6697 is described below

commit bd682df6697c701155775a7b5f3e6eae3f1c01c6
Author: Danny McCormick <dannymccorm...@google.com>
AuthorDate: Wed Oct 28 17:58:41 2026 +0000

    [Dataflow Streaming] Avoid LinkedBlockingQueue.removeIf data race in 
BoundedQueueExecutor (#40812)
---
 .../dataflow/worker/util/BoundedQueueExecutor.java    | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git 
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
 
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
index 1d10d074b11..819f7d37b6f 100644
--- 
a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
+++ 
b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java
@@ -23,6 +23,7 @@ import static 
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Pr
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -536,7 +537,7 @@ public class BoundedQueueExecutor {
    * (e.g. retries or finalizations). {@code elementsOutstanding} counts 
non-shell work items, so
    * comparing {@code workQueue.size()} against {@code elementsOutstanding +
    * maximumElementsOutstanding} means there is at least {@code 
maximumElementsOutstanding} closed
-   * stale shells before running an O(n) {@code removeIf}.
+   * stale shells before running an O(n) purge scan.
    */
   private void onWorkStolen() {
     if (workQueue.size() <= (long) elementsOutstanding.get() + 
maximumElementsOutstanding
@@ -545,7 +546,21 @@ public class BoundedQueueExecutor {
     }
     try {
       if (!executor.isShutdown()) {
-        workQueue.removeIf(BoundedQueueExecutor::isStaleShell);
+        // Avoid LinkedBlockingQueue#removeIf, whose bulkRemove implementation 
reads the
+        // non-volatile head and Node.item fields without holding takeLock 
(causing TSAN data races
+        // with concurrent worker dequeues). LinkedBlockingQueue's iterator 
acquires fullyLock() on
+        // all queue/node accesses while still unlinking in amortized O(1) 
time per element.
+        boolean removed = false;
+        Iterator<Runnable> it = workQueue.iterator();
+        while (it.hasNext()) {
+          if (isStaleShell(it.next())) {
+            it.remove();
+            removed = true;
+          }
+        }
+        if (removed && executor.isShutdown()) {
+          executor.purge();
+        }
       }
     } finally {
       purgingStaleShells.set(false);