Implemented for all handlers except alibaba.

Different handlers may be implemented in slightly different ways depending on their characteristics.

Blob storage handlers work by writing to file and then uploading when task is complete. For these handlers, each trigger writes to its own file and at trigger end the file is uploaded with a suffix that distinguishes it from the task log file.

Blob storage handlers include the following:

• GCS
• S3
• WASB
• Alibaba (not implemented due to missing functionality in hook)

For file-based handlers, we have to do two things to make this work with triggers.

The first is, we don't emit to them synchronously. We pipe the messages through a queue to a QueueListener that runs in a thread and emits them to the potentially-blocking handler.

The second is, we need to create a distinct instance of the task handler for each trigger because each instance corresponds to a specific file (and each trigger needs to write to a distinct file). To accomplish this we add a wrapper handler that manages the individual handlers and routes the messages accordingly.

The other category of handlers is a type of "streaming" handler, where messages are pushed through an API to a remote logging service more or less as they are emitted. This category includes the following handlers:

• Cloudwatch
• Elasticsearch
• Stackdriver

Each of the streaming handlers has slightly different characteristics.

Stackdriver essentially has fully native support for triggers. It doesn't require wrapping or routing messages through queuelistener. This is because for one it already runs in a thread and routes messages through a queue internally so as not to be blocking. Additionally it already attached the necessary labels to the record in each call of emit so updathing it to generate those based on the task instance attached to the LogRecord was a trivial matter.

Cloudwatch could be made "native" like stackdriver but the underlying library (watchtower, a 3rd party community library) doesn't quite have the necessary behavior. So for now it still requires using the wrapper and queuelistener.

Elasticsearch is a bit of a hybrid. By default it is filebased and assumes you have a something like filebeat to ship the files. In that way it behaves the same as blob storage handlers. But it has an optional config where messages go to stdout. It it is possible that it could be modifed such that when run in stdout mode it could be "native" like stackdriver where there's just one instance of the handler but such work is not taken on here.

Try it with sample dag:

with DAG(
    dag_id="example_time_delta_sensor_async",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
):
    wait = TimeDeltaSensorAsync(task_id="wait", delta=datetime.timedelta(seconds=30))

Example: