Wire async task state access for AIP-98 async tasks

[amoghrajesh]

async tasks (AIP-98, in 3.2) run in the triggerer event loop but the current task state and asset state accessors are synchronous and calling them blocks the event loop. Currently, all context accessors — variables, connections, XCom, and task state, go through SUPERVISOR_COMMS.send() which is synchronous and blocks the event loop on every call. No async client path exists in the SDK yet.

The server-side async methods (aget/aset/adelete/aclear) on MetastoreStateBackend already exist. The gap is entirely client-side.

One potential approach would be to:

• Add async methods to the relevant operations in task sdk client
• Add async def get/set/delete/clear to TaskStateAccessor (and other accessors) that bypass supervisor comms and hit the execution API directly via async HTTP
• Wire the async context at task startup when running in triggerer mode

[amoghrajesh]

cc @dabla if you have any plans / thoughts here.

[dabla]

async tasks (AIP-98, in 3.2) run in the triggerer event loop but the current task state and asset state accessors are synchronous and calling them blocks the event loop. Currently, all context accessors — variables, connections, XCom, and task state, go through SUPERVISOR_COMMS.send() which is synchronous and blocks the event loop on every call. No async client path exists in the SDK yet.

The server-side async methods (aget/aset/adelete/aclear) on MetastoreStateBackend already exist. The gap is entirely client-side.

One potential approach would be to:

• Add async methods to the relevant operations in task sdk client
• Add async def get/set/delete/clear to TaskStateAccessor (and other accessors) that bypass supervisor comms and hit the execution API directly via async HTTP
• Wire the async context at task startup when running in triggerer mode

What do you actually mean by async tasks run in the triggerer event loop? To my knowledge async tasks run on the worker event loop, not the triggerer but maybe I’m missing something? Good point regarding the TaskStateAccessor, will have a look.