This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 10e310e57ab Fix Celery worker crash on Airflow 3.0 with json_logs
(#69919)
10e310e57ab is described below
commit 10e310e57ab6e6375944091cc8d0559badf38918
Author: Rahul Vats <[email protected]>
AuthorDate: Thu Jul 16 20:39:23 2026 +0530
Fix Celery worker crash on Airflow 3.0 with json_logs (#69919)
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 <[email protected]>
---
.../airflow/providers/celery/cli/celery_command.py | 18 ++++++++++-----
.../src/airflow/providers/celery/version_compat.py | 9 +++++++-
.../tests/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 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..b189f0f9078 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 airflow.providers.common.compat.sdk import conf
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 @@ class TestWorkerJsonLogs:
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 @@ class TestWorkerJsonLogs:
_, 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 @@ class TestWorkerJsonLogs:
_, 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 @@ 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")
