Warn only about bindings of Dags missing from the parse result

A Dag rejected for another team's plugin class already gets an import error on every parse, so warning about its dropped bindings as well was noise that buried the case worth a warning: a binding for a Dag the parse never produced, which points at a bug in the parse.

Author: jason810496

airflow-core/src/airflow/dag_processing/collection.py

@@ -730,6 +730,7 @@ def _reconcile_task_handler_bindings(
     bindings: Iterable[TaskHandlerBinding],
     *,
     dag_ids: Collection[str],
+    rejected_dag_ids: Collection[str],
     artifact_bundle_names: Collection[str],
     dag_bundle_name: str,
     dag_relative_fileloc: str,
@@ -742,18 +743,19 @@ def _reconcile_task_handler_bindings(
     that moved to another file, which that file's parse reconciles. A Dag bound to an artifact outside
     *artifact_bundle_names*, or to one with no recorded row, keeps its rows as they are. Artifact rows are
     only read: the probed-artifact write records them, and the Dag processor's orphan sweep deletes them.
+    Bindings of *rejected_dag_ids*, Dags the parse produced but that are not persisted, are dropped
+    quietly: their import error explains why.
     """
     by_task: dict[tuple[str, str], list[TaskHandlerBinding]] = defaultdict(list)
-    ignored_dag_ids: set[str] = set()
+    unknown_dag_ids: set[str] = set()
     for binding in bindings:
         if binding.dag_id in dag_ids:
             by_task[(binding.dag_id, binding.task_id)].append(binding)
-        else:
-            ignored_dag_ids.add(binding.dag_id)
-    if ignored_dag_ids:
+        elif binding.dag_id not in rejected_dag_ids:
+            unknown_dag_ids.add(binding.dag_id)
+    if unknown_dag_ids:
         log.warning(
-            "Ignoring task handler bindings of Dags this parse does not persist",
-            dag_ids=sorted(ignored_dag_ids),
+            "Ignoring task handler bindings of Dags not in the parse result", dag_ids=sorted(unknown_dag_ids)
         )
     # The parse reports a task bound twice as an import error; this only keeps the rows as they are.
     if conflicting := {dag_id for (dag_id, _), group in by_task.items() if len(group) > 1}:
@@ -848,6 +850,7 @@ def update_dag_parsing_results_in_db(
     if task_handler_bindings is not None and task_handler_artifact_bundles is None:
         raise ValueError("task_handler_artifact_bundles is required with task_handler_bindings")
     accepted = _reject_other_teams_plugin_classes(bundle_name, dags, import_errors, session=session)
+    rejected_ids: set[str] = set()
     if len(accepted) != len(dags):
         # A rejected Dag may have no ``dag`` row yet, and dag_warning has a foreign key to it.
         rejected_ids = {dag.dag_id for dag in dags} - {dag.dag_id for dag in accepted}
@@ -891,6 +894,7 @@ def update_dag_parsing_results_in_db(
                         _reconcile_task_handler_bindings(
                             task_handler_bindings,
                             dag_ids={dag.dag_id for dag in dags},
+                            rejected_dag_ids=rejected_ids,
                             artifact_bundle_names=task_handler_artifact_bundles,
                             dag_bundle_name=bundle_name,
                             dag_relative_fileloc=relative_fileloc,

airflow-core/tests/unit/dag_processing/test_collection.py

@@ -1961,6 +1961,27 @@ def test_team_class_is_only_stored_for_its_team(
             assert list(errors) == [(bundle_name, "team_dag.py")]
             assert f"belonging to {plugin_team}" in errors[(bundle_name, "team_dag.py")]
 
+    @conf_vars({("core", "multi_team"): "True"})
+    def test_task_handler_bindings_of_a_rejected_dag_are_dropped_quietly(self, bundle, session, caplog):
+        bundle_name = bundle(True)
+        _record(session, "etl.jar")
+        with mock_plugin_manager(plugins=[self._plugin("other_team", "timetables", AfterWorkdayTimetable)]):
+            update_dag_parsing_results_in_db(
+                bundle_name=bundle_name,
+                bundle_version=None,
+                dags=[self._serialized(schedule=AfterWorkdayTimetable()), self._serialized("plain_dag")],
+                import_errors={},
+                parse_duration=None,
+                warnings=set(),
+                session=session,
+                relative_fileloc="team_dag.py",
+                task_handler_bindings=[_make_binding("team_dag"), _make_binding("plain_dag")],
+                task_handler_artifact_bundles={ARTIFACT_BUNDLE},
+            )
+
+        assert session.scalars(select(LangSDKTaskHandler.dag_id)).all() == ["plain_dag"]
+        assert not any("task handler bindings" in entry["event"] for entry in caplog.entries)
+
     @conf_vars({("core", "multi_team"): "True"})
     @pytest.mark.parametrize(
         "weight_rule",
@@ -2478,7 +2499,7 @@ def test_task_handler_bindings_for_dags_outside_the_result_are_dropped(self, ses
 
         assert _get_recorded_handlers(session) == {("etl", "extract", "etl.jar")}
         assert {
-            "event": "Ignoring task handler bindings of Dags this parse does not persist",
+            "event": "Ignoring task handler bindings of Dags not in the parse result",
             "dag_ids": ["ghost"],
         } in caplog