Move in-process API Client creation to DagFileProcessorManager

[kaxil]

closes #50097
closes #49887

Previously, each DagFileProcessorProcess created its own InProcessExecutionAPI client instance, leading to unnecessary thread creation and memory build up.

This commit ensures that a single Client backed by InProcessExecutionAPI is created and owned by DagFileProcessorManager, and passed into all DAG file processor subprocesses.

Reproduction

Use following DAG:

from airflow.sdk import Variable, dag, task

@dag
def xample_simplest_dag():

    Variable.get("my_variable", "d")

    @task
    def my_task():
        print("hellooo")

    my_task()


xample_simplest_dag()

Run the dag-processor with following command for quicker reproduction:

AIRFLOW__DAG_PROCESSOR__MIN_FILE_PROCESS_INTERVAL=2 \
AIRFLOW__CORE__LOAD_EXAMPLES=False \
AIRFLOW__DAG_PROCESSOR__DAG_FILE_PROCESSOR_TIMEOUT=2 \
AIRFLOW__DAG_PROCESSOR__BUNDLE_REFRESH_CHECK_INTERVAL=1 \
airflow dag-processor

Check the number of threads grow as the dag is parsed:

Add following diff to check in logs:

diff --git a/task-sdk/src/airflow/sdk/api/client.py b/task-sdk/src/airflow/sdk/api/client.py
index 5f76e8360b..cbee5c2609 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -102,6 +102,11 @@ __all__ = [
 def get_json_error(response: httpx.Response):
     """Raise a ServerResponseError if we can extract error info from the error."""
     err = ServerResponseError.from_response(response)
+    import os;
+    import threading
+    log.exception(f"Total threads at runtime: {len(threading.enumerate())}")
+    for t in threading.enumerate():
+        log.exception(f"  - {t.name}")
     if err:
         log.warning("Server error", detail=err.detail)
         raise err

Another easy way to reproduce, thanks to @tirkarthi in #49887 (comment)

Command:

watch -n 2 "pgrep -f dag-processor | xargs -I{} ls -l /proc/{}/fd/ 2>/dev/null | grep -i socket | wc -l"

---

Before:

After:

In a follow-up, I will add some cleanup to InProcessExecutionAPI itself so our test have no side-effects when used with dag.test.

The Triggerer is the other thing that uses it but because it is a long-running process that doesn't spawn any other processes like DAG File processor, it remains unaffected.

---

^ 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 airflow-core/newsfragments.

[kaxil]

Doc failures are unrelated

[github-actions]

Backport successfully created: v3-0-test

Status	Branch	Result
✅	v3-0-test