This is an automated email from the ASF dual-hosted git repository.
potiuk 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 49789f2846e Fix plugin registration returning early on duplicate names
in Task SDK (#72253)
49789f2846e is described below
commit 49789f2846e06c25fe7b78505e0f6ea1a4fa874b
Author: Oleksii Maksymiv <[email protected]>
AuthorDate: Sat Aug 29 19:12:33 2026 +0300
Fix plugin registration returning early on duplicate names in Task SDK
(#72253)
__register_plugins used `return` instead of `continue` when encountering
a duplicate plugin name, so the first duplicate abandoned the rest of the
batch and every plugin after it was silently dropped.
airflow-core carries a duplicate of this function and was fixed twice: in
#60498 (`return` became a warning plus `continue`) and in #66649 (also
record the duplicate in import_errors). Neither fix reached the Task SDK
copy. Because that copy is the one used by the task-running process,
plugins would load for the scheduler and API server but silently go
missing inside tasks - most visibly, task instance listeners registered
by a plugin would never fire.
The __register_plugins body is now byte-identical to the airflow-core one.
---
task-sdk/src/airflow/sdk/plugins_manager.py | 6 +-
task-sdk/tests/task_sdk/test_plugins_manager.py | 112 ++++++++++++++++++++++++
2 files changed, 117 insertions(+), 1 deletion(-)
diff --git a/task-sdk/src/airflow/sdk/plugins_manager.py
b/task-sdk/src/airflow/sdk/plugins_manager.py
index dca92ea0be8..3934e54ef5e 100644
--- a/task-sdk/src/airflow/sdk/plugins_manager.py
+++ b/task-sdk/src/airflow/sdk/plugins_manager.py
@@ -85,7 +85,11 @@ def _get_plugins() -> tuple[list[AirflowPlugin], dict[str,
str]]:
def __register_plugins(plugin_instances: list[AirflowPlugin], errors:
dict[str, str]) -> None:
for plugin_instance in plugin_instances:
if plugin_instance.name in loaded_plugins:
- return
+ message = f"Plugin {plugin_instance.name!r} already
registered, skipping"
+ log.warning(message)
+ name = str(plugin_instance.source) if plugin_instance.source
else plugin_instance.name or ""
+ import_errors[name] = message
+ continue
loaded_plugins.add(plugin_instance.name)
try:
diff --git a/task-sdk/tests/task_sdk/test_plugins_manager.py
b/task-sdk/tests/task_sdk/test_plugins_manager.py
new file mode 100644
index 00000000000..92d10c7fc9d
--- /dev/null
+++ b/task-sdk/tests/task_sdk/test_plugins_manager.py
@@ -0,0 +1,112 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+
+from airflow.sdk import plugins_manager
+from airflow.sdk.plugins_manager import AirflowPlugin
+
+# ``_disable_ol_plugin`` in ``task-sdk/tests/conftest.py`` is a session-scoped
autouse fixture that
+# replaces ``_get_plugins`` with a stub returning no plugins. Capture the
real, ``@cache``-wrapped
+# function here at collection time — before any fixture runs — so these tests
can put it back.
+_REAL_GET_PLUGINS = plugins_manager._get_plugins
+
+
+class TestGetPlugins:
+ """Mirrors ``TestPluginsManager`` in
``airflow-core/tests/unit/plugins/test_plugins_manager.py``."""
+
+ @pytest.fixture(autouse=True)
+ def _use_real_get_plugins(self, monkeypatch):
+ monkeypatch.setattr(plugins_manager, "_get_plugins", _REAL_GET_PLUGINS)
+ _REAL_GET_PLUGINS.cache_clear()
+ yield
+ _REAL_GET_PLUGINS.cache_clear()
+
+ def
test_duplicate_plugin_name_does_not_prevent_loading_subsequent_plugins(self):
+ """
+ A duplicate name must skip that one plugin, not abandon the rest of
the batch.
+
+ ``__register_plugins`` used to ``return`` on the first duplicate, so
every plugin
+ enumerated after it was dropped without an error. Here ``plugin_b`` is
loaded from
+ the plugins directory and again from an entry point; ``plugin_c``
follows the
+ duplicate in that second batch and is the one that used to disappear.
+ """
+
+ class PluginA(AirflowPlugin):
+ name = "plugin_a"
+
+ class PluginB(AirflowPlugin):
+ name = "plugin_b"
+
+ class PluginC(AirflowPlugin):
+ name = "plugin_c"
+
+ plugin_a = PluginA()
+ plugin_b = PluginB()
+ plugin_b_dup = PluginB()
+ plugin_c = PluginC()
+
+ with (
+ mock.patch.object(plugins_manager.settings, "PLUGINS_FOLDER",
"/dev/null"),
+ mock.patch.object(plugins_manager.settings, "LAZY_LOAD_PROVIDERS",
False),
+ mock.patch.object(
+ plugins_manager,
+ "_load_plugins_from_plugin_directory",
+ return_value=([plugin_a, plugin_b], {}),
+ ),
+ mock.patch.object(
+ plugins_manager, "_load_entrypoint_plugins",
return_value=([plugin_b_dup, plugin_c], {})
+ ),
+ mock.patch.object(plugins_manager, "_load_providers_plugins",
return_value=([], {})),
+ ):
+ plugins, _ = plugins_manager._get_plugins()
+
+ plugin_names = [plugin.name for plugin in plugins]
+ assert "plugin_a" in plugin_names
+ assert "plugin_b" in plugin_names
+ assert "plugin_c" in plugin_names
+ assert len(plugins) == 3
+
+ def test_duplicate_plugin_name_is_reported_as_import_error(self):
+ """The skipped duplicate is surfaced to the caller rather than dropped
silently."""
+
+ class PluginA(AirflowPlugin):
+ name = "plugin_a"
+
+ class PluginADuplicateName(AirflowPlugin):
+ name = "plugin_a"
+
+ plugin_a = PluginA()
+ plugin_a_dup = PluginADuplicateName()
+
+ with (
+ mock.patch.object(plugins_manager.settings, "PLUGINS_FOLDER",
"/dev/null"),
+ mock.patch.object(plugins_manager.settings, "LAZY_LOAD_PROVIDERS",
False),
+ mock.patch.object(
+ plugins_manager, "_load_plugins_from_plugin_directory",
return_value=([plugin_a], {})
+ ),
+ mock.patch.object(plugins_manager, "_load_entrypoint_plugins",
return_value=([plugin_a_dup], {})),
+ mock.patch.object(plugins_manager, "_load_providers_plugins",
return_value=([], {})),
+ ):
+ plugins, import_errors = plugins_manager._get_plugins()
+
+ assert [plugin.name for plugin in plugins] == ["plugin_a"]
+ assert len(import_errors) == 1
