gopidesupavan opened a new pull request, #63795:
URL: https://github.com/apache/airflow/pull/63795
### What
Dag processor callback fetching now stores bundle_name on dag-processor
callback rows and filters callbacks by bundle in SQL before applying
max_callbacks_per_loop.
### Why
Previously, _fetch_callbacks() applied LIMIT before bundle filtering. In a
multi-bundle setup, callbacks for unrelated bundles could consume the manager’s
fetch window and starve callbacks for the bundles that manager actually owned.
Approaches considered:
#### Appraoche 1:
Loop-based fetching in _fetch_callbacks()
Why not chosen:
it would fix the symptom, but it keeps callback routing dependent on
repeatedly deserializing mixed-bundle rows and makes the fetch path more
complex.
#### Approach 2 choosen:
Add bundle_name to callback rows and filter in SQL:
It fixes the problem at the source, keeps _fetch_callbacks() simple, avoids
unnecessary deserialization of unrelated callbacks, and makes pending
dag-processor callbacks queryable by bundle.
Repro:
Configure a dag processor manager that handles bundle testing.
Set max_callbacks_per_loop = 2.
Insert three pending dag-processor callbacks,
bundle other-bundle-a
bundle other-bundle-b
bundle testing
Call _fetch_callbacks()
the newly added test proves this issue
`test_fetch_callbacks_filters_by_bundle_before_limit`
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
