This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy 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 b6d67b7fbb1 Fix Dag callbacks dropped when bundle versioning is
disabled (#72930)
b6d67b7fbb1 is described below
commit b6d67b7fbb1952debee089f1cbc71cbc935ffb4b
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Fri Sep 18 10:59:07 2026 +0100
Fix Dag callbacks dropped when bundle versioning is disabled (#72930)
* Fix Dag callbacks dropped when bundle versioning is disabled
A Dag run that is not pinned to a bundle version -- the normal case under
disable_bundle_versioning -- produces a callback request that carries no
bundle version. Callback queueing built a second, uninitialized bundle
instance for it and read a path off it, so for a versioned bundle whose path
is only defined after initialization the callback was queued against an
unresolved location and lost. Nothing surfaced to the user, because a
callback-only run records no import error.
Such a request wants the bundle's current contents, which is what this
processor's own instance already tracks, so serving it from there resolves
the
same path the file scan queues and keeps a fetch and checkout out of the
parsing loop. Nothing else is a valid answer for an unversioned request:
callbacks are claimed only for bundles this processor parses, and a
versioned
instance that is still uninitialized has just failed initialization in the
refresh that precedes callback fetching, which is also where it is retried.
closes: #64891
* Preserve Dag callbacks while bundles are unavailable
An initialization outage must not consume pending callbacks before the
processor can execute them. Callback work needs to survive temporary bundle
failures and remain diagnosable until recovery.
* Apply suggestion from @ephraimbuddy
---
.../src/airflow/dag_processing/bundles/base.py | 4 +-
airflow-core/src/airflow/dag_processing/manager.py | 38 ++-
airflow-core/src/airflow/utils/db_cleanup.py | 9 +-
.../tests/unit/dag_processing/test_manager.py | 300 ++++++++++++++++++++-
4 files changed, 329 insertions(+), 22 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/bundles/base.py
b/airflow-core/src/airflow/dag_processing/bundles/base.py
index 344a3349fec..c5fa9f7bd0e 100644
--- a/airflow-core/src/airflow/dag_processing/bundles/base.py
+++ b/airflow-core/src/airflow/dag_processing/bundles/base.py
@@ -298,6 +298,9 @@ class BaseDagBundle(ABC):
supports_versioning: bool = False
+ is_initialized: bool = False
+ """Set by ``super().initialize()``; overrides must call it last."""
+
_locked: bool = False
def __init__(
@@ -313,7 +316,6 @@ class BaseDagBundle(ABC):
self.version = version
self.version_data = version_data
self.refresh_interval = refresh_interval
- self.is_initialized: bool = False
self.base_dir = get_bundle_base_folder(bundle_name=self.name)
"""Base directory for all bundle files for this bundle."""
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index ee71d86a9a8..b1d891d5375 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -713,12 +713,23 @@ class DagFileProcessorManager(LoggingMixin):
*,
session: Session = NEW_SESSION,
) -> list[CallbackRequest]:
- """Fetch callbacks from database and add them to the internal queue
for execution."""
+ """Claim callbacks for ready bundles, leaving the rest pending."""
self.log.debug("Fetching callbacks from the database.")
callback_queue: list[CallbackRequest] = []
with prohibit_commit(session) as guard:
- bundle_names = [bundle.name for bundle in self._dag_bundles]
+ # Claiming deletes rows, so defer unavailable bundles before
applying the limit.
+ bundle_names = [
+ bundle.name
+ for bundle in self._dag_bundles
+ if not bundle.supports_versioning or bundle.is_initialized
+ ]
+ if unready_bundles := [
+ bundle.name
+ for bundle in self._dag_bundles
+ if bundle.supports_versioning and not bundle.is_initialized
+ ]:
+ self.log.debug("Skipping callback fetch for uninitialized
bundles: %s", unready_bundles)
query: Select[tuple[DbCallbackRequest]] = with_row_locks(
select(DbCallbackRequest)
.where(DbCallbackRequest.bundle_name.in_(bundle_names))
@@ -743,12 +754,22 @@ class DagFileProcessorManager(LoggingMixin):
def prepare_callback_bundle(self, request: CallbackRequest) ->
BaseDagBundle | None:
"""
- Return the bundle to run the callback against, or ``None`` to skip the
callback.
+ Return a usable bundle or ``None`` to skip; override for API-backed
bundles.
- Default implementation looks the bundle up via
:class:`DagBundlesManager` and, for
- versioned requests on bundles that support versioning, calls
``bundle.initialize()``.
- Override to source the bundle from an API.
+ Reuse loaded bundles for unversioned requests; versioning bundles must
be initialized.
"""
+ if request.bundle_version is None:
+ # Reuse the scan path without fetching or checking out per
callback.
+ loaded = next((b for b in self._dag_bundles if b.name ==
request.bundle_name), None)
+ if loaded is None:
+ self.log.error(
+ "Bundle %s is not parsed by this processor, skipping
callback", request.bundle_name
+ )
+ return None
+ if loaded.supports_versioning and not loaded.is_initialized:
+ self.log.error("Bundle %s is not initialized, skipping
callback", request.bundle_name)
+ return None
+ return loaded
try:
bundle = DagBundlesManager().get_bundle(
name=request.bundle_name,
@@ -758,7 +779,7 @@ class DagFileProcessorManager(LoggingMixin):
except ValueError:
self.log.error("Bundle %s no longer configured, skipping
callback", request.bundle_name)
return None
- if bundle.supports_versioning and request.bundle_version:
+ if bundle.supports_versioning:
try:
bundle.initialize()
except Exception:
@@ -1405,9 +1426,8 @@ class DagFileProcessorManager(LoggingMixin):
self._symlink_latest_log_directory()
self._latest_log_symlink_date = datetime.today()
- bundle = next(b for b in self._dag_bundles if b.name ==
dag_file.bundle_name)
relative_path = Path(dag_file.rel_path)
- return os.path.join(self._get_log_dir(), bundle.name,
f"{relative_path}.log")
+ return os.path.join(self._get_log_dir(), dag_file.bundle_name,
f"{relative_path}.log")
def _get_logger_for_dag_file(self, dag_file: DagFileInfo):
log_filename = self._render_log_filename(dag_file)
diff --git a/airflow-core/src/airflow/utils/db_cleanup.py
b/airflow-core/src/airflow/utils/db_cleanup.py
index 53e1971b8e7..a6213d7de9e 100644
--- a/airflow-core/src/airflow/utils/db_cleanup.py
+++ b/airflow-core/src/airflow/utils/db_cleanup.py
@@ -282,12 +282,9 @@ config_list: list[_TableConfig] = [
table_name="callback",
recency_column_name="created_at",
extra_columns=["id", "state"],
- # Purging a callback cascades to its deadline row, so only finished
callbacks are purged;
- # a state this code does not know keeps its rows. An unfired
deadline's callback sits in
- # SCHEDULED, which is neither active nor terminal, until the deadline
is missed; it is
- # purged only once no deadline references it, as deleting a Dag run
cascades away the
- # deadline at the database level and leaves the callback behind.
Dag-processor callbacks
- # carry no state and are deleted as they are dispatched.
+ # Callback deletion cascades to deadlines, so preserve active or
unknown states and
+ # SCHEDULED callbacks still referenced by a deadline. Stateless
Dag-processor
+ # callbacks are eligible even while pending, once older than the
cleanup cutoff.
extra_filters=[
or_(
column("state").in_(sorted(TERMINAL_STATES)),
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 8dd6a8eca98..fbdc9bc9b75 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -39,6 +39,7 @@ from unittest.mock import MagicMock
import msgspec
import pytest
+import structlog
import time_machine
from sqlalchemy import event, func, select
from sqlalchemy.exc import OperationalError
@@ -46,7 +47,7 @@ from uuid6 import uuid7
from airflow._shared.timezones import timezone
from airflow.callbacks.callback_requests import DagCallbackRequest
-from airflow.dag_processing.bundles.base import BaseDagBundle
+from airflow.dag_processing.bundles.base import BaseDagBundle, BundleVersion
from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.dag_processing.collection import update_dag_parsing_results_in_db
from airflow.dag_processing.dagbag import DagBag
@@ -56,7 +57,12 @@ from airflow.dag_processing.manager import (
DagFileProcessorManager,
DagFileStat,
)
-from airflow.dag_processing.processor import DagFileParsingResult,
DagFileProcessorProcess
+from airflow.dag_processing.processor import (
+ DagFileParseRequest,
+ DagFileParsingResult,
+ DagFileProcessorProcess,
+ _parse_file,
+)
from airflow.models import DagModel, DbCallbackRequest
from airflow.models.asset import TaskOutletAssetReference
from airflow.models.dag_version import DagVersion
@@ -277,6 +283,24 @@ SWEEP_FILES = 4
SWEEP_CALLS = 4
+class LateResolvingBundle(BaseDagBundle):
+ supports_versioning = True
+
+ def __init__(self, *, resolved_path: Path, **kwargs):
+ super().__init__(**kwargs)
+ self._resolved_path = resolved_path
+
+ @property
+ def path(self) -> Path:
+ return self._resolved_path if self.is_initialized else
Path("/dev/null")
+
+ def get_current_version(self) -> BundleVersion:
+ return BundleVersion(version="some_commit_hash")
+
+ def refresh(self) -> None:
+ pass
+
+
class TestDagFileProcessorManager:
@pytest.fixture(autouse=True)
def _disable_examples(self):
@@ -2331,6 +2355,140 @@ class TestDagFileProcessorManager:
"other-bundle-b",
}
+ @mock.patch.object(LateResolvingBundle, "initialize", autospec=True)
+ def test_callback_executes_after_bundle_initialization_recovers(self,
mock_initialize, tmp_path):
+ dag_file = tmp_path / "callback_recovery.py"
+ dag_file.write_text(
+ textwrap.dedent(
+ """
+ from pathlib import Path
+ from airflow.sdk import DAG
+
+ def on_success(context):
+
Path(__file__).with_suffix(".callback").write_text(context["run_id"])
+
+ dag = DAG("callback_recovery", schedule=None,
on_success_callback=on_success)
+ """
+ )
+ )
+ bundle = LateResolvingBundle(name="testing", resolved_path=tmp_path)
+ manager = DagFileProcessorManager(max_runs=1)
+ manager._dag_bundles = [bundle]
+ request = DagCallbackRequest(
+ dag_id="callback_recovery",
+ run_id="run1",
+ filepath=dag_file.name,
+ bundle_name=bundle.name,
+ bundle_version=None,
+ is_failure_callback=False,
+ )
+ with create_session() as session:
+ session.add(DagBundleModel(name=bundle.name))
+ session.add(DbCallbackRequest(callback=request, priority_weight=1))
+
+ mock_initialize.side_effect = RuntimeError("Bundle unavailable")
+ known_files: dict[str, set[DagFileInfo]] = {}
+ manager._refresh_dag_bundles(known_files)
+ for _ in range(3):
+ assert manager.fetch_callbacks() == []
+ mock_initialize.assert_called_once_with(bundle)
+ assert not manager._force_refresh_bundles
+ with create_session() as session:
+ [pending] = session.scalars(select(DbCallbackRequest)).all()
+ assert pending.get_callback_request() == request
+
+ mock_initialize.side_effect = BaseDagBundle.initialize
+ manager._bundles_last_refreshed = 0
+ manager._refresh_dag_bundles(known_files)
+ assert bundle.is_initialized
+ claimed = manager.fetch_callbacks()
+ assert claimed == [request]
+ for callback in claimed:
+ manager._add_callback_to_queue(callback)
+
+ [(file_info, callbacks)] = manager._callback_to_execute.items()
+ _parse_file(
+ DagFileParseRequest(
+ file=str(file_info.absolute_path),
+ bundle_path=file_info.bundle_path,
+ bundle_name=file_info.bundle_name,
+ callback_requests=callbacks,
+ ),
+ log=structlog.get_logger(),
+ )
+ assert dag_file.with_suffix(".callback").read_text() == request.run_id
+ assert manager.fetch_callbacks() == []
+ with create_session() as session:
+ assert session.scalars(select(DbCallbackRequest)).all() == []
+
+ @pytest.mark.parametrize("bundle_version", ["", "v1"])
+ def test_fetch_pinned_callbacks_waits_for_bundle_initialization(self,
tmp_path, bundle_version):
+ bundle = LateResolvingBundle(name="testing", resolved_path=tmp_path)
+ manager = DagFileProcessorManager(max_runs=1)
+ manager._dag_bundles = [bundle]
+ request = DagCallbackRequest(
+ dag_id="dag1",
+ run_id="run1",
+ filepath="dag.py",
+ bundle_name=bundle.name,
+ bundle_version=bundle_version,
+ )
+ with create_session() as session:
+ session.add(DbCallbackRequest(callback=request, priority_weight=1))
+
+ assert manager.fetch_callbacks() == []
+ with create_session() as session:
+ [pending] = session.scalars(select(DbCallbackRequest)).all()
+ assert pending.get_callback_request() == request
+
+ bundle.initialize()
+ assert manager.fetch_callbacks() == [request]
+ with create_session() as session:
+ assert session.scalars(select(DbCallbackRequest)).all() == []
+
+ @pytest.mark.parametrize("supports_versioning", [False, True])
+ @conf_vars({("dag_processor", "max_callbacks_per_loop"): "1"})
+ def test_fetch_callbacks_filters_uninitialized_bundles_before_limit(
+ self, tmp_path, supports_versioning, caplog
+ ):
+ unavailable = LateResolvingBundle(name="unavailable",
resolved_path=tmp_path)
+ ready = MagicMock(spec=BaseDagBundle)
+ ready.name = "ready"
+ ready.supports_versioning = supports_versioning
+ ready.is_initialized = supports_versioning
+ manager = DagFileProcessorManager(max_runs=1)
+ manager._dag_bundles = [unavailable, ready]
+ requests = [
+ DagCallbackRequest(
+ dag_id="dag1",
+ run_id="run1",
+ filepath="dag.py",
+ bundle_name=bundle.name,
+ bundle_version=None,
+ )
+ for bundle in (unavailable, ready)
+ ]
+ with create_session() as session:
+ session.add(DbCallbackRequest(callback=requests[0],
priority_weight=100))
+ session.add(DbCallbackRequest(callback=requests[1],
priority_weight=1))
+
+ with caplog.at_level(logging.DEBUG):
+ assert manager.fetch_callbacks() == [requests[1]]
+ diagnostic = "Skipping callback fetch for uninitialized bundles:
['unavailable']"
+ assert {"event": diagnostic, "log_level": "debug"} in caplog
+ with create_session() as session:
+ [pending] = session.scalars(select(DbCallbackRequest)).all()
+ assert pending.get_callback_request() == requests[0]
+
+ unavailable.initialize()
+ caplog.clear()
+ with caplog.at_level(logging.DEBUG):
+ assert manager.fetch_callbacks() == [requests[0]]
+ assert not any(
+ entry["event"].startswith("Skipping callback fetch for
uninitialized bundles:")
+ for entry in caplog.entries
+ )
+
@mock.patch.object(DagFileProcessorManager, "_get_logger_for_dag_file")
def test_callback_queue(self, mock_get_logger,
configure_testing_dag_bundle):
mock_logger = MagicMock()
@@ -2486,9 +2644,104 @@ class TestDagFileProcessorManager:
name="testing", version="some_commit_hash",
version_data=version_data
)
- @mock.patch("airflow.dag_processing.manager.DagBundlesManager")
- def
test_prepare_callback_bundle_skips_initialize_for_unversioned_request(self,
mock_bundle_manager):
+ @pytest.mark.parametrize(
+ ("supports_versioning", "is_initialized"),
+ [
+ pytest.param(True, True, id="versioned-and-initialized"),
+ pytest.param(False, False, id="non-versioning"),
+ ],
+ )
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager",
autospec=True)
+ def
test_prepare_callback_bundle_reuses_loaded_bundle_for_unversioned_request(
+ self, mock_bundle_manager, supports_versioning, is_initialized
+ ):
+ manager = DagFileProcessorManager(max_runs=1)
+ loaded = MagicMock(spec=BaseDagBundle)
+ loaded.name = "testing"
+ loaded.supports_versioning = supports_versioning
+ loaded.is_initialized = is_initialized
+ manager._dag_bundles = [loaded]
+
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version=None,
+ msg=None,
+ )
+
+ assert manager.prepare_callback_bundle(request) is loaded
+ loaded.initialize.assert_not_called()
+ mock_bundle_manager.return_value.get_bundle.assert_not_called()
+
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager",
autospec=True)
+ def test_prepare_callback_bundle_does_not_retry_uninitialized_bundle(self,
mock_bundle_manager):
manager = DagFileProcessorManager(max_runs=1)
+ loaded = MagicMock(spec=BaseDagBundle)
+ loaded.name = "testing"
+ loaded.supports_versioning = True
+ loaded.is_initialized = False
+ manager._dag_bundles = [loaded]
+
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version=None,
+ msg=None,
+ )
+
+ for _ in range(3):
+ assert manager.prepare_callback_bundle(request) is None
+ loaded.initialize.assert_not_called()
+ mock_bundle_manager.return_value.get_bundle.assert_not_called()
+ assert not manager._force_refresh_bundles
+
+ @pytest.mark.parametrize(
+ "loaded_bundle_names",
+ [
+ pytest.param([], id="no-bundle-loaded"),
+ pytest.param(["other"], id="other-bundle-loaded"),
+ ],
+ )
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager",
autospec=True)
+ def
test_prepare_callback_bundle_skips_unversioned_request_for_unparsed_bundle(
+ self, mock_bundle_manager, loaded_bundle_names
+ ):
+ manager = DagFileProcessorManager(max_runs=1)
+ for name in loaded_bundle_names:
+ loaded = MagicMock(spec=BaseDagBundle)
+ loaded.name = name
+ loaded.supports_versioning = True
+ loaded.is_initialized = True
+ manager._dag_bundles.append(loaded)
+
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version=None,
+ msg=None,
+ )
+
+ assert manager.prepare_callback_bundle(request) is None
+ mock_bundle_manager.return_value.get_bundle.assert_not_called()
+
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager",
autospec=True)
+ def test_prepare_callback_bundle_keeps_empty_string_version_pinned(self,
mock_bundle_manager):
+ manager = DagFileProcessorManager(max_runs=1)
+ loaded = MagicMock(spec=BaseDagBundle)
+ loaded.name = "testing"
+ loaded.supports_versioning = True
+ loaded.is_initialized = True
+ manager._dag_bundles = [loaded]
+
bundle = MagicMock(spec=BaseDagBundle)
bundle.supports_versioning = True
mock_bundle_manager.return_value.get_bundle.return_value = bundle
@@ -2499,12 +2752,15 @@ class TestDagFileProcessorManager:
run_id="run1",
is_failure_callback=False,
bundle_name="testing",
- bundle_version=None,
+ bundle_version="",
msg=None,
)
assert manager.prepare_callback_bundle(request) is bundle
- bundle.initialize.assert_not_called()
+ mock_bundle_manager.return_value.get_bundle.assert_called_once_with(
+ name="testing", version="", version_data=None
+ )
+ bundle.initialize.assert_called_once()
@mock.patch("airflow.dag_processing.manager.DagBundlesManager")
def
test_prepare_callback_bundle_skips_initialize_for_non_versioning_bundle(self,
mock_bundle_manager):
@@ -2614,6 +2870,38 @@ class TestDagFileProcessorManager:
bundle.initialize.assert_called_once()
assert not manager._callback_to_execute
+ @mock.patch("airflow.dag_processing.manager.DagBundlesManager",
autospec=True)
+ def test_add_callback_reuses_loaded_bundle_path_for_unversioned_request(
+ self, mock_bundle_manager, tmp_path
+ ):
+ manager = DagFileProcessorManager(max_runs=1)
+ bundle = LateResolvingBundle(name="testing", resolved_path=tmp_path)
+ bundle.initialize()
+ manager._dag_bundles = [bundle]
+
+ request = DagCallbackRequest(
+ filepath="file1.py",
+ dag_id="dag1",
+ run_id="run1",
+ is_failure_callback=False,
+ bundle_name="testing",
+ bundle_version=None,
+ msg=None,
+ )
+
+ manager._add_callback_to_queue(request)
+
+ mock_bundle_manager.return_value.get_bundle.assert_not_called()
+ [(file_info, _)] = manager._callback_to_execute.items()
+ assert file_info.bundle_path == tmp_path
+ assert file_info in manager._file_queue
+
+ def test_render_log_filename_for_file_whose_bundle_is_not_loaded(self,
tmp_path):
+ manager = DagFileProcessorManager(max_runs=1,
base_log_dir=str(tmp_path))
+ dag_file = DagFileInfo(rel_path=Path("file1.py"),
bundle_name="testing", bundle_path=tmp_path)
+
+ assert
manager._render_log_filename(dag_file).endswith("/testing/file1.py.log")
+
@mock.patch("airflow.dag_processing.manager.DagBundlesManager")
def test_add_callback_skips_when_bundle_unconfigured(self,
mock_bundle_manager):
manager = DagFileProcessorManager(max_runs=1)
