You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/07/15 12:34:17 UTC
(airflow) 01/01: Fix Celery worker crash on Airflow 3.0 with json_logs
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch fix-celery-json-output-af30
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 23205ae932224b999853c102c2b8016a745f38c9
Author: Rahul Vats <ra...@gmail.com>
AuthorDate: Wed Jul 15 18:03:29 2026 +0530
Fix Celery worker crash on Airflow 3.0 with json_logs
The celery worker CLI unconditionally passed `json_output` to
`airflow.sdk.log.configure_logging` under an `AIRFLOW_V_3_0_PLUS` gate.
That parameter was only added to the Task SDK in 1.1 (Airflow 3.1, the
structlog migration), so on Airflow 3.0.x the call raises
`TypeError: configure_logging() got an unexpected keyword argument
'json_output'` and the worker crashes on startup.
Gate the `json_logs` handling on a new `AIRFLOW_V_3_1_PLUS` constant and
fall back to `configure_logging(output=sys.stdout.buffer)` on 3.0.x, so
the provider stays compatible across the full Airflow 3.x range it
supports.
Co-Authored-By: Claude Opus 4.8 (1M context) <no...@anthropic.com>
Signed-off-by: Rahul Vats <ra...@gmail.com>
---
.../src/airflow/providers/celery/cli/celery_command.py | 18 +++++++++++++-----
.../src/airflow/providers/celery/version_compat.py | 9 ++++++++-
.../tests/unit/celery/cli/test_celery_command.py | 16 ++++++++++++++++
3 files changed, 37 insertions(+), 6 deletions(-)
diff --git a/providers/celery/src/airflow/providers/celery/cli/celery_command.py b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
index 6e16888a142..987233241e1 100644
--- a/providers/celery/src/airflow/providers/celery/cli/celery_command.py
+++ b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
@@ -39,6 +39,7 @@ from airflow.cli.simple_table import AirflowConsole
from airflow.exceptions import AirflowConfigException
from airflow.providers.celery.version_compat import (
AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_1_PLUS,
AIRFLOW_V_3_2_PLUS,
AIRFLOW_V_3_3_PLUS,
)
@@ -263,12 +264,19 @@ def worker(args):
if AIRFLOW_V_3_0_PLUS:
from airflow.sdk.log import configure_logging
- _celery_json = config.get("celery", "json_logs", fallback="")
- if _celery_json and _celery_json.lower() != "none":
- json_output = config.getboolean("celery", "json_logs")
+ if AIRFLOW_V_3_1_PLUS:
+ _celery_json = config.get("celery", "json_logs", fallback="")
+ if _celery_json and _celery_json.lower() != "none":
+ json_output = config.getboolean("celery", "json_logs")
+ else:
+ json_output = config.getboolean("logging", "json_logs", fallback=False)
+ configure_logging(output=sys.stdout.buffer, json_output=json_output)
else:
- json_output = config.getboolean("logging", "json_logs", fallback=False)
- configure_logging(output=sys.stdout.buffer, json_output=json_output)
+ # Airflow 3.0.x ships an SDK whose configure_logging predates the
+ # json_output parameter (added in the Task SDK 1.1 / Airflow 3.1
+ # structlog migration). Passing it there raises TypeError and crashes
+ # the worker, so honor json_logs only from 3.1 onwards.
+ configure_logging(output=sys.stdout.buffer)
else:
# Disable connection pool so that celery worker does not hold an unnecessary db connection
settings.reconfigure_orm(disable_connection_pool=True)
diff --git a/providers/celery/src/airflow/providers/celery/version_compat.py b/providers/celery/src/airflow/providers/celery/version_compat.py
index 83dd3980f08..8841c057bed 100644
--- a/providers/celery/src/airflow/providers/celery/version_compat.py
+++ b/providers/celery/src/airflow/providers/celery/version_compat.py
@@ -27,8 +27,15 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
+AIRFLOW_V_3_1_PLUS = get_base_airflow_version_tuple() >= (3, 1, 0)
AIRFLOW_V_3_1_9_PLUS = get_base_airflow_version_tuple() >= (3, 1, 9)
AIRFLOW_V_3_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 0)
AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
-__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_1_9_PLUS", "AIRFLOW_V_3_2_PLUS", "AIRFLOW_V_3_3_PLUS"]
+__all__ = [
+ "AIRFLOW_V_3_0_PLUS",
+ "AIRFLOW_V_3_1_PLUS",
+ "AIRFLOW_V_3_1_9_PLUS",
+ "AIRFLOW_V_3_2_PLUS",
+ "AIRFLOW_V_3_3_PLUS",
+]
diff --git a/providers/celery/tests/unit/celery/cli/test_celery_command.py b/providers/celery/tests/unit/celery/cli/test_celery_command.py
index 1b8b3a07d4c..46e8bda4637 100644
--- a/providers/celery/tests/unit/celery/cli/test_celery_command.py
+++ b/providers/celery/tests/unit/celery/cli/test_celery_command.py
@@ -471,6 +471,22 @@ class TestWorkerJsonLogs:
_, kwargs = mock_configure_logging.call_args
assert kwargs.get("json_output") is False
+ @mock.patch("airflow.providers.celery.cli.celery_command.AIRFLOW_V_3_1_PLUS", False)
+ @mock.patch("airflow.providers.celery.cli.celery_command.Process")
+ @mock.patch("airflow.providers.celery.executors.celery_executor.app")
+ @mock.patch("airflow.sdk.log.configure_logging")
+ def test_json_output_not_passed_on_airflow_3_0(
+ self, mock_configure_logging, mock_celery_app, mock_popen, mock_pre_exec
+ ):
+ # Airflow 3.0.x's configure_logging has no json_output parameter; passing it
+ # crashes the worker with TypeError. Ensure we omit it below Airflow 3.1.
+ args = self.parser.parse_args(["celery", "worker"])
+ with conf_vars({("logging", "json_logs"): "True"}):
+ celery_command.worker(args)
+ mock_configure_logging.assert_called_once()
+ _, kwargs = mock_configure_logging.call_args
+ assert "json_output" not in kwargs
+
@pytest.mark.backend("mysql", "postgres")
@pytest.mark.usefixtures("conf_stale_bundle_cleanup_disabled")