This is an automated email from the ASF dual-hosted git repository.
mobuchowski 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 7abf0487f44 fix: Prevent stats from breaking executor initializer
(#66351)
7abf0487f44 is described below
commit 7abf0487f44807faef1226773f6986058c15a9ee
Author: Kacper Muda <[email protected]>
AuthorDate: Mon May 4 16:27:34 2026 +0200
fix: Prevent stats from breaking executor initializer (#66351)
---
.../providers/openlineage/plugins/listener.py | 32 +++++--
.../unit/openlineage/plugins/test_listener.py | 104 +++++++++++++++++++++
2 files changed, 126 insertions(+), 10 deletions(-)
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
index b33c0fed57e..6ab1944c393 100644
---
a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
+++
b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
@@ -29,12 +29,13 @@ from openlineage.client.serde import Serde
from airflow import settings
from airflow.models import DagRun, TaskInstance
-from airflow.providers.common.compat.sdk import Stats, hookimpl, timeout,
timezone
+from airflow.providers.common.compat.sdk import Stats, conf as airflow_conf,
hookimpl, timeout, timezone
from airflow.providers.openlineage import conf
from airflow.providers.openlineage.extractors import ExtractorManager,
OperatorLineage
from airflow.providers.openlineage.plugins.adapter import OpenLineageAdapter,
RunState
from airflow.providers.openlineage.utils.utils import (
AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_2_PLUS,
get_airflow_dag_run_facet,
get_airflow_debug_facet,
get_airflow_job_facet,
@@ -81,21 +82,32 @@ def _executor_initializer():
parent's ``Stats.initialize(...)`` call from scheduler startup does not
propagate across the
spawn boundary.
"""
+ log = logging.getLogger(__name__)
# This initializer is used only on the scheduler
# We can configure_orm regardless of the Airflow version, as DB access is
always allowed from scheduler.
settings.configure_orm()
+
+ if not AIRFLOW_V_3_2_PLUS: # Initialize stats only for AF 3.2+
+ return
+
+ # Stats-related errors must not block initialization of executor (this
will block DAG event emission).
try:
from airflow.observability.metrics import stats_utils
- Stats.initialize(
- factory=stats_utils.get_stats_factory(),
- export_legacy_names=conf.getboolean("metrics", "legacy_names_on",
fallback=True),
- )
- except ImportError:
- # ``stats_utils`` lives under ``airflow.observability.metrics`` in
current Airflow; if the
- # import path changes or is unavailable, fall through silently — gauge
calls will simply
- # land on ``NoStatsLogger`` as before, which is no worse than current
behavior.
- pass
+ try:
+ Stats.initialize(
+ factory=stats_utils.get_stats_factory(),
+ export_legacy_names=airflow_conf.getboolean("metrics",
"legacy_names_on", fallback=True),
+ )
+ except TypeError:
+ # args were changed in #63932, for compat we try to fall back to
old way if first one fails
+ Stats.initialize(factory=stats_utils.get_stats_factory(Stats))
+
+ except Exception as err:
+ # Intentional catch-all: stats-related errors must not block DAG event
emission.
+ # If any errors are raised it will fall back to NoStatsLogger, which
is no worse than before.
+ log.warning("OpenLineage failed to initialize Stats in executor
initializer: `%s`", err)
+ log.debug("Exception details:", exc_info=True)
def _emit_manual_state_change_event(adapter_method, stats_key, **kwargs):
diff --git
a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
index 1183ce91f60..25bee70ce75 100644
--- a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
+++ b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
@@ -121,6 +121,110 @@ class MockExecutor:
print("Shutting down")
+class TestExecutorInitializer:
+ """Tests for _executor_initializer function."""
+
+ @patch("airflow.settings.configure_orm")
+ def test_executor_initializer_calls_configure_orm(self,
mock_configure_orm):
+ """Test that _executor_initializer always calls
settings.configure_orm().
+
+ This test runs on all Airflow versions to ensure ORM initialization
always happens.
+ """
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ _executor_initializer()
+
+ mock_configure_orm.assert_called_once()
+
+ @pytest.mark.skipif(AIRFLOW_V_3_2_PLUS, reason="Testing older Airflow
behavior")
+ @patch("airflow.settings.configure_orm")
+ @patch("airflow.providers.openlineage.plugins.listener.Stats")
+ def test_executor_initializer_skips_stats_on_older_af(self, mock_stats,
mock_configure_orm):
+ """Test that _executor_initializer skips Stats initialization on
Airflow < 3.2."""
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ _executor_initializer()
+
+ # configure_orm should be called
+ mock_configure_orm.assert_called_once()
+ # But Stats.initialize should NOT be called (early return)
+ mock_stats.initialize.assert_not_called()
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="Airflow 3.2+ tests")
+ @patch("airflow.settings.configure_orm")
+ @patch("airflow.providers.openlineage.plugins.listener.Stats")
+ def test_executor_initializer_initializes_stats_on_af32(self, mock_stats,
mock_configure_orm):
+ """Test that _executor_initializer initializes Stats on Airflow
3.2+."""
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ with
patch("airflow.observability.metrics.stats_utils.get_stats_factory") as
mock_get_factory:
+ mock_get_factory.return_value = MagicMock()
+
+ _executor_initializer()
+
+ mock_configure_orm.assert_called_once()
+ mock_stats.initialize.assert_called_once()
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="Airflow 3.2+ tests")
+ @patch("airflow.settings.configure_orm")
+ @patch("airflow.providers.openlineage.plugins.listener.Stats")
+ def test_executor_initializer_stats_fallback_on_type_error(self,
mock_stats, mock_configure_orm):
+ """Test that _executor_initializer falls back to old Stats.initialize
signature on TypeError."""
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ with
patch("airflow.observability.metrics.stats_utils.get_stats_factory") as
mock_get_factory:
+ mock_get_factory.return_value = MagicMock()
+ # First call to Stats.initialize raises TypeError (old signature
not supported)
+ # Second call succeeds (using the fallback)
+ mock_stats.initialize.side_effect = [TypeError("unexpected keyword
argument"), None]
+
+ _executor_initializer()
+
+ # Both calls to Stats.initialize should happen
+ assert mock_stats.initialize.call_count == 2
+ # get_stats_factory should be called twice: once for new
signature, once for fallback
+ assert mock_get_factory.call_count == 2
+ # Second call should pass Stats as argument
+ mock_get_factory.assert_called_with(mock_stats)
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="Airflow 3.2+ tests")
+ @patch("airflow.settings.configure_orm")
+ @patch("airflow.providers.openlineage.plugins.listener.Stats")
+ def test_executor_initializer_exception_does_not_break(self, mock_stats,
mock_configure_orm):
+ """Test that _executor_initializer doesn't break on exception and logs
warning."""
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ with
patch("airflow.observability.metrics.stats_utils.get_stats_factory") as
mock_get_factory:
+ # Make stats_utils.get_stats_factory raise an exception
+ mock_get_factory.side_effect = RuntimeError("stats initialization
failed")
+
+ # Should not raise, exception should be caught and logged
+ _executor_initializer()
+
+ # configure_orm should still have been called
+ mock_configure_orm.assert_called_once()
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="Airflow 3.2+ tests")
+ @patch("airflow.settings.configure_orm")
+ def test_executor_initializer_with_real_stats(self, mock_configure_orm):
+ """Integration test that calls real stats code to catch API changes.
+
+ This test ensures that the stats initialization code remains compatible
+ with the actual airflow.observability.metrics.stats_utils
implementation.
+ If the stats API changes in the future, this test will fail and alert
+ developers to update the initializer accordingly.
+
+ Note: configure_orm is mocked to avoid side effects; only stats code
is real.
+ """
+ from airflow.providers.openlineage.plugins.listener import
_executor_initializer
+
+ # Call with real stats code (not mocked), but ORM initialization mocked
+ # Should not raise and configure_orm should be called
+ _executor_initializer()
+
+ mock_configure_orm.assert_called_once()
+
+
@pytest.mark.skipif(AIRFLOW_V_3_0_PLUS, reason="Airflow 2 tests")
class TestOpenLineageListenerAirflow2:
@patch("airflow.models.TaskInstance.xcom_push")
