AIP 72: Handling "deferrable" tasks in execution_api and task SDK

[amoghrajesh]

closes: #44137

This PR is trying to port the "deferral" logic from airflow 2 to the airflow 3 (execution api + task sdk)

Summary of changes:

Server side changes (execution api):

1. Handling TIDeferredStatePayload in ti_update_state -> covered by unit test: test_ti_update_state_to_deferred
   a. Didn't piggy back on ti.defer_task() as it extracts the trigger out of TaskDeferred exception. It is much more expensive to send across multiple models like TaskInstance, TaskDeferred, Trigger instead of just the required minimal properties
   b. returning and not proceeding with query execution as we already do it above https://github.com/apache/airflow/pull/44241/files#diff-d44a72566870079ee943e24bac2af74fb84c426c54d210561a251549a7078ed7L129
2. Defining a datamodel for TIDeferredStatePayload and adding it to the discriminator: ti_state_discriminator

Client side changes (task sdk):

HTTP client:

Added a new function defer that sends a patch request to the task-instances/{id}/state execution api with payload: PatchTIToDeferred

Comms:

Defining a new data model to send a request to patch ti as "deferred" from task runner to supervisor: PatchTIToDeferred (Added to ToSupervisor)

Supervisor:

1. Adding a new class property as _final_state to support @property final_state which is final state of a TI
   a. Added a setter to set values for this final state for cases like deferred so that the finish is not called for tasks those aren't in terminal stage: https://github.com/apache/airflow/pull/44241/files#diff-c2651fdee1a25e091e2a9d4f937f8032ca3d289d0de76f38ed88aee5df0f880dL392-L394
2. Deferred tasks first enter the "wait" in supervisor and then mark themselves as deferred by setting up a trigger. So skipping finish() when final_state is not TerminalTIState
3. Extended handle_requests to receive requests from task runner and forwarding the message to http client to call defer

TaskRunner:

Task runner executes: ti.task.execute and raises TaskDeferred for deferral. This sends a request to supervisor using SUPERVISOR_COMMS

How was this tested?

1. The end to end flow isn't available as of now, but I have tried to cover up the ground using unit tests
2. New case under test_handle_requests covers the supervisor + client side of things along with a mock of the message from task runner
3. test_ti_update_state_to_deferred covers the scenario for execution API
4. Added an integration test for task runner component:

• It creates a DAG that raises the deferral exception:
  
• test_run_deferred_basic tests if the DAG raised a task exception and sent the right message across the SUPERVISOR_COMMS or not.

Additional sanity tests

Testing the "task runner" entries and exits (client side)

1. Wrote a async operator DAG:
   

2. Breakpoint inside task_runner.py#run

3. Validated the exception entrypoint with all the variables to be set as required by running the test: test_run_basic
   

Testing the DB state in the server side (execution API)

1. Used the test: test_ti_update_state_to_deferred

2. Debugger inside ti_update_state method

3. Validated the TaskDeferred and related logic
   

4. Checked state of the DB on execution
   

---

^ 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.

[ashb]

A good start!

[kaxil]

Approving pre-emptively as the code looks good but the tests needs a bit more work.

[amoghrajesh]

@kaxil I wrote a new integration test that tests out the deferred exception functionality in task runner as well as tests the message sent across to the SUPERVISOR_COMMS through the send_request function

In this commit: dc39226

[kaxil]

Squashed commits into 1 and resolved conflicts.

Doing one final pass now