vuonghoangbntt opened a new issue, #75839:
URL: https://github.com/apache/airflow/issues/75839
### Under which category would you file this issue?
Providers
### Apache Airflow version
3.3.1
### What happened and how to reproduce it?
**Issue Description**
On a bundle without versioning (for example `LocalDagBundle`, or with
`disable_bundle_versioning = True`), a callback-only run wipes the file's
parse stats.
A callback with no `bundle_version` gets the same `_file_stats` key as the
regular bundle
scan. For a callback-only result, `process_parse_results` builds a fresh
stat:
```python
stat = DagFileStat(
last_duration=run_duration,
run_count=run_count, # Don't increment for callback-only processing
)
```
and `handle_parsing_result` stores it over the existing one. So
`last_finish_time`,
`num_dags` and `import_errors` are reset to `None`, 0 and 0. Until the next
regular parse:
- the file looks never parsed, so it is queued again right away (one extra
parse per
callback);
- `dagbag_size` and `dag_processing.import_errors` drop by that file's
numbers;
- stale-Dag detection skips the file, because it only considers stats with a
`last_finish_time`.
This comes from #55698, which stopped callback-only runs from updating the
timestamps.
Its intent was to leave them alone, but the whole stat is replaced.
**Steps to reproduce:**
Add this to `TestDagFileProcessorManager` in
`airflow-core/tests/unit/dag_processing/test_manager.py`. It fails on main
at the last
line, where the stat comes back as `(None, 0, 0)`:
```python
def
test_callback_only_parse_keeps_the_scan_stat_on_an_unversioned_bundle(self):
manager = DagFileProcessorManager(max_runs=3)
file = _get_file_infos(["file_1.py"])[0] # unversioned: the callback's
key is the scan's key
parsed_at = timezone.datetime(2020, 1, 5)
manager._file_stats[file] = DagFileStat(
num_dags=3, import_errors=1, last_finish_time=parsed_at,
last_duration=1.0, run_count=2
)
processor, _ = self.mock_processor(start_time=time.monotonic() - 1)
processor.had_callbacks = True
processor.parsing_result = None
manager.handle_parsing_result(file, processor)
stat = manager._file_stats[file]
assert stat.run_count == 2
assert (stat.last_finish_time, stat.num_dags, stat.import_errors) ==
(parsed_at, 3, 1) # fails
```
### What you think should happen instead?
A callback-only run should leave the parse stats as they are: carry over the
previous
`last_finish_time`, `num_dags` and `import_errors`, and only update
`last_duration`
(`run_count` already stays the same). It should not stamp `now` either,
since that brings
back the stale-detection problem #55698 fixed.
### Operating System
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
