make plugin_executor mandatory

Author: zhongkechen

src/aws_durable_execution_sdk_python/execution.py

@@ -260,6 +260,7 @@ def __init__(
             durable_execution_arn=self._invocation_input.durable_execution_arn,
             initial_checkpoint_token=self._invocation_input.checkpoint_token,
             operations={},
+            plugin_executor=self._plugin_executor,
             service_client=self._service_client,
             replay_status=ReplayStatus.NEW,
         )
@@ -333,8 +334,8 @@ def execute(self):
             ThreadPoolExecutor(
                 max_workers=2, thread_name_prefix="dex-handler"
             ) as executor,
-            contextlib.closing(self._execution_state),
             contextlib.closing(self._plugin_executor),
+            contextlib.closing(self._execution_state),
         ):
             # execute the plugins
             self._plugin_executor.on_invocation_start(

src/aws_durable_execution_sdk_python/state.py

@@ -241,7 +241,7 @@ def __init__(
         initial_checkpoint_token: str,
         operations: MutableMapping[str, Operation],
         service_client: DurableServiceClient,
-        plugin_executor: PluginExecutor | None = None,
+        plugin_executor: PluginExecutor,
         batcher_config: CheckpointBatcherConfig | None = None,
         replay_status: ReplayStatus = ReplayStatus.NEW,
     ):

tests/e2e/map_with_concurrent_waits_int_test.py

@@ -42,6 +42,7 @@
     OperationUpdate,
     OperationType,
 )
+from aws_durable_execution_sdk_python.plugin import PluginExecutor
 from aws_durable_execution_sdk_python.state import (
     CheckpointBatcherConfig,
     ExecutionState,
@@ -68,6 +69,7 @@ def _make_state(
         operations={},
         service_client=mock_client,
         batcher_config=config,
+        plugin_executor=PluginExecutor([]),
     )
 
 

tests/logger_test.py

@@ -229,6 +229,7 @@ def test_logger_with_log_info():
         initial_checkpoint_token="test_token",  # noqa: S106
         operations={},
         service_client=Mock(),
+        plugin_executor=PluginExecutor([]),
     )
     new_info = LogInfo(execution_state_new, "parent2", "op123", "new_name")
     new_logger = logger.with_log_info(new_info)
@@ -379,6 +380,7 @@ def test_logger_replay_no_logging():
         operations={"op1": operation},
         service_client=Mock(),
         replay_status=ReplayStatus.REPLAY,
+        plugin_executor=PluginExecutor([]),
     )
     log_info = LogInfo(replay_execution_state, "parent123", "test_name", 5)
     mock_logger = Mock()
@@ -406,6 +408,7 @@ def test_logger_replay_then_new_logging():
         operations={"op1": operation1, "op2": operation2},
         service_client=Mock(),
         replay_status=ReplayStatus.REPLAY,
+        plugin_executor=PluginExecutor([]),
     )
     log_info = LogInfo(execution_state, "parent123", "test_name", 5)
     mock_logger = Mock()