From 6adb60cd2b92ab987f96c35433551b0fde64e726 Mon Sep 17 00:00:00 2001 From: Rahul Vats Date: Wed, 15 Jul 2026 18:03:29 +0530 Subject: [PATCH] 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. Signed-off-by: Rahul Vats --- .../providers/celery/cli/celery_command.py | 18 +++++++++---- .../providers/celery/version_compat.py | 9 ++++++- .../unit/celery/cli/test_celery_command.py | 26 +++++++++++++++++++ 3 files changed, 47 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 6e16888a142c7..987233241e105 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.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 83dd3980f084c..8841c057beda9 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 1b8b3a07d4c5f..b189f0f9078ae 100644 --- a/providers/celery/tests/unit/celery/cli/test_celery_command.py +++ b/providers/celery/tests/unit/celery/cli/test_celery_command.py @@ -38,6 +38,7 @@ from tests_common.test_utils.config import conf_vars from tests_common.test_utils.version_compat import ( AIRFLOW_V_3_0_PLUS, + AIRFLOW_V_3_1_PLUS, AIRFLOW_V_3_2_PLUS, AIRFLOW_V_3_3_PLUS, ) @@ -433,6 +434,9 @@ def setup_class(cls): importlib.reload(cli_parser) cls.parser = cli_parser.get_parser() + @pytest.mark.skipif( + not AIRFLOW_V_3_1_PLUS, reason="json_output only passed to configure_logging on Airflow 3.1+" + ) @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") @@ -445,6 +449,9 @@ def test_json_logs_defaults_to_false( _, kwargs = mock_configure_logging.call_args assert kwargs.get("json_output") is False + @pytest.mark.skipif( + not AIRFLOW_V_3_1_PLUS, reason="json_output only passed to configure_logging on Airflow 3.1+" + ) @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") @@ -458,6 +465,9 @@ def test_json_logs_uses_global_logging_config( _, kwargs = mock_configure_logging.call_args assert kwargs.get("json_output") is True + @pytest.mark.skipif( + not AIRFLOW_V_3_1_PLUS, reason="json_output only passed to configure_logging on Airflow 3.1+" + ) @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") @@ -471,6 +481,22 @@ def test_celery_json_logs_overrides_global( _, 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")