Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

I have a flow with a section of mapped tasks, one of them failed due to a missing heartbeat - the final flow state is `success` though. why wouldn't this flow's overall state be set to `failed`?

in addition to this, I recently set `PREFECT__CLOUD__HEARTBEAT_MODE` to "thread" in order to (hopefully) prevent these heartbeat issues but they remain - this flow has been running stably for a long time (without heartbeat) issues. how can we be sure heartbeats don't just kill this flow?

it depends on which task is set as a reference task

if your task is say create_flow_run, and this is the last task in your flow, then your flow run will be considered successful already when the child flow run was triggered, not when the child flow run was successful

&gt; how can we be sure heartbeats don't just kill this flow?
could you elaborate?

image.png

this is what the flow looks like - one of the load tasks failed

well, the missing heartbeat is a false positive - the task actually ran successfully. i thought that setting `PREFECT__CLOUD__HEARTBEAT_MODE` to "thread" was meant to help avoid issues like this.

heartbeat for external tasks is a challenging problem and we are aware that the way it is currently implemented is not a perfect UX so we'll try to address it in a better way in Prefect 2.0

it's meant to help mitigate the problem, but switching to threads alone doesn't fully solve the issue

my flow code looks like:
```with DHFlow(FLOW_NAME) as flow:

    anchor_date, do_extract, do_load, do_transform = get_elt_context(
        do_extract=Parameter(name="extract", default=True if IS_PRD else False),
        do_load=Parameter(name="load", default=True),
        do_transform=Parameter(name="transform", default=True),
        anchor_date=Parameter(name="anchor_date", default=None),
    )

    layouts = get_layouts(f"{VENDOR}/layouts", anchor_date=anchor_date)

    e = extract.map(
        layout=layouts,
        anchor=unmapped(anchor_date),
        execute=unmapped(do_extract),
    )
    glue_db_name = manage_glue_catalog(
        base_db_name=f"base-{VENDOR}",
        description=get_l1_glue_description(vendor=VENDOR),
        schema_name=f"base_{VENDOR}",
        execute=unmapped(do_load),
    )
    ld = load.map(
        layout=layouts,
        anchor=unmapped(anchor_date),
        glue_catalog=unmapped(glue_db_name),
        upstream_tasks=[e],
        execute=unmapped(do_load),
    )
    dbt_run = dbt(
        command=f"run --models models/l2_groom/{VENDOR}", upstream_tasks=[ld], execute=unmapped(do_transform)
    )
    dbt_run.name = "transform"```

i'm still confused about how these mapped tasks play together, if one of the `load` tasks fails, why would my `dbt` task run if `load` is an upstream task? is it because i'm not doing `upstream_tasks=[unmapped(ld)]`?

totally forgot i also asked about <https://prefect-community.slack.com/archives/CL09KU1K7/p1650467978229159|this a month ago> - looks like the core issue wasn't really resolved for me

<https://prefect-community.slack.com/archives/CL09KU1K7/p1649771737782739|this discussion> might also be relevant

Hi <@U02A2MYGML0>, checking after holidays: LMK if this is still an open issue

I think this should be:
```upstream_tasks=unmapped([ld])```