Honour dag_ignore_file_syntax in Task SDK and example plugin loading

shared/plugins_manager/src/airflow_shared/plugins_manager/plugins_manager.py

@@ -241,7 +241,7 @@ def _load_plugins_from_plugin_directory(
         log.debug("Note: Loading plugins from examples as well: %s", plugins_folder)
         example_plugins = importlib.import_module(example_plugins_module)
         example_plugins_folder = next(iter(example_plugins.__path__))
-        example_files = find_path_from_directory(example_plugins_folder, ".airflowignore")
+        example_files = find_path_from_directory(example_plugins_folder, ".airflowignore", ignore_file_syntax)
         plugin_search_locations.append((example_plugins.__name__, example_files))
 
     plugins: list[AirflowPlugin] = []

shared/plugins_manager/tests/plugins_manager/test_plugins_manager.py

@@ -29,8 +29,17 @@
     EntryPointSource,
     PluginsDirectorySource,
     _load_entrypoint_plugins,
+    _load_plugins_from_plugin_directory,
 )
 
+_PLUGIN_SOURCE = """
+from airflow_shared.plugins_manager import AirflowPlugin
+
+
+class Plugin(AirflowPlugin):
+    name = "{name}"
+"""
+
 
 @pytest.fixture
 def mock_metadata_distribution(mocker):
@@ -112,6 +121,32 @@ def test_entrypoint_plugin_errors_dont_raise_exceptions(self, mock_metadata_dist
             ) in import_errors.items()
 
 
+class TestLoadPluginsFromPluginDirectory:
+    def test_example_plugins_airflowignore_uses_ignore_file_syntax(self, tmp_path, monkeypatch):
+        """The example plugins folder is filtered with the same ignore syntax as the plugins folder."""
+        example_plugins = tmp_path / "example_plugins_for_test"
+        example_plugins.mkdir()
+        (example_plugins / "__init__.py").touch()
+        # ``^`` and ``$`` are literal characters to the glob reader, so only the regexp reader matches.
+        (example_plugins / ".airflowignore").write_text(r"^ignored_.*\.py$")
+        for stem in ("kept_plugin", "ignored_plugin"):
+            (example_plugins / f"{stem}.py").write_text(_PLUGIN_SOURCE.format(name=stem))
+        plugins_folder = tmp_path / "plugins"
+        plugins_folder.mkdir()
+        monkeypatch.syspath_prepend(tmp_path)
+        monkeypatch.delitem(sys.modules, "example_plugins_for_test", raising=False)
+
+        plugins, import_errors = _load_plugins_from_plugin_directory(
+            plugins_folder=str(plugins_folder),
+            load_examples=True,
+            example_plugins_module="example_plugins_for_test",
+            ignore_file_syntax="regexp",
+        )
+
+        assert import_errors == {}
+        assert [plugin.name for plugin in plugins] == ["kept_plugin"]
+
+
 class TestAirflowPluginTeamName:
     def test_team_name_defaults_to_none(self):
         class GlobalPlugin(AirflowPlugin):

task-sdk/src/airflow/sdk/plugins_manager.py

@@ -103,11 +103,13 @@ def __register_plugins(plugin_instances: list[AirflowPlugin], errors: dict[str,
 
     with stats.timer() as timer:
         load_examples = conf.getboolean("core", "LOAD_EXAMPLES")
+        ignore_file_syntax = conf.get_mandatory_value("core", "DAG_IGNORE_FILE_SYNTAX", fallback="glob")
         __register_plugins(
             *_load_plugins_from_plugin_directory(
                 plugins_folder=settings.PLUGINS_FOLDER,
                 load_examples=load_examples,
                 example_plugins_module="airflow.example_dags.plugins" if load_examples else None,
+                ignore_file_syntax=ignore_file_syntax,
             )
         )
         __register_plugins(*_load_entrypoint_plugins())

task-sdk/tests/task_sdk/test_plugins_manager.py

@@ -24,11 +24,21 @@
 from airflow.sdk import plugins_manager
 from airflow.sdk.plugins_manager import AirflowPlugin
 
+from tests_common.test_utils.config import conf_vars
+
 # ``_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
 
+_PLUGIN_SOURCE = """
+from airflow.sdk.plugins_manager import AirflowPlugin
+
+
+class Plugin(AirflowPlugin):
+    name = "{name}"
+"""
+
 
 class TestGetPlugins:
     """Mirrors ``TestPluginsManager`` in ``airflow-core/tests/unit/plugins/test_plugins_manager.py``."""
@@ -110,3 +120,27 @@ class PluginADuplicateName(AirflowPlugin):
 
         assert [plugin.name for plugin in plugins] == ["plugin_a"]
         assert len(import_errors) == 1
+
+    @conf_vars({("core", "dag_ignore_file_syntax"): "regexp", ("core", "load_examples"): "False"})
+    def test_plugins_folder_airflowignore_uses_dag_ignore_file_syntax(self, tmp_path):
+        """
+        ``.airflowignore`` in the plugins folder is read with ``[core] dag_ignore_file_syntax``.
+
+        ``airflow.plugins_manager`` already passes the setting through. If this copy silently falls
+        back to ``glob``, the scheduler and the task-running process load a different set of plugins
+        from the same folder.
+        """
+        # ``^`` and ``$`` are literal characters to the glob reader, so only the regexp reader matches.
+        (tmp_path / ".airflowignore").write_text(r"^ignored_.*\.py$")
+        for stem in ("kept_plugin", "ignored_plugin"):
+            (tmp_path / f"{stem}.py").write_text(_PLUGIN_SOURCE.format(name=stem))
+
+        with (
+            mock.patch.object(plugins_manager.settings, "PLUGINS_FOLDER", str(tmp_path)),
+            mock.patch.object(plugins_manager.settings, "LAZY_LOAD_PROVIDERS", True),
+            mock.patch.object(plugins_manager, "_load_entrypoint_plugins", return_value=([], {})),
+        ):
+            plugins, import_errors = plugins_manager._get_plugins()
+
+        assert import_errors == {}
+        assert [plugin.name for plugin in plugins] == ["kept_plugin"]