improvement: Stop execution early when min_successful reached.

[wangyb-A]

Description

When using MapConfig with CompletionConfig.min_successful, the map operation processes all items instead of stopping execution once the minimum successful threshold is reached.

Expected Behavior

Execution should stop as soon as min_successful items complete successfully, without processing remaining items.

Actual Behavior

All items are processed, ignoring the min_successful threshold.

Reproduce & Examples

Merge the tests cases into your dev branch and run the integration tests. (test cases: aws/aws-durable-execution-sdk-python-testing#146)

[PartiallyUntyped]

Right, so we did implement this here:

8367bcc

However, we can't guarantee that we will actually cancel the futures before they finish execution.

When a future runs, it invokes:

aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py

Lines 293 to 311 in e848c76

	self.counters.complete_task()
	except TimedSuspendExecution as tse:
	exe_state.suspend_with_timeout(tse.scheduled_timestamp)
	scheduler.schedule_resume(exe_state, tse.scheduled_timestamp)
	except SuspendExecution:
	exe_state.suspend()
	# For indefinite suspend, don't schedule resume
	except Exception as e: # noqa: BLE001
	exe_state.fail(e)
	self.counters.fail_task()
	# Check if execution should complete or suspend
	if self.counters.should_complete():
	self._completion_event.set()
	else:
	suspend_result = self.should_execution_suspend()
	if suspend_result.should_suspend:
	self._suspend_exception = suspend_result.exception
	self._completion_event.set()

Which then increments the counters, and when we can finish we call:

aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py

Lines 304 to 307 in e848c76

	# Check if execution should complete or suspend
	if self.counters.should_complete():
	self._completion_event.set()
	else:

The problem is that we can't guarantee that from the time the completion event is set, to the time the futures are cancelled that the executor will not run the rest of the child tasks.

aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/concurrency/executor.py

Lines 226 to 233 in e848c76

	# Wait for completion
	self._completion_event.wait()
	# Cancel remaining futures so
	# that we don't wait for them to join.
	for future in futures:
	future.cancel()