Enable ending the task directly from the triggerer without going into the worker.

[sunank200]

Why

For some operators such as DateTimeSensorAsync , TimeDeltaSensorAsync and TimeSensorAsync it would be better if we could terminate the task without needing a worker.

What

Introduce BaseTaskEndEvent to end the task without resuming on the worker.

Triggers have two options: they can either send execution back to the task or end the task instance directly. If the trigger ends the task instance, the method_name should be None. Otherwise, provide the method_name to be used when resuming the execution in the task.

Previously, we only paid attention to failure requests in _execute_task_callbacks. Now, we also execute any given callback type directly. In this scenario, the callback is submitted remotely, so we assume there's no need to change the state; we simply run the callback.

Note: Exiting from the trigger works only when listeners are not integrated for the deferrable operator. To use this feature, unregister the listeners, and set end_from_trigger=True on the given operator.

Introduce end_from_trigger in each Async operator which end the task directly from the triggerer without going into the worker. For example:

with DAG(
    "async_dag",
) as dag:
    async_date_time_sensor = DateTimeSensorAsync(
        task_id="async_date_time_sensor",
        target_time="""{{ macros.datetime.utcnow() + macros.timedelta(minutes=1) }}""",
        end_from_trigger=True,
    )

    async_time_delta = TimeDeltaSensorAsync(
        task_id="async_time_delta",
        delta=datetime.timedelta(minutes=1),
        end_from_trigger=False,
    )

    async_time_sensor = TimeSensorAsync(
        task_id="async_time_sensor",
        timeout=1,
        soft_fail=True,
        target_time=(datetime.datetime.now(tz=dt.timezone.utc) + datetime.timedelta(minutes=1)).time(),
        end_from_trigger=True,
    )

By default end_from_trigger is False.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[jedcunningham]

Does this work with listeners?

[tirkarthi]

Related : #31718

[Lee-W]

158 files change does seem right 🤔 we probably need to rebase again

[sunank200]

158 files change does seem right 🤔 we probably need to rebase again

Yes rebasing again

[Lee-W]

As this is approved by many, I'm planning on merging it within the next hour. Let me know if anyone still want to take a look