#7117 Fix: Correctly set 'child data dir/output dir' in branch handler's Abort and Finalize Pull request opened by fg91 Why are the changes needed? from flytekit import conditional, dynamic, task, workflow from flytekit.core.task import Echo echo = Echo(name="echo_str", inputs={"a": str}) @task def test_task() -> str: import time time.sleep(600) # Give time for aborting the execution while this task is running return "foo" @dynamic def dyn() -> str: return test_task() @workflow def wf(a: str = ""): bar = ( conditional("foo-conditional") .if_(a == "") .then(dyn()) .else_() .then(echo(a=a)) ) When aborting the execution while the task called

test_task

within the

@dynamic

within the

conditional

is running, flytepropeller cannot find the

futures.pb

containing the dynamic workflow spec: Failed to propagate Abort for workflow. Error: 0: 0: 0: [system] unable to read futures file, maybe corrupted, caused by: [system] Failed to read futures protobuf file., caused by: path: .../futures.pb: not found Flytepropeller tries to abort the workflows until the system retry limit is exhausted and then stops trying. The resources are not cleaned up. What changes were proposed in this pull request? The reason this is happening is that the

recurseDownstream

method of

branchHandler

modifies the so-called "child data dir" and "child output dir" depending which of the branch of the conditional is taken. This influences where the

futures.pb

file is written. The

Abort

and

Finalize

methods don't modify these dirs, causing the dynamic handler to look for the

futures.pb

under the wrong path when aborting the workflow. To avoid this, we need to modify the "child data dir" and "child output dir" in the

Abort

and

Finalize

methods as well. How was this patch tested? • Extended/added unit tests • Tested that the minimal example workflow above passes with the fixed logic Check all the applicable boxes • I updated the documentation accordingly. • All new and existing tests passed. • All commits are signed-off. flyteorg/flyte