refactor: use queue-based concurrency coordinator

[zhongkechen]

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Issue Link, if available

N/A

Description

Replace the concurrency operation's rebuilt CompletableFuture.anyOf coordination with a persistent event queue modeled after the Python SDK coordinator.

• Give one coordinator exclusive ownership of running children and completion counters.
• Publish child completion, state-change, and terminal failure events through a blocking queue.
• Preserve FIFO branch admission and bounded concurrency.
• Keep the synchronized active-thread handoff required to avoid lost wakeups during suspension.
• Add regression coverage for queue admission and exceptional completion while the coordinator is waiting.

Demo/Screenshots

N/A - internal coordination refactor with no user-facing UI changes.

Checklist

• I have filled out every section of the PR template
• I have thoroughly tested this change

Testing

Unit Tests

Yes. Added coordinator queue admission and exceptional wakeup tests. Ran the full SDK unit suite: 1,139 tests passed.

Integration Tests

Existing map and parallel integration coverage was run: 169 tests passed, including replay, suspension, failures, nesting modes, early completion, and bounded concurrency.

Examples

Not applicable. This change does not alter the public API or add a customer-facing workflow.

[github-actions]

Codex AI review

[P1] Preserve managed blocking for custom ForkJoinPool executors. BlockingQueue.take() parks the coordinator without notifying a user-configured ForkJoinPool. With a single-worker pool, the root's CompletableFuture.join() allows the coordinator to run, but the coordinator then occupies that worker here while its child remains queued, hanging map/parallel execution. Use ForkJoinPool.managedBlock, an equivalent future-based notification, or an SDK coordination executor, and add a single-parallelism ForkJoinPool regression test.

[zhongkechen]

Thanks for flagging this. I added singleWorkerForkJoinPoolDoesNotStarveCoordinator in ParallelIntegrationTest using a custom ForkJoinPool(1), two parallel branches, and a five-second timeout. It completes successfully, and the full ParallelIntegrationTest suite passes (65 tests).

On Corretto/OpenJDK 17, LinkedBlockingQueue.take() waits through an AQS ConditionObject; its condition node implements ForkJoinPool.ManagedBlocker, and ConditionObject.await() invokes ForkJoinPool.managedBlock. The pool therefore compensates for the blocked coordinator even though this call site does not invoke managedBlock explicitly. Wrapping take() in another managed blocker would be redundant for the supported runtime.

The regression coverage is in commit 12a99c2.

[github-actions]

Codex AI review

One thread-coordination regression remains. Tests use cached thread pools and do not cover managed blocking with a ForkJoinPool-backed custom executor.

Reviewed commit d99cee1e6230a67eb8761b66cced5cd49a16153f. Workflow run