Fix Celery worker crash on Airflow 3.0 with json_logs

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)

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",
+]

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")