Hey, I am doing something like this:

def my_func() -> pd.DataFrame:
    futures = []
    for x in range(0, 100):
        futures.append(some_task.submit(x))

    results = []
    for f in futures:
        try:
            results.append(f.result(raise_on_failure=False))
        except Exception:
            continue

    return pd.concat([r for r in results if isinstance(r, pd.DataFrame)])

and my code runs fine, but the entire flow fails with something like:

19:18:21.583 | ERROR   | Flow run 'stalwart-lorikeet' - Finished in state Failed('1/228 states failed.')

I thought this should handle the failure?

I think this is working as expected. Your code didn’t raise an exception here because of the

raise_on_failure

argument. The final state of your flow is determined by its return value. If you didn’t return a value, the final state is a failure if one of the tasks or subflows in the flow run run failed, which is probably what happened here. https://docs.prefect.io/2.13.8/concepts/flows/#final-state-determination

Ah I did not know that. That's a bit nuanced but easy, to overcome. Thanks @Lee Mendelowitz

I think the Prefect philosophy is you are in complete control over what you want to consider a flow run “failure”. But if you don’t return a value from your flow, the default behavior Prefect chooses for you is that the flow is a success if all tasks and subflows succeed. That seems like a good default to me. I personally rely on this behavior in a lot of our production flows, so I don’t need to explicitly check task state myself and raise a failure at the end.