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 <[email protected]> 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) <[email protected]> Signed-off-by: Rahul Vats <[email protected]> --- .../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")
