You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "bunnysayzz (via GitHub)" <gi...@apache.org> on 2026/09/30 18:22:41 UTC
[PR] Stop killing callback processors for zipped Dags on bundle refresh [airflow]
bunnysayzz opened a new pull request, #73979:
URL: https://github.com/apache/airflow/pull/73979
Bundle refreshes were SIGKILLing callback processors for Dags packaged inside zip archives even though the Dag files were still present.
The scanner reports a zip archive as a single entry (`my_dags.zip`) while callback processors are keyed by the Dag file path inside the archive (`my_dags.zip/my_dag.py`). `terminate_orphan_processes` compared the two directly, so each refresh saw the inner path as missing and killed the processor. The same direct comparison in `purge_removed_files_from_queue` and `remove_orphaned_file_stats` dropped the Dag from the queue and wiped its stats on every refresh too.
The presence check now treats a path inside a zip as observed while its containing archive is in the scan set (`_file_is_present` in `manager.py`), applied to all three call sites so they stay consistent.
Closes #73858.
Tests: four new unit tests in `airflow-core/tests/unit/dag_processing/test_manager.py` covering the kill/no-kill directions for `terminate_orphan_processes`, the queue purge, and the stats cleanup.
Note: I could not run the Airflow test suite in this environment, so CI will need to confirm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop killing callback processors for zipped Dags on bundle refresh [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #73979:
URL: https://github.com/apache/airflow/pull/73979#discussion_r4148584316
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1218,24 +1218,48 @@ def handle_removed_files(self, known_files: dict[str, set[DagFileInfo]]):
def purge_removed_files_from_queue(self, present: set[DagFileInfo]):
"""Remove from queue any files no longer observed locally."""
present_keys = {file.presence_key for file in present}
- self._file_queue = OrderedDict((x, None) for x in self._file_queue if x.presence_key in present_keys)
+ self._file_queue = OrderedDict(
+ (x, None) for x in self._file_queue if self._file_is_present(x, present_keys)
+ )
stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
def remove_orphaned_file_stats(self, present: set[DagFileInfo]):
"""Remove the stats for any dag files that don't exist anymore."""
present_keys = {file.presence_key for file in present}
- stats_to_remove = {file for file in self._file_stats if file.presence_key not in present_keys}
+ stats_to_remove = {
+ file for file in self._file_stats if not self._file_is_present(file, present_keys)
+ }
for file in stats_to_remove:
del self._file_stats[file]
+ @staticmethod
+ def _file_is_present(file: DagFileInfo, present_keys: set[tuple[str, Path]]) -> bool:
+ """
+ Check whether a tracked file is still observed in the bundle scan.
+
+ Dag files inside a zip archive are scanned as the archive itself
+ (``_find_files_in_bundle`` yields the ``.zip`` entry, not the inner
+ files), so an inner path counts as present while its containing
+ archive is still observed. Without this, a bundle refresh would
+ treat e.g. ``my_dags.zip/my_dag.py`` as removed and kill a
+ still-running callback processor for it.
Review Comment:
Keep the first line as the docstring. Then remaining lines in a comment right above the code,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop killing callback processors for zipped Dags on bundle refresh [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #73979:
URL: https://github.com/apache/airflow/pull/73979#issuecomment-5917175430
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](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-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](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) 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](https://www.apache.org/foundation/policies/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]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Stop killing callback processors for zipped Dags on bundle refresh [airflow]
Posted by "bunnysayzz (via GitHub)" <gi...@apache.org>.
bunnysayzz commented on code in PR #73979:
URL: https://github.com/apache/airflow/pull/73979#discussion_r4150073456
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1218,24 +1218,48 @@ def handle_removed_files(self, known_files: dict[str, set[DagFileInfo]]):
def purge_removed_files_from_queue(self, present: set[DagFileInfo]):
"""Remove from queue any files no longer observed locally."""
present_keys = {file.presence_key for file in present}
- self._file_queue = OrderedDict((x, None) for x in self._file_queue if x.presence_key in present_keys)
+ self._file_queue = OrderedDict(
+ (x, None) for x in self._file_queue if self._file_is_present(x, present_keys)
+ )
stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
def remove_orphaned_file_stats(self, present: set[DagFileInfo]):
"""Remove the stats for any dag files that don't exist anymore."""
present_keys = {file.presence_key for file in present}
- stats_to_remove = {file for file in self._file_stats if file.presence_key not in present_keys}
+ stats_to_remove = {
+ file for file in self._file_stats if not self._file_is_present(file, present_keys)
+ }
for file in stats_to_remove:
del self._file_stats[file]
+ @staticmethod
+ def _file_is_present(file: DagFileInfo, present_keys: set[tuple[str, Path]]) -> bool:
+ """
+ Check whether a tracked file is still observed in the bundle scan.
+
+ Dag files inside a zip archive are scanned as the archive itself
+ (``_find_files_in_bundle`` yields the ``.zip`` entry, not the inner
+ files), so an inner path counts as present while its containing
+ archive is still observed. Without this, a bundle refresh would
+ treat e.g. ``my_dags.zip/my_dag.py`` as removed and kill a
+ still-running callback processor for it.
Review Comment:
done, moved the explanation above the def as a comment and kept the docstring to one line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org