All_Done_One_Success Trigger Rule or Map task failure when no mapped data

[tagoad]

I have seen some chatter about All_Done_One_Success Trigger but i didn't find anywhere that it was considered for addition. I feel like with the introduction of mapped tasks this could be very useful. My example use cause:

Get x orders from source(sometimes there two steps before it gets to mapping) > map x tasks to process orders > map x tasks for sending orders

Now the way the way i have the mapping tasks set up is that they always map off the get step and have their trigger rule set as all done. There is a check in the operator to see if the data mapped from the get step actually was processed in the previous step, and it errors out if it didn't. This allows us to easily go back and clear a set of tasks that belong to the same order without clearing the whole task to remap the newly succeeded ones. The problem here is if the get step fails the mapping tasks try to run but they don't have anything to map so the DAG just sits there running with no tasks running.
Now this trigger rule would solve this, because i do want the mapping tasks to wait for upstream to be done, but if there wasn't any successes then there isn't any data to map with so they should just go upstream_failed.

That being said, maybe there's a simpler approach that just has the mapping fail if it didn't find any data to map with. Thoughts?

[tagoad]

I did figure out a way to mimic what i am looking for. This is attached as the on_fail_callback on any tasks before the mapped ones. Only works though because the DAGs don't branch.

def fail_all_downstream(context):
tasks = context['dag_run'].get_task_instances()
for task in tasks:
if task.current_state() == State.NONE:
task.set_state(State.UPSTREAM_FAILED)

[potiuk]

I have hard time understanding what you want to say and do here.
Can you maybe open PR and explain clearly there what you really want to achieve? I

[tagoad]

I wouldn't mind trying to do a PR but I couldn't figure out where the actual logic for the Trigger Rules are defined. The util.trigger_rule basically just outlines the names of the rules, not how they function.

I will try to clarify the issue I see, and how this new rule address it. Context is data integration. Flowing orders from a retailer system to a warehouse system. You don't want an issue with one order to cause the rest of the batch not to process. So say you have 20 orders, and you have to adjust their schema to match the warehouse. The DAG maps out 20 instances of task 1 to convert the 20 orders, now if the mapped task that sends the orders has a rule of All_Done(which i am using currently) it will trigger even if none succeed. With mapped tasks I have noticed if there is no actual data to map it goes into a limbo state, it doesn't queue, doesn't fail, just sits there and the DAG keeps running taking up resources. If that second task is set to All_Success, even one failure grinds the whole system to a halt. The mapped tasks need to be able to wait for everything to be done upstream, but also make sure at least something succeeded so there is data to actually process and it doesn't go into the limbo state.

Now my on_fail_solution address a slightly different issue that would be solved with the same trigger rule. This DAG is dynamically generated so all mapped tasks have the same trigger rule, and the initial step of each DAG is to get the data to process. Now if this task fails, there is no data to process and the first mapping task goes into the limbo mode. Now this one could be addressed with just setting the trigger rule as all success but the following tasks would then run into the issue mentioned above.

My function above only solves the issue of the task that gets the data failing, not the mapped tasks themselves. Basically i know that if any none mapped task fails I want all downstream tasks to be set to upstream_failed, otherwise any downstream mapped tasks would go into the limbo state.

I hope that clarifies, and if you have suggestions on how to get started building a new trigger rule for a pull request i am not opposed to giving it a shot.

[potiuk]

With mapped tasks I have noticed if there is no actual data to map it goes into a limbo state

What do you mean by that - can you show example? I believe that "All done" should trigger the task even if none of the 20 tasks succeeded and it should get an input with empty array of inputs. But maybe there is an error in the implementaiton and it does not behave this way?. I think adding a special rule to trigger the case that should already be handled is wrong. Please explain and show an example.

BTW. Re trigger rules, it just requires a little curiosity and rather than stopping wheer they are defined, you should look where they are used. Any modern IDE can help. It's as easy as looking for where the enum is used. Look a `trigger_rule_dep.py'

[tagoad]

Here's a screenshot:

Get orders is where all the mapped tasks map from and this is what happens when it errors out. This is the only DAG i had running, and i let it sit just in case the scheduler needed some time. The DAG kept running and the map_orders_to_shipments never got queued or failed. If this behavior is a bug i am definitely for addressing it instead of building a new trigger rule. Maybe have an option for what happens if a mapped task doesn't receive data? Failed vs skipped? I could see use cases for both.

I appreciate the pointer, I did check references in Github as i don't usually interact with the base airflow since I use astronomer for hosting. Now that i have it cloned, i do see the reference in trigger_rule_dep.py though and looks pretty straightforward. Thanks