When I execute the following flow it completes successfully even if there was an exception raised

foo = FilterTask(filter_func=lambda x: isinstance(x, int))


@task
def m1(x):
    if x == 5:
        raise ValueError()
    return x


with Flow('foo') as f:
    l1 = m1.map([1, 2, 3, 4, 5])
    foo(l1)

My understanding of the filter task is to pass the mapped results to the downstream tasks. The flow run will be considered a success because foo filters out the failed tasks - does it make sense?

also, to change the behavior of when you would like the flow run to be considered a success, you can use:

flow.set_reference_tasks([your_important_task])

this way, regardless of the end result of the filterTask, you can customize when the flow run is considered Successful

The flow run will be considered a success because foo filters out the failed tasks - does it make sense?

that was my question, it seems like in addition to the

filter_func

the task will also filter out exceptions

the reference task doesn't work for me because i need to know whether any task in any part of the flow fails

that was my question, it seems like in addition to the

filter_func

the task will also filter out exceptions

yup! you're right

the reference task doesn't work for me because i need to know whether any task in any part of the flow fails

Got it. I think state handler seems reasonable to figure that out and get notified. The only downside is: if you have 200 tasks and 20 of them fail, you'll get 20 notifications (which can get annoying), not a single one. If that's OK, then a state handler would be the best option