[damccorm]

Summary

Follow-up to #40593.

BoundedQueueExecutor.onWorkStolen() calls workQueue.removeIf(BoundedQueueExecutor::isStaleShell) on LinkedBlockingQueue while worker threads concurrently dequeue tasks via ThreadPoolExecutor.getTask() -> LinkedBlockingQueue.poll().

In OpenJDK's LinkedBlockingQueue.bulkRemove(Predicate) (used by removeIf), the non-volatile head field (Node<E> p = null, ancestor = head;) and Node.item field (nodes[i].item) are read outside fullyLock(), racing with concurrent consumer threads writing head = first and first.item = null in LinkedBlockingQueue.dequeue() under takeLock. This triggers a ThreadSanitizer (TSAN) data race in BoundedQueueExecutorTest.testConcurrentExecuteAndPollWork_processesEachWorkExactlyOnce.

Replace workQueue.removeIf(...) with workQueue.iterator() + Iterator.remove() (matching the idiom used by ThreadPoolExecutor.purge()):

• LinkedBlockingQueue.Itr acquires fullyLock() for all reads/writes of head, Node.next, and Node.item, while still caching ancestor across remove() calls for amortized O(1) unlinking per element (O(n) total scan).
• If executor.shutdown() races concurrently with queue removal, invoke executor.purge() so ThreadPoolExecutor.tryTerminate() is notified in case the queue became empty after shutdown() checked it.

Testing

• ./gradlew :runners:google-cloud-dataflow-java:worker:test --tests org.apache.beam.runners.dataflow.worker.util.BoundedQueueExecutorTest --tests org.apache.beam.runners.dataflow.worker.util.KeyGroupWorkIndexTest

[damccorm]

R: @arunpandianp

[github-actions]

📋 Review Summary

This PR resolves a ThreadSanitizer (TSAN) data race in BoundedQueueExecutor.onWorkStolen() by replacing LinkedBlockingQueue.removeIf(...) with explicit iteration and Iterator.remove(). The change properly ensures all queue and node accesses acquire fullyLock(), while preserving amortized O(1) per-element unlinking and safely notifying ThreadPoolExecutor termination via executor.purge() if shutdown races with removal.

🔍 General Feedback

• The root cause analysis of LinkedBlockingQueue.bulkRemove reading non-volatile head and Node.item fields outside fullyLock() is spot on.
• The replacement iterator loop and post-removal isShutdown() -> executor.purge() check cleanly mirror ThreadPoolExecutor.purge() semantics and maintain thread-safety without performance regressions.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[codecov]

Codecov Report

❌ Patch coverage is 75.00000% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 75.97%. Comparing base (a8d6afd) to head (2c5bdcc).
⚠️ Report is 86 commits behind head on master.

Files with missing lines	Patch %	Lines
...ers/dataflow/worker/util/BoundedQueueExecutor.java	75.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40812      +/-   ##
============================================
- Coverage     80.04%   75.97%   -4.08%     
- Complexity        0     2275    +2275     
============================================
  Files           490      259     -231     
  Lines         82916    14384   -68532     
  Branches          0     1482    +1482     
============================================
- Hits          66373    10928   -55445     
+ Misses        16543     2764   -13779     
- Partials          0      692     +692     

Flag	Coverage Δ
java	75.97% <75.00%> (?)
python	?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.