Repository: Netflix/metaflow
Issue category: Bug Report

Issue:
BUG: Data store error - AWS batch/step execution
**Environment:**
metaflow version: 2.12.29
Python 3.11 (Docker Image from public.ecr.aws/docker/library/python:3.11)
Running on AWS Batch

**Description:**
Tested with version 2.12.28 and it runs successfully, with this latest version we get:
Data store error: No completed attempts of the task was found for task `MyFlow/sfn-*/_parameters/*-params`.

Maybe worth mentioning that we include a json file into `MyFlow` like:
```

json_config = IncludeFile(
        name="my_config",
        required=True,
        help="The Configuration",
        default=f"./{PARAMS_JSON}",
    )

```



Hints:
I also got this error when running on argo workflows. My flow does not use `IncludeFile` but just usual parameters.

I can also confirm it happens for `2.12.29` but not `2.12.28`
And another confirmation with step on batch.  2.12.29 displays the error, 2.12.28 does not.
I also got this error on Argo Workflows. Same problematic version (`2.12.29`) and the same fix (downgrade to `2.12.28`).
we are triaging
also, for quicker resolution/response, you can always ping us on chat.metaflow.org

Patch:
diff --git a/metaflow/cli.py b/metaflow/cli.py
index 1fc6a14953..a318b84a3e 100644
--- a/metaflow/cli.py
+++ b/metaflow/cli.py
@@ -282,31 +282,21 @@ def dump(obj, input_path, private=None, max_value_size=None, include=None, file=
     else:
         ds_list = list(datastore_set)  # get all tasks
 
-    tasks_processed = False
     for ds in ds_list:
-        if ds is not None:
-            tasks_processed = True
-            echo(
-                "Dumping output of run_id=*{run_id}* "
-                "step=*{step}* task_id=*{task_id}*".format(
-                    run_id=ds.run_id, step=ds.step_name, task_id=ds.task_id
-                ),
-                fg="magenta",
-            )
-
-            if file is None:
-                echo_always(
-                    ds.format(**kwargs),
-                    highlight="green",
-                    highlight_bold=False,
-                    err=False,
-                )
-            else:
-                output[ds.pathspec] = ds.to_dict(**kwargs)
+        echo(
+            "Dumping output of run_id=*{run_id}* "
+            "step=*{step}* task_id=*{task_id}*".format(
+                run_id=ds.run_id, step=ds.step_name, task_id=ds.task_id
+            ),
+            fg="magenta",
+        )
 
-    if not tasks_processed:
-        echo(f"No task(s) found for pathspec {input_path}", fg="red")
-        return
+        if file is None:
+            echo_always(
+                ds.format(**kwargs), highlight="green", highlight_bold=False, err=False
+            )
+        else:
+            output[ds.pathspec] = ds.to_dict(**kwargs)
 
     if file is not None:
         with open(file, "wb") as f:


Test patch: