Hi there. I am using the new

get_task_run_result

and

create_flow_run

tasks described here. I'd like to treat the child flow as a single unit that can be retried or restarted if the flow fails. Currently, these are two separate tasks and I haven't been able to set them up this way. How could I change my flow to support retries/restart when the child flow fails?

Hey @Pedro Machado, I think you need to wrap it like this:

@task(max_retries = ..., retry_delay = ...)
def wrapper():
    create_flow_run(...).run()

But this would create new flow runs. I think the better approach is to apply the retries on that sub flow because you ideally don’t want to rerun tasks that already succeeded.

Thanks, Kevin. In this particular case the child flow would need to be rerun completely so I'll try your suggestion. Thanks!

Actually it’ll be better like this

@task(max_retries = ..., retry_delay = ...)
def wrapper():
    flow_run_id = create_flow_run(...).run()
    return flow_run_id

You also may be able to create a copy of

create_flow_run

and set the retries, e.g.

my_create_flow_run = create_flow_run.copy()
my_create_flow_run.max_retries = 3