fix(dag-processing): don't drop import errors when no listener is reg…

[kjh0623]

…istered

_update_import_errors() eagerly evaluated ParseImportError.full_file_path() to build the on_new/on_existing_dag_import_error listener arguments, even when no listener was registered. full_file_path() instantiates the DAG bundle via DagBundlesManager().get_bundle(), which for some bundle types (e.g. the Git bundle) resolves an Airflow Connection. Performing that work inside the DAG-parsing transaction can disturb it and roll back the not-yet-flushed ParseImportError row, so the DAG ends up flagged as broken (has_import_errors) while no import error is ever shown in the UI.

Guard both listener notifications with get_listener_manager().has_listeners so full_file_path() is only resolved when a listener is actually registered. This fixes the silent drop in the common (no-listener) case and avoids the wasted bundle instantiation on every import error.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Vamsi-klu]

Had a look through this and the guard reads right to me. full_file_path() in models/errors.py does DagBundlesManager().get_bundle(...).path, so gating both notification calls in _update_import_errors behind get_listener_manager().has_listeners keeps that bundle instantiation out of the parsing transaction when no listener is registered, which is the common case. Nice that it reuses the same has_listeners check already in dagbag.py instead of inventing a new signal. The regression test mocking full_file_path with assert_not_called() and then asserting the ParseImportError row still persists lands right on the behavior that was broken, and as a bonus the existing-error branch now skips the extra select(ParseImportError) query that only ever existed to feed the listener arg.

[kjh0623]

Thanks for the review!

[dhkim1920]

Thanks for fixing the no-listener case.

I think there may still be a similar transaction issue when any listener plugin is registered.

ListenerManager.has_listeners only checks whether any listener plugin exists, not whether an import-error hook is actually implemented. That means ParseImportError.full_file_path() is still evaluated while building the hook arguments, even if none of the registered listeners handle import errors.

For bundle types like the Git bundle, calling full_file_path() can instantiate the bundle and resolve an Airflow Connection inside the parsing transaction. If that ends up rolling back the session, the pending import-error row could still be lost.

Would it be reasonable to pass the already-known bundle_path down the parsing call chain and build the listener filename with str(bundle_path / relative_fileloc) instead? DagFileInfo and DagBag already have that information, so it avoids initializing the bundle again inside the transaction regardless of which listeners are registered.

As a side benefit, the existing-error path would no longer need the extra select(ParseImportError) query just to construct the listener filename.

I tried this approach locally with a registered listener for both new and existing import errors, and the listener still received the expected filename while the ParseImportError row remained persisted without calling full_file_path().