This is an automated email from the ASF dual-hosted git repository.
ferruzzi 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 9ecd53d3a90 Fix Celery workers rejecting deadline callbacks on Airflow
3.2 (#75438)
9ecd53d3a90 is described below
commit 9ecd53d3a90972254839be6ba4b197f958e4fa0b
Author: femii1234 <[email protected]>
AuthorDate: Tue Nov 10 02:22:26 2026 +0000
Fix Celery workers rejecting deadline callbacks on Airflow 3.2 (#75438)
The pre-3.3 fallback handled task workloads but raised ValueError for
ExecuteCallback, so SyncCallback deadline alerts never ran. Dispatch
those callbacks through the Airflow 3.2 helper.
Signed-off-by: Olufemi Anjorin <[email protected]>
---
.../celery/executors/celery_executor_utils.py | 10 ++++++
.../unit/celery/executors/test_celery_executor.py | 42 ++++++++++++++++++++++
2 files changed, 52 insertions(+)
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 8e3cfa4adca..9642ee3b2a0 100644
---
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -314,6 +314,16 @@ def _execute_workload_pre_3_3(input: str) -> None:
log_path=workload.log_path,
subprocess_logs_to_stdout=subprocess_logs_to_stdout,
)
+ elif AIRFLOW_V_3_2_PLUS and isinstance(workload,
workloads.ExecuteCallback):
+ # Airflow 3.3 runs callbacks in BaseExecutor.run_workload().
Airflow 3.2 still
+ # queues deadline alerts as ExecuteCallback and provides
execute_callback_workload.
+ execute_callback_workload = getattr(
+ import_module("airflow.executors.workloads.callback"),
+ "execute_callback_workload",
+ )
+ success, error_msg = execute_callback_workload(workload.callback,
log)
+ if not success:
+ raise RuntimeError(error_msg or "Callback execution failed")
else:
raise ValueError(f"CeleryExecutor does not know how to handle
{type(workload)}")
except Exception as e:
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 41e091012dc..dfbb38eefb4 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -1289,6 +1289,48 @@ def
test_execute_workload_runs_execute_task_before_airflow_3_3(execute_task_work
assert mock_supervise.call_args.kwargs["log_path"] == "test.log"
[email protected](not AIRFLOW_V_3_2_PLUS, reason="ExecuteCallback requires
Airflow 3.2+")
[email protected](
+ ("callback_result", "expected_error"),
+ [
+ pytest.param((True, None), None, id="callback-succeeds"),
+ pytest.param((False, "Callback failed"), "Callback failed",
id="callback-fails"),
+ ],
+)
+def test_execute_workload_runs_execute_callback_before_airflow_3_3(
+ mock_celery_app, monkeypatch, callback_result, expected_error
+):
+ """The pre-3.3 fallback executes ExecuteCallback with the Airflow 3.2
callback helper."""
+ from airflow.executors import workloads
+ from airflow.executors.workloads.callback import CallbackDTO
+
+ workload = workloads.ExecuteCallback(
+ callback=CallbackDTO(
+ id="00000000-0000-0000-0000-000000000003",
+ fetch_method=workloads.CallbackFetchMethod.IMPORT_PATH,
+ data={"path": "tests.callbacks.test_callback", "kwargs": {}},
+ ),
+ dag_rel_path=Path("test_dag.py"),
+ bundle_info=workloads.BundleInfo(name="test-bundle", version=None),
+ token="test-token",
+ log_path="callback.log",
+ )
+ monkeypatch.setattr(celery_executor_utils, "AIRFLOW_V_3_3_PLUS", False)
+ with mock.patch(
+ "airflow.executors.workloads.callback.execute_callback_workload",
+ create=True,
+ return_value=callback_result,
+ ) as mock_execute_callback:
+ execute = celery_executor_utils.execute_workload.__wrapped__
+ if expected_error:
+ with pytest.raises(RuntimeError, match=expected_error):
+ execute(workload.model_dump_json())
+ else:
+ execute(workload.model_dump_json())
+
+ mock_execute_callback.assert_called_once_with(workload.callback,
celery_executor_utils.log)
+
+
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
@pytest.mark.skipif(AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
@pytest.mark.parametrize(
