uranusjr commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4389716627
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -797,9 +828,47 @@ def claim_priority_files(self) -> list[DagFileInfo]:
Default implementation reads from the metadata DB; override to source
requests from an API.
Review Comment:
Since this function is designed as an override point, we should leverage it.
Extract a requested-work source that the manager calls at fixed lifecycle
points. The DB implementation wraps `_claim_priority_files` and
`_fetch_callbacks_from_db` and does nothing on the other hooks, because DB mode
deletes requests when it claims them. All the claim and ack state moves into
the API implementation.
The manager would take `work_source: RequestedWorkSource`, and
`run_with_api` would pass `ExecutionAPIWorkSource(client)`. Then:
* `_finish_callback_attempt` and `_finish_priority_claims` become
`self.work_source.attempt_finished(proc.id)`.
* The claim release in `_discard_unusable_bundle_source` becomes
`bundle_unusable(...)`.
The new claim/ack condition in `max_runs_reached` and the ack in `end()`
become `has_outstanding_work()` and `flush(...)`.
Most of the `api_client is None` checks would disappear, and `api_client`
would only be needed for publication and secrets. This is the main blocker. The
branching started in #74793, so the refactor probably belongs there.
--
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]
