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 750617c3867 Fix entry-point plugins reported as duplicates on
RHEL/Fedora venvs (#74531)
750617c3867 is described below
commit 750617c3867ea632a59d1a56ffa33b43397aa767
Author: Shikhar Goel <[email protected]>
AuthorDate: Thu Nov 5 18:29:59 2026 +0530
Fix entry-point plugins reported as duplicates on RHEL/Fedora venvs (#74531)
* Fix entry-point plugins reported as duplicates on RHEL/Fedora venvs
On RHEL and Fedora, sys.path in a virtualenv holds both lib64/ and lib/
site-packages, and lib64 is a symlink to lib. importlib.metadata returns
every
distribution once per sys.path entry, so each entry-point plugin was loaded
twice. Since duplicate plugin names are recorded as plugin import errors, a
fresh install with e.g. the Kafka provider showed "Plugin
'kafka_event_producer' already registered" in the UI and logs.
* Deduplicate distributions when collecting entry points
Skipping identical entry points in the plugins manager only hid
byte-identical copies. The same package found on two sys.path entries
in different versions (a venv with --system-site-packages, a user-site
copy) still loaded twice, and two different packages declaring the
same plugin were silently collapsed instead of surfacing an error.
Keeping only the first distribution per normalized name, as
importlib.metadata.entry_points() does, fixes this for every caller of
entry_points_with_dist at once.
* Keep entry points of distributions without a Name when deduplicating
dist.metadata can be None, and on Python 3.11 the importlib_metadata
backport raises KeyError for a missing field instead of returning None.
Indexing it directly made such a distribution fall into the error handler,
so its entry points were dropped instead of being collected undeduplicated,
and mypy rejected the possibly-None metadata.
---------
Co-authored-by: sgoel2be24-cyber
<[email protected]>
---
.../src/airflow_shared/module_loading/__init__.py | 23 ++++--
.../tests/module_loading/test_entry_points.py | 83 +++++++++++++++++++---
2 files changed, 91 insertions(+), 15 deletions(-)
diff --git
a/shared/module_loading/src/airflow_shared/module_loading/__init__.py
b/shared/module_loading/src/airflow_shared/module_loading/__init__.py
index 60bead945e3..7bf01c475ac 100644
--- a/shared/module_loading/src/airflow_shared/module_loading/__init__.py
+++ b/shared/module_loading/src/airflow_shared/module_loading/__init__.py
@@ -21,6 +21,7 @@ import functools
import inspect
import logging
import pkgutil
+import re
import sys
from collections import defaultdict
from collections.abc import Callable, Iterator
@@ -149,8 +150,6 @@ def is_valid_dotpath(path: str) -> bool:
:param path: String to check
"""
- import re
-
if not isinstance(path, str):
return False
@@ -165,11 +164,26 @@ def is_valid_dotpath(path: str) -> bool:
return bool(re.match(pattern, path))
+def _normalize_dist_name(name: str) -> str:
+ # PEP 503 normalization, the same key importlib.metadata.entry_points()
deduplicates on.
+ return re.sub(r"[-_.]+", "-", name).lower()
+
+
@functools.cache
def _get_grouped_entry_points() -> dict[str, list[EPnD]]:
mapping: dict[str, list[EPnD]] = defaultdict(list)
+ seen_dist_names: set[str] = set()
for dist in metadata.distributions():
try:
+ # A distribution is yielded once per sys.path entry it is found on
(e.g. RHEL/Fedora
+ # venvs list both lib64/ and lib/ site-packages), possibly in
different versions.
+ # Only the first one is importable, so it is the only one whose
entry points count.
+ dist_metadata = dist.metadata
+ if dist_metadata and (name := dist_metadata.get("Name")):
+ normalized_name = _normalize_dist_name(name)
+ if normalized_name in seen_dist_names:
+ continue
+ seen_dist_names.add(normalized_name)
for e in dist.entry_points:
mapping[e.group].append((e, dist))
except Exception as e:
@@ -184,9 +198,8 @@ def entry_points_with_dist(group: str) -> Iterator[EPnD]:
This is like the ``entry_points()`` function from ``importlib.metadata``,
except it also returns the distribution the entry point was loaded from.
- Note that this may return multiple distributions to the same package if
they
- are loaded from different ``sys.path`` entries. The caller site should
- implement appropriate deduplication logic if needed.
+ Like ``entry_points()``, a distribution found on several ``sys.path``
entries
+ is only returned once: the first one found, which is the one that is
imported.
:param group: Filter results to only this entrypoint group
:return: Generator of (EntryPoint, Distribution) objects for the specified
groups
diff --git a/shared/module_loading/tests/module_loading/test_entry_points.py
b/shared/module_loading/tests/module_loading/test_entry_points.py
index 87b9f15b76e..ff22042aefe 100644
--- a/shared/module_loading/tests/module_loading/test_entry_points.py
+++ b/shared/module_loading/tests/module_loading/test_entry_points.py
@@ -20,18 +20,39 @@ from __future__ import annotations
from collections.abc import Iterable
from unittest import mock
-from airflow_shared.module_loading import entry_points_with_dist, metadata
+import pytest
+
+from airflow_shared.module_loading import _get_grouped_entry_points,
entry_points_with_dist, metadata
class MockDistribution:
- def __init__(self, name: str, entry_points: Iterable[metadata.EntryPoint])
-> None:
- self.metadata = {"Name": name}
+ def __init__(
+ self, name: str, entry_points: Iterable[metadata.EntryPoint], version:
str = "1.0.0"
+ ) -> None:
+ self.metadata: dict[str, str] | None = {"Name": name}
self.entry_points = entry_points
+ self.version = version
class MockMetadata:
+ def __init__(self, distributions: list[MockDistribution]) -> None:
+ self._distributions = distributions
+
def distributions(self):
- return [
+ return self._distributions
+
+
[email protected](autouse=True)
+def clear_entry_points_cache():
+ _get_grouped_entry_points.cache_clear()
+ yield
+ _get_grouped_entry_points.cache_clear()
+
+
[email protected](
+ "airflow_shared.module_loading.metadata",
+ MockMetadata(
+ [
MockDistribution(
"dist1",
[metadata.EntryPoint("a", "b", "group_x"),
metadata.EntryPoint("c", "d", "group_y")],
@@ -39,12 +60,54 @@ class MockMetadata:
MockDistribution("Dist2", [metadata.EntryPoint("e", "f",
"group_x")]),
MockDistribution("dist2", [metadata.EntryPoint("g", "h",
"group_x")]), # Duplicated name.
]
-
-
[email protected]("airflow_shared.module_loading.metadata", MockMetadata())
+ ),
+)
def test_entry_points_with_dist():
entries = list(entry_points_with_dist("group_x"))
- # Only "group_x" entries are loaded. Distributions are not deduplicated.
- assert [dist.metadata["Name"] for _, dist in entries] == ["dist1",
"Dist2", "dist2"]
- assert [ep.name for ep, _ in entries] == ["a", "e", "g"]
+ # Only "group_x" entries are loaded, and only from the first "dist2" found.
+ assert [dist.metadata["Name"] for _, dist in entries] == ["dist1", "Dist2"]
+ assert [ep.name for ep, _ in entries] == ["a", "e"]
+
+
[email protected](
+ "airflow_shared.module_loading.metadata",
+ MockMetadata(
+ [
+ # The same distribution on two sys.path entries (e.g. a venv and
the system
+ # site-packages it falls back to), installed there in different
versions.
+ MockDistribution(
+ "apache-airflow-providers-foo",
+ [metadata.EntryPoint("foo",
"airflow.providers.foo.plugins:FooPlugin", "airflow.plugins")],
+ version="2.0.0",
+ ),
+ MockDistribution(
+ "apache_airflow_providers_foo",
+ [metadata.EntryPoint("foo",
"airflow.providers.foo.plugin:FooPlugin", "airflow.plugins")],
+ version="1.0.0",
+ ),
+ ]
+ ),
+)
+def test_entry_points_with_dist_same_dist_in_two_versions_returns_first_only():
+ entries = list(entry_points_with_dist("airflow.plugins"))
+
+ assert [(ep.value, dist.version) for ep, dist in entries] == [
+ ("airflow.providers.foo.plugins:FooPlugin", "2.0.0")
+ ]
+
+
+# No metadata file at all, or metadata without a Name field (the key is
absent, not None).
[email protected]("dist_metadata", [None, {"Version": "1.0.0"}],
ids=["no-metadata", "no-name"])
+def
test_entry_points_with_dist_without_name_are_not_deduplicated(dist_metadata):
+ distributions = [
+ MockDistribution("unnamed", [metadata.EntryPoint("a", "b",
"group_x")]),
+ MockDistribution("unnamed", [metadata.EntryPoint("c", "d",
"group_x")]),
+ ]
+ for dist in distributions:
+ dist.metadata = dist_metadata
+
+ with mock.patch("airflow_shared.module_loading.metadata",
MockMetadata(distributions)):
+ entries = list(entry_points_with_dist("group_x"))
+
+ assert [ep.name for ep, _ in entries] == ["a", "c"]
