Add capability to exit tasks from trigger

[dstandish]

Description

We should be able to exit the "task" from the trigger.

We can already log from triggers so that's one hurdle out of the way.

The remaining hurdle I think is xcom, given that triggers run in asyncio.

One solution that comes to mind is add a queue listener running in a thread that processes the “messages” and handles submitting to xcom.

Another would be use sqlalchemy's asyncio support.

Sqlalchemy asyncio support would not help us though with custom xcom backend -- so maybe we either do just listener or we add an "optimized" path when not using custom xcom backend.

One implementation question is... would you just make it so that the trigger must call Trigger.xcom_push manually (which sends to this queue listener) or would you also want to make it possible, perhaps in an analogy with execute , possible to send the xcom payload(s) in the trigger event, or perhaps through a dedicated Trigger.success method?

I think that I’d probably prefer forcing Trigger writers to do the xcom push manually because … an interface to handle both the default return_value as well as handling multiple key-values isn’t quickly coming to mind.

cc: @jedcunningham @phanikumv @vandonr-amz

Use case/motivation

This would save users money or increase cluster bandwidth because for sensor-like deferrable tasks, no task needs to get spun up just to exit.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[jedcunningham]

One (acceptable imo) impact of this would be no mini scheduler, right?

[dstandish]

Sounds right to me. Yes since the task would not run at exit, then it would not be able to do minischedule. But I suppose the question is, is it worth spinning up a pod just to run the minischeduler? Doubt it??? And, I suppose you could add a "minischeduler listener" too, running in a thread on triggerer but .... is there really a need for this? Isn't the scheduler pretty fast anyway? I don't really know that much about mini-scheduler or know that well how important it is / its reason for being.

[dstandish]

in addition to the happy "exited successful" case, we also need to keep in mind the "cancelled due to timeout" case -- this is another case we should probably not need to go back to the task.

[jedcunningham]

I don't think it's an issue that the mini scheduler isn't doing its thing in these cases, I was more just verbalizing that that is a side effect of this. It makes sense to just rely on the scheduler here.

[dstandish]

I don't think it's an issue that the mini scheduler isn't doing its thing in these cases, I was more just verbalizing that that is a side effect of this. It makes sense to just rely on the scheduler here.

Yeah, understood; and I was just trying to explore the thought of whether it might make sense to "mitigate" in some way :)

[eladkal]

One (acceptable imo) impact of this would be no mini scheduler, right?

We removed mini scheduler in Airflow 3 #43741 so we no longer need to consider this.

I did came across the problem we discuss here. Operators with return statement behave differently when using defer=True. Users expect the value in return be pushed to xcom which works with defer=False but not with defer=True.

I think this issue is more a bug rather than a new feature.