Repository navigation
Add HTTP publication to the Dag processor manager - #74793
Merged
ephraimbuddy merged 13 commits intoNov 6, 2026
Merged
Conversation
This was referenced Oct 15, 2026
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 16, 2026 15:35
7a475ef to
b878ba9
Compare
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 16, 2026 16:59
b878ba9 to
1ca9748
Compare
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
3 times, most recently
from
October 20, 2026 17:05
2c27096 to
f8222be
Compare
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 20, 2026 17:24
f8222be to
4a32963
Compare
ephraimbuddy
marked this pull request as ready for review
October 20, 2026 17:26
ephraimbuddy
requested review from
amoghrajesh,
ashb and
jedcunningham
as code owners
October 20, 2026 17:26
|
👋 hi -- we have been noticing it has been taking hours to start some of our CI jobs. I was poking around what jobs are running in the apache org (which all share a pool of runners) One set of jobs that happened to be running when I checked were the ones related to this stack of PRs. It seems like each of the stacked pRs kicks off 11 tests that take around 22m -- so like each push / rebase of the stack results in maybe like 32 hours of CI time used. I wonder if there is any way to reduce this (for example, by making these PRs in a fork rather than in the main repo, which triggers CI on each commit) More details here |
Contributor
Author
Hi, thanks for bringing this to my notice. The stack feature is not supported on a fork. What I will do now that the PR is already up is to reduce the frequency of pushes. |
Thank you @ephraimbuddy -- I appreciate it I would say DataFusion currently also wastes far too many worker hours too, but hopefully working together we can figure out how to reduce the load on the shared runners |
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 22, 2026 16:58
4a32963 to
3e78842
Compare
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 28, 2026 20:16
86918a2 to
820006c
Compare
Contributor
|
The re-run of the jobs stopped by EC2 Spot interruptions finished: attempt 2 passed. |
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 29, 2026 16:55
820006c to
f16729c
Compare
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
October 30, 2026 15:40
f16729c to
e2211cf
Compare
kaxil
reviewed
Nov 4, 2026
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
2 times, most recently
from
November 5, 2026 13:11
5610406 to
ad44144
Compare
kaxil
approved these changes
Nov 5, 2026
A processor without metadata database access needs to retain completed imports until the server accepts them, while continuing to supervise its other parsing processes. Bundle setup and parse-time credentials also need the bundle's authorization context throughout this path.
A processor waiting to publish a bundle listing must not resend it every loop while the API server is failing, adding load when the server can least absorb it. A parse result the API would refuse should reach users as an import error, as a result the server rejects already does, rather than being reparsed silently. Nothing needed to sync bundles without their URLs.
API-mode processors must collect finished work without metadata database access. Eager session construction fails before result handling can choose the API path, and a failed database commit must not stop other files. A lost parse-token response also needs to remain visible to the caller so the manager can retry without blocking its processing loop.
Requested parses and callbacks need a replaceable delivery boundary so the manager can schedule work independently of its database or Execution API transport.
Parser failures must reach the API so stale Dags can age out and users can see import errors. Inventory recovery and transient API failures must leave the processor responsive to heartbeats and parsing work.
The Task SDK answers a failed Variable or Connection lookup as a missing value, so Variable.get(default=...) returns its default. A parse that met a dropped connection, a 5xx or any other unanswered request could therefore publish a Dag built from fallback values as a new Dag version. Such a parse is now handled like one that timed out or exited without a result: the file keeps its previous publication, and an import error naming the request is reported only once that publication is older than stale_dag_threshold. The API deactivates a file's Dags as soon as it records an import error, while DB mode keeps them scheduled until the threshold passes, so one crash or OOM kill no longer pauses them. A restarted manager measures from its own start, so restarting the API alongside the processor does not deactivate every file whose first parse fails. Bundle code on the parsing loop looked secrets up with the Task SDK's retry policy, waiting up to eight seconds between attempts, while the same loop answers every parse process. One unavailable API stalled parsing for up to half a minute per lookup. On the loop a failed request is now retried once, at once, and only when the Task SDK would retry it, so a timeout, 408 or 429 is not resent. Async lookups were answered as missing because the manager's channel had no async path. Startup keeps the waiting retries, and its catalog read no longer gives up on a response slower than a second, so the processor starts while the API is briefly slow or restarting. A refresh request for a bundle that could not be refreshed yet, because its inventory was cooling down after a 413 or still waiting to be published, or because the catalog read had failed, read the catalog on every loop; it now waits for the next periodic read. Once the API reports that this processor's Job has ended, the manager stops publishing instead of logging and dropping every remaining result. An archive held back for one unrepresentable member is now named in the log, so an operator can tell that its other Dags stopped updating.
API mode relies on statement order in methods that also serve DB mode to keep the manager off the metadata database, and nothing tested it: removing the stale Dag scan guard, the startup provisioning check, the stale bundle version cleanup branch or the skip after a failed catalog read passed every test. The catalog guard sits under a broad exception handler, so a database read there would only have logged an error. Each guard now has a non-DB test, and one test runs the parsing loop with every database entry point forbidden. Several existing tests could not fail. The requested-work test patched the manager after the default work source had bound the original methods, and the timed-out parse test published before asserting, while publication drops an obsolete result on its own. Removing the pending-work checks in max_runs_reached or the exponential term of the publication retry delay passed every test, and the failed-publication case asserted the same as the successful one. The periodic refresh of an unchanged versioned bundle, which every Git bundle takes, had no API-mode test; taking the DB-mode shortcut there would skip the inventory and make the API refuse every publication. Tests that start an API-mode manager also left _AIRFLOW_PROCESS_CONTEXT set to "client" for the rest of the worker's tests.
An archive with one member whose path the Execution API cannot accept, such as a backslash in a name, was held back whole. Its other members' Dags stayed at their last publication with no import error, and a new archive never appeared in the UI. Standalone files with such a path are already skipped and logged by name, and DB mode still parses the other members of the archive, so only that member is left out now.
A Dag processor that had no result for a file held back the import error until its own last publication of the file outlasted stale_dag_threshold. That happened when the parse timed out, exited without a result or hit a failed Execution API request. With several Dag processors on a bundle, the eventual report deactivated Dags that another processor had published seconds earlier, and until then users saw no error. The API server now keeps such a file's Dags until they outlast stale_dag_threshold by their own last parse, as DB mode does after a timeout, so these reports can go out at once. A bundle whose listing raised got no retry time, so every bundle_refresh_check_interval re-ran bundle.refresh(), a git fetch for Git bundles, and listed it again. A listing the API rejects already waits for the bundle's refresh interval, and a failed listing should too.
A Dag processor that publishes through the Execution API parses nothing from a bundle until its listing succeeds. Waiting the bundle's whole refresh interval after a listing error meant one transient failure, such as a file removed during the walk, could hold up the bundle's parsing and its reparse requests for an hour. Listing again at every check instead would re-run the refresh, a git fetch for Git bundles, every few seconds for as long as the error lasts, so the wait grows from one check interval to the refresh interval. The bundle docs described only the database-mode behaviour. The refused and oversize failure reports had no test for result_unavailable, so a change that dropped the flag from them would have let those reports deactivate the file's Dags again unnoticed.
The deadline for retrying a failed listing was taken after the bundle refresh and the listing had run, but the periodic check it is compared with is timed from the check's start. Any refresh that took time, such as a git fetch, pushed the first retry past the next check, to about two check intervals after the failure instead of one. The bundle docs did not say that, through the Execution API, parsing also waits until the API server accepts the bundle's listing, or that a listing it refuses as too large holds parsing for the refresh interval. Which failed Execution API requests are worth resending was decided in three places, written three different ways, so one could change without the others. The tests also drop setup and a duplicate test that the removed failure-report hold-back left behind.
ephraimbuddy
force-pushed
the
dag-processor-http-manager
branch
from
November 6, 2026 09:07
ad44144 to
5c9861c
Compare
"resends" is not in the docs spelling wordlist, so the documentation build failed on the sentence about resending a bundle's listing.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Connect the internal manager to API bundle metadata, scoped secrets, inventories, and parse-result publication. Retain pending results across transient failures, recover lost acknowledgments, and apply bounded publication backpressure.
Layer 5 of the seven-part replacement for #74233, in GitHub native stack #74792. Depends on #74791. Base:
dag-processor-work-delivery.GitHub's Files changed tab shows this layer. Compare this layer.
Merge order:
Includes the two remaining compatibility commits from #74728 (JSON media types and the httpx import guard); its main client conversion is already merged via #74731.
Focused manager, processor, client, and bundle tests: 584 passed. Post-integration manager/bundle tests: 335 passed, 5 PostgreSQL-only skips. Regular/manual static checks passed.
Was generative AI tooling used to co-author this PR?
Generated-by: Codex (GPT-6) following the guidelines