You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2025/10/30 22:56:23 UTC
(airflow) branch main updated: Fix connection retrieval in `DagProcessorManager` for bundle initialization (#57459)
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ae2a4fd5044 Fix connection retrieval in `DagProcessorManager` for bundle initialization (#57459)
ae2a4fd5044 is described below
commit ae2a4fd50442b04346d58383f16fd9fa0de00c26
Author: Dheeraj Turaga <dh...@gmail.com>
AuthorDate: Thu Oct 30 17:56:15 2025 -0500
Fix connection retrieval in `DagProcessorManager` for bundle initialization (#57459)
The dag_processor was unable to retrieve connections from the database,
causing GitHook (and other hooks) to fail with:
AirflowNotFoundException: The conn_id `<conn_id>` isn't defined
Root cause: DagProcessorManager was running in FALLBACK context, which only
loads EnvironmentVariablesBackend, not MetastoreBackend. This meant
connections stored in the database were inaccessible.
The `DagProcessorManager` (parent process) needs database access for connection
retrieval during bundle initialization (e.g., `GitDagBundle.__init__` → `GitHook`
needs git credentials). Child `DagFileProcessorProcess` instances run user code
and should remain isolated from direct database access.
This ensures correct secrets backend chains (when no external secrets backend is configured):
- Manager (parent): `EnvironmentVariablesBackend` → `MetastoreBackend` (database access)
- Parser (child): `EnvironmentVariablesBackend`
Note: This is temporary until AIP-92 removes direct DB access from DagProcessorManager.
Long-term, the manager should use the Execution API instead of direct database access.
Affects: DAG bundle processing with GitHook and any other hooks that rely on
database-stored connections during bundle initialization in the manager process.
---
airflow-core/src/airflow/dag_processing/manager.py | 10 ++++++++++
airflow-core/src/airflow/dag_processing/processor.py | 4 ++++
2 files changed, 14 insertions(+)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py
index ff6ff81ab1e..232c0d978ac 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -251,6 +251,16 @@ class DagFileProcessorManager(LoggingMixin):
By processing them in separate processes, we can get parallelism and isolation
from potentially harmful user code.
"""
+ # TODO: Temporary until AIP-92 removes DB access from DagProcessorManager.
+ # The manager needs MetastoreBackend to retrieve connections from the database
+ # during bundle initialization (e.g., GitDagBundle.__init__ → GitHook needs git credentials).
+ # This marks the manager as "server" context so ensure_secrets_backend_loaded() provides
+ # MetastoreBackend instead of falling back to EnvironmentVariablesBackend only.
+ # Child parser processes explicitly override this by setting _AIRFLOW_PROCESS_CONTEXT=client
+ # in _parse_file_entrypoint() to prevent inheriting server privileges.
+ # Related: https://github.com/apache/airflow/pull/57459
+ os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
+
self.register_exit_signals()
self.log.info("Processing files using up to %s processes at a time ", self._parallelism)
diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py
index bf937eac53f..122b2fd0768 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -174,6 +174,10 @@ def _pre_import_airflow_modules(file_path: str, log: FilteringBoundLogger) -> No
def _parse_file_entrypoint():
+ # Mark as client-side (runs user DAG code)
+ # Prevents inheriting server context from parent DagProcessorManager
+ os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "client"
+
import structlog
from airflow.sdk.execution_time import comms, task_runner