feat(map): Fix map replay, wait-inside-map, and concurrency race conditions

[ayushiahjolia]

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

#39

Description

• Updated Map implementation to rely on ConcurrencyOperation.
• Added tests for all map usages.

Follow-ups/Remaining Work

• If function returns null output, are we allowing checkpointing that?
• Is it okay to enqueue all items? or we want to start executing item as soon as possible.
• Doc updates in next CR, will remove dead code from Test as well.
• Thread issues during replay in LocalDurableTestRunner.
• If minSuccessful is reached then remaining iterations should return NOT_STARTED or SUCCESSFUL?

Demo/Screenshots

Checklist

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

Testing

Unit Tests

Have unit tests been written for these changes? Updated.

Integration Tests

Have integration tests been written for these changes? Yes

Examples

Has a new example been added for the change? (if applicable) Yes

[zhongkechen]

unit tests failed:

[ERROR] Errors: 
[ERROR]   ParallelOperationTest.branchCreation_multipleBranchesAllCreated:106 » IllegalState Cannot add items to a completed operation

This is due to the race condition introduced in the following change:

-        // Block until operation completes. No-op if the future is already completed.
-        completionFuture.join();
+        // Block until operation completes or execution suspends.
+        // Using runUntilCompleteOrSuspend races completionFuture against executionExceptionFuture,
+        // so when all active threads suspend (e.g., wait inside map branches), the
+        // SuspendExecutionException propagates and this thread is freed — preventing thread leaks
+        // on shared executor pools across invocations.
+        executionManager.runUntilCompleteOrSuspend(completionFuture).join();

Removing this change should fix the tests

[ayushiahjolia]

When all map branches call wait(), every branch thread deregisters and suspendExecution() fires. But the parent thread sitting on completionFuture.join() in BaseConcurrentOperation.get() is stuck forever - no branch will ever complete to trigger onChildContextComplete and finalize the map's completionFuture. runUntilCompleteOrSuspend races completionFuture against executionExceptionFuture, so when suspension fires, the parent thread gets freed via SuspendExecutionException instead of blocking indefinitely.

[wangyb-A]

Wondering the expected behavior of early termination for map, do we skip the rest of the items?

ChildOperation will not send checkpoint if parent operation is completed.

[ayushiahjolia]

yes we skip remaining items with NOT_STARTED status.

[ayushiahjolia]

But if there were some items already started, they will complete.

[zhongkechen]

calling enqueueItem for all items, followed by startPendingItems, seems no difference with calling addItem for all items. Just for calculation of failure percentile?

[wangyb-A]

The calculation of failure numbers can also be calculated for eager start, because when user create Map they already pass the items as a list?

[ayushiahjolia]

addItem calls executeNextItemIfAllowed() immediately, and on replay child execution is synchronous - so early termination (e.g., minSuccessful=1) can mark the operation completed before the remaining items are even registered. The next addItem would then throw IllegalStateException. Enqueue-then-start ensures all items are registered before any execution begins.

[zhongkechen]

As soon as this is done, we need to prevent more children from checkpointing their results so that we can keep the result consistent.

[ayushiahjolia]

ChildContextOperation.checkpointSuccess checks if (parentOperation != null && parentOperation.isOperationCompleted()) before sending any checkpoint. So once handleComplete marks the parent as completed, any in-flight children that finish afterwards will skip their checkpoint and just call markAlreadyCompleted() locally.

[wangyb-A]

We will have some minor refactors and fix in the future.

@zhongkechen Please also take a look 😀

[zhongkechen]

We will have some minor refactors and fix in the future.

@zhongkechen Please also take a look 😀

We still need some major fixes 😅