Hey everyone I have a weird situation that I like to hear what you think I have a parent flow that triggers child flow when I get a failed status in the parent flow with error that the new flow didn't get the params But the new flow is actually triggered with the right params What I can tell on the child flow is that he waits for a minute to scale-up it's resources before starting

@Noam polak I assume you’re using a state handler? Can you share how you defined it? This may occur when you trigger a flow that has required parameters, but you didn’t pass them. The easiest way to solve it would be to assign sensible default values to your parameter tasks. What do you mean by scaling up resources before a flow run?

I'm not using any state handler on the parent flow but only in the child flow The wired thing is child flow is triggered with the right parameters but the parent gets a failed task status

can you share both the child and parent flow? otherwise, it’s hard to give any recommendations

I mean that the pod I am using the the child flow is pretty large so it takes some time before it really starts

It looks like you’re calling the start_flow_run task twice in the child flow and this causes the error that you see: #1: Here you’re calling it without parameters

flow.set_dependencies(start_flow_run, [internal_input_data_updated, exists, run_config])

#2: Here you’re calling it correctly

start_flow_run(
        upstream_tasks=[company_ex_id, exists, internal_input_data_updated],
        parameters={
            "input_data": input_data,
            "internal_input_data": internal_input_data_updated,
            "request_id": request_id,
            "fq_run_id": fq_run_id,
        },
        run_config=run_config,
    )

What would be better and easier is when you instantiate the

StartFlowRun

class before the Flow constructor:

start_flow_run = StartFlowRun(flow_name=FS_FQ_FLOW, project_name="default")

with Flow(FLOW_NAME) as flow:
    internal_input_data_updated = update_internal_input_data(internal_input_data)
    start_flow_run_task = start_flow_run(
        upstream_tasks=[company_ex_id, exists, run_config, internal_input_data_updated],
        parameters={
            "input_data": input_data,
            "internal_input_data": internal_input_data_updated,
            "request_id": request_id,
            "fq_run_id": fq_run_id,
        },
        run_config=run_config,
    )

Alternatively, you can replace StartFlowRun with `create_flow_run`:

from prefect.tasks.prefect import create_flow_run

with Flow(FLOW_NAME) as flow:
    internal_input_data_updated = update_internal_input_data(internal_input_data)
    start_flow_run_task = create_flow_run(flow_name=FS_FQ_FLOW, project_name="default",
        upstream_tasks=[company_ex_id, exists, run_config, internal_input_data_updated],
        parameters={
            "input_data": input_data,
            "internal_input_data": internal_input_data_updated,
            "request_id": request_id,
            "fq_run_id": fq_run_id,
        },
        run_config=run_config,
    )

Hey It helped Thanks a lot