Update dynamic pipeline default execution mode (#4317)

* Update dynamic pipeline default execution mode

* Docs

Author: schustmi

docs/book/how-to/steps-pipelines/dynamic_pipelines.md

@@ -327,6 +327,10 @@ When you call `.load()` on an artifact in a dynamic pipeline, it synchronously l
 - Mapping is currently supported only over artifacts produced within the same pipeline run (mapping over raw data or external artifacts is not supported).
 - Chunk size for mapped collection loading defaults to 1 and is not yet configurable.
 
+### Execution mode
+
+Currently only the `STOP_ON_FAILURE` execution mode is supported for dynamic pipelines, and will be used as a default.
+
 ## Best Practices
 
 1. **Use `runtime="isolated"` for parallel steps**: This ensures better resource isolation and prevents interference between concurrent step executions.

src/zenml/orchestrators/base_orchestrator.py

@@ -636,6 +636,10 @@ def _validate_execution_mode(
         Raises:
             ValueError: If the execution mode is not supported.
         """
+        if snapshot.is_dynamic:
+            # We can't validate execution modes for dynamic pipelines yet
+            return
+
         execution_mode = snapshot.pipeline_configuration.execution_mode
 
         if execution_mode not in self.supported_execution_modes:

src/zenml/pipelines/dynamic/pipeline_definition.py

@@ -26,6 +26,7 @@
 
 from zenml.client import Client
 from zenml.config.source import Source
+from zenml.enums import ExecutionMode
 from zenml.execution.pipeline.utils import (
     should_prevent_pipeline_execution,
 )
@@ -48,18 +49,20 @@ class DynamicPipeline(Pipeline):
 
     def __init__(
         self,
-        *args: Any,
+        *,
         depends_on: Optional[List["BaseStep"]] = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the pipeline.
 
         Args:
-            *args: Pipeline constructor arguments.
             depends_on: The steps that the pipeline depends on.
             **kwargs: Pipeline constructor keyword arguments.
         """
-        super().__init__(*args, **kwargs)
+        # This is the only execution mode that is currently supported for
+        # dynamic pipelines, so we default to it.
+        kwargs.setdefault("execution_mode", ExecutionMode.STOP_ON_FAILURE)
+        super().__init__(**kwargs)
         self._depends_on = depends_on or []
         self._validate_depends_on(self._depends_on)
 

src/zenml/pipelines/pipeline_definition.py

@@ -130,6 +130,7 @@ class Pipeline:
 
     def __init__(
         self,
+        *,
         name: str,
         entrypoint: F,
         enable_cache: Optional[bool] = None,