fix: report duplicate plugin names as import errors

airflow-core/src/airflow/plugins_manager.py

@@ -99,7 +99,10 @@ 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:
-                log.warning("Plugin %r already registered, skipping", plugin_instance.name)
+                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)

airflow-core/tests/unit/plugins/test_plugins_manager.py

@@ -160,7 +160,34 @@ class PluginC(AirflowPlugin):
         assert "plugin_b" in plugin_names
         assert "plugin_c" in plugin_names
         assert len(plugins) == 3
-        assert not import_errors
+
+    def test_duplicate_plugin_name_is_reported_as_import_error(self):
+        from airflow import plugins_manager
+
+        class PluginA(AirflowPlugin):
+            name = "plugin_a"
+
+        class PluginADuplicateName(AirflowPlugin):
+            name = "plugin_a"
+
+        plugin_a = PluginA()
+        plugin_a_dup = PluginADuplicateName()
+
+        with (
+            mock.patch(
+                "airflow.plugins_manager._load_plugins_from_plugin_directory",
+                return_value=([plugin_a], {}),
+            ),
+            mock.patch(
+                "airflow.plugins_manager._load_entrypoint_plugins",
+                return_value=([plugin_a_dup], {}),
+            ),
+            mock.patch("airflow.plugins_manager._load_providers_plugins", return_value=([], {})),
+        ):
+            plugins, import_errors = plugins_manager._get_plugins()
+
+        assert [p.name for p in plugins] == ["plugin_a"]
+        assert len(import_errors) == 1
 
     def test_should_warning_about_incompatible_plugins(self, caplog):
         class AirflowAdminViewsPlugin(AirflowPlugin):