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 13b2969c783 Fix one bad callback request crashing the Dag processor
and dropping the rest (#70990)
13b2969c783 is described below
commit 13b2969c783e2e30ee2290ac9902fc8858b8f12e
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Fri Aug 14 09:51:22 2026 +0100
Fix one bad callback request crashing the Dag processor and dropping the
rest (#70990)
* Fix one bad callback request crashing the Dag processor and dropping the
rest
A callback request that hits the documented race condition (the Dag or
task was removed between scheduling the callback and parsing the file)
raises out of _execute_callbacks, killing the parse subprocess. The
remaining requests in the batch were already popped from the manager's
queue, so on_failure_callback, on_retry_callback, and failure-email
requests behind the bad one are lost permanently. Isolate each request
so a failure is logged and the rest still execute, and route the
last_ti task lookup through _get_dag_with_task so a removed task
surfaces as the race-condition error instead of a raw TaskNotFound.
* Serialize callback request once per loop iteration
Avoids a second to_json() in the exception handler, where a
serialization failure would mask the original error and abort the
batch the handler exists to protect.
* Address review: keep run conf out of logs and deliver Dag callbacks when
only the task is gone
The last_ti task lookup only enriches the callback context, so a task
removed since the run should downgrade to the minimal context (as
produce_dag_callback already does for an unrepresentable last_ti)
instead of costing the user the callback. The failure log leaked
user-supplied dag_run.conf at ERROR via the serialized request and
carried no structured identifiers; it now excludes context_from_server
and binds dag_id/run_id/ti_id so log filters can match the line.
* fixup! Address review: keep run conf out of logs and deliver Dag
callbacks when only the task is gone
---
.../src/airflow/dag_processing/processor.py | 87 +++++++++++-------
.../tests/unit/dag_processing/test_processor.py | 102 +++++++++++++++++++++
2 files changed, 155 insertions(+), 34 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index ca7539131f6..a44f9f37b9e 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -329,23 +329,33 @@ def _execute_callbacks(
) -> None:
for request in callback_requests:
if isinstance(request, (TaskCallbackRequest, EmailRequest)):
- log.debug(
- "Processing Callback Request",
- request=request.to_json(),
- ti_id=str(request.ti.id),
- )
+ log_extra = {
+ "dag_id": request.ti.dag_id,
+ "run_id": request.ti.run_id,
+ "ti_id": str(request.ti.id),
+ }
else:
- log.debug("Processing Callback Request", request=request.to_json())
- with BundleVersionLock(
- bundle_name=request.bundle_name,
- bundle_version=request.bundle_version,
- ):
- if isinstance(request, TaskCallbackRequest):
- _execute_task_callbacks(dagbag, request, log)
- elif isinstance(request, DagCallbackRequest):
- _execute_dag_callbacks(dagbag, request, log)
- elif isinstance(request, EmailRequest):
- _execute_email_callbacks(dagbag, request, log)
+ log_extra = {"dag_id": request.dag_id, "run_id": request.run_id}
+ # context_from_server can carry user-supplied run conf, and the masker
cannot
+ # redact inside an already-serialized string, so keep it out of log
payloads.
+ request_json = request.to_json(exclude={"context_from_server"})
+ log.debug("Processing Callback Request", request=request_json,
**log_extra)
+ # A failed request (e.g. the Dag or task was removed since the callback
+ # was scheduled) must not abort the remaining requests in this batch --
+ # they were already popped from the manager's queue and would be lost.
+ try:
+ with BundleVersionLock(
+ bundle_name=request.bundle_name,
+ bundle_version=request.bundle_version,
+ ):
+ if isinstance(request, TaskCallbackRequest):
+ _execute_task_callbacks(dagbag, request, log)
+ elif isinstance(request, DagCallbackRequest):
+ _execute_dag_callbacks(dagbag, request, log)
+ elif isinstance(request, EmailRequest):
+ _execute_email_callbacks(dagbag, request, log)
+ except Exception:
+ log.exception("Failed to execute callback request",
request=request_json, **log_extra)
def _execute_dag_callbacks(dagbag: DagBag, request: DagCallbackRequest, log:
FilteringBoundLogger) -> None:
@@ -360,25 +370,34 @@ def _execute_dag_callbacks(dagbag: DagBag, request:
DagCallbackRequest, log: Fil
callbacks = callbacks if isinstance(callbacks, list) else [callbacks]
ctx_from_server = request.context_from_server
+ context: Context = {
+ "dag": dag,
+ "run_id": request.run_id,
+ "reason": request.msg,
+ }
if ctx_from_server is not None and ctx_from_server.last_ti is not None:
- task = dag.get_task(ctx_from_server.last_ti.task_id)
-
- runtime_ti = RuntimeTaskInstance.model_construct(
- **ctx_from_server.last_ti.model_dump(exclude_unset=True),
- task=task,
- _ti_context_from_server=TIRunContext.model_construct(
- dag_run=ctx_from_server.dag_run,
- max_tries=task.retries,
- ),
- )
- context = runtime_ti.get_template_context()
- context["reason"] = request.msg
- else:
- context: Context = { # type: ignore[no-redef]
- "dag": dag,
- "run_id": request.run_id,
- "reason": request.msg,
- }
+ try:
+ task = dag.get_task(ctx_from_server.last_ti.task_id)
+ except TaskNotFound:
+ # The task only enriches the callback context; a task removed
since the
+ # run must not cost the user the callback itself
(produce_dag_callback
+ # makes the same call for an unrepresentable last_ti).
+ log.warning(
+ "Task from callback context no longer exists in the Dag;
running callback with minimal context",
+ dag_id=request.dag_id,
+ task_id=ctx_from_server.last_ti.task_id,
+ )
+ else:
+ runtime_ti = RuntimeTaskInstance.model_construct(
+ **ctx_from_server.last_ti.model_dump(exclude_unset=True),
+ task=task,
+ _ti_context_from_server=TIRunContext.model_construct(
+ dag_run=ctx_from_server.dag_run,
+ max_tries=task.retries,
+ ),
+ )
+ context = runtime_ti.get_template_context()
+ context["reason"] = request.msg
for callback in callbacks:
log.info(
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index 776dec298f0..2ced95916f7 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -814,6 +814,42 @@ class TestExecuteCallbacks:
mock_lock.return_value.__exit__.assert_called_once()
mock_execute.assert_called_once_with(dagbag, callbacks[0], log)
+ def test_execute_callbacks_continues_after_failed_request(self,
spy_agency):
+ """A request for a removed Dag must not abort the remaining requests
in the batch."""
+ called = False
+
+ def on_failure(context):
+ nonlocal called
+ called = True
+
+ dag = DAG(dag_id="a", on_failure_callback=on_failure)
+
+ def fake_collect_dags(self, *args, **kwargs):
+ self.dags[dag.dag_id] = dag
+
+ spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags,
owner=DagBag)
+ dagbag = DagBag()
+ dagbag.collect_dags()
+
+ def make_request(dag_id):
+ return DagCallbackRequest(
+ filepath="test.py",
+ dag_id=dag_id,
+ run_id="test_run",
+ bundle_name="testing",
+ bundle_version=None,
+ is_failure_callback=True,
+ msg="Message",
+ )
+
+ log = MagicMock(spec=FilteringBoundLogger)
+ _execute_callbacks(dagbag, [make_request("removed_dag"),
make_request("a")], log)
+
+ assert called is True
+ log.exception.assert_called_once()
+ assert log.exception.call_args.kwargs["dag_id"] == "removed_dag"
+ assert "context_from_server" not in
log.exception.call_args.kwargs["request"]
+
class TestExecuteDagCallbacks:
"""Test the _execute_dag_callbacks function with context_from_server"""
@@ -892,6 +928,72 @@ class TestExecuteDagCallbacks:
assert "ts" in context_received
assert "params" in context_received
+ def
test_execute_dag_callbacks_missing_last_ti_task_falls_back_to_minimal_context(self,
spy_agency):
+ """A removed last_ti task must not drop the callback; it runs with the
minimal context."""
+ called = False
+ context_received = None
+
+ def on_failure(context):
+ nonlocal called, context_received
+ called = True
+ context_received = context
+
+ with DAG(dag_id="test_dag", on_failure_callback=on_failure) as dag:
+ BaseOperator(task_id="test_task")
+
+ def fake_collect_dags(self, *args, **kwargs):
+ self.dags[dag.dag_id] = dag
+
+ spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags,
owner=DagBag)
+ dagbag = DagBag()
+ dagbag.collect_dags()
+
+ current_time = timezone.utcnow()
+ dag_run_data = DRDataModel(
+ dag_id="test_dag",
+ run_id="test_run",
+ logical_date=current_time,
+ data_interval_start=current_time,
+ data_interval_end=current_time,
+ run_after=current_time,
+ start_date=current_time,
+ end_date=None,
+ run_type="manual",
+ state="running",
+ consumed_asset_events=[],
+ partition_key=None,
+ )
+ ti_data = TIDataModel(
+ id=uuid.uuid4(),
+ dag_id="test_dag",
+ task_id="removed_task",
+ run_id="test_run",
+ map_index=-1,
+ try_number=1,
+ dag_version_id=uuid.uuid4(),
+ )
+
+ request = DagCallbackRequest(
+ filepath="test.py",
+ dag_id="test_dag",
+ run_id="test_run",
+ bundle_name="testing",
+ bundle_version=None,
+ context_from_server=DagRunContext(dag_run=dag_run_data,
last_ti=ti_data),
+ is_failure_callback=True,
+ msg="Test failure message",
+ )
+
+ _execute_dag_callbacks(dagbag, request, structlog.get_logger())
+
+ assert called is True
+ assert context_received is not None
+ assert context_received["dag"] is dag
+ assert context_received["run_id"] == "test_run"
+ assert context_received["reason"] == "Test failure message"
+ # The rich template context requires the task; the fallback has none
of it
+ assert "ts" not in context_received
+
def test_execute_dag_callbacks_without_context_from_server(self,
spy_agency):
"""Test _execute_dag_callbacks falls back to simple context when
context_from_server is None"""
called = False
