kaxil commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4483605504
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -257,8 +261,11 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
)
if msg.callback_requests:
+ targets_loaded = any(_is_callback_target_loaded(bag, request) for
request in msg.callback_requests)
# If the request is for callback, we shouldn't serialize the Dags
_execute_callbacks(bag, msg.callback_requests, log)
+ if not targets_loaded:
+ sys.exit(_CALLBACK_TARGETS_MISSING_EXIT_CODE)
Review Comment:
This exit also happens in DB mode, and in API mode whenever a callback's Dag
was simply removed. The supervisor logs any non-zero exit as `Process exited
abnormally` with `exit_code=75` (`supervisor.py:1503`), where these processes
used to exit 0. Could 75 skip that warning, or only be used when the processor
runs in API mode?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -2169,8 +2245,23 @@ def _collect_results(self):
for file in finished:
processor = self._processors.pop(file)
+ if processor.had_callbacks:
+ self._finish_callback_attempt(processor)
processor.close()
+ def _finish_callback_attempt(self, processor: BaseDagFileProcessorProcess)
-> None:
+ # Running a callback again repeats its side effects, so retry only a
process that ran none of them.
+ if processor.api_request_error is None or not
processor.callback_targets_missing:
Review Comment:
Nothing tests `api_request_error is None` with `callback_targets_missing`
true, which is the removed-Dag case that should ack right away. The recovery
tests cover an API error with missing targets and no error with loaded targets,
but not this one. Could the two new recovery tests become one parametrized
table over both values?
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -557,14 +587,165 @@ parsing subprocess can without this client.
``Variable.set`` upserts on the key
multi-team mode a write can overwrite another team's Variable with the same
key, as it can with task
tokens.
-The client attempts renewal at 80% of token lifetime. Transient early-renewal
failures leave
-the existing token usable until expiry. A retired registration signals that
its owner must
-restart. Completion retains its original identity and outcome across retries.
-
-File and attempt identity are carried in the token but are not currently used
for access
-checks or request attribution. Archives retain one file identity under the
current processor
-model. These credentials do not isolate hostile code from other credentials
accessible on
-the processor host; the process and deployment limitations above still apply.
+For each ``dag_parse`` token, the manager chooses the bundle, bundle-relative
file location,
+and parsing-attempt UUID.
+
+The manager publishes completed imports through ``POST
/jobs/{job_id}/parse-results``
+using its Job credential. Parsing tokens cannot publish. A request contains
one file or
+container's serialized Dags, diagnostics, source text (or an explicit
unavailable marker),
+and the bundle version captured at dispatch. The API never reads the
processor's source
+paths. It checks bundle access, existing Dag ownership and team-scoped plugin
classes
+before deserializing results and rechecks ownership on the rows used for
persistence.
+A Dag can move between files within its bundle. Recovery from a removed bundle
is
+limited to stale legacy rows with neither a relative file location nor a Dag
version.
+
+Each request carries an attempt UUID and a dispatch sequence within its Job. A
publication
+receipt and the metadata writes commit together. Replaying the latest
identical request returns
+the saved receipt; conflicting payloads and superseded sequences are rejected.
Publication
+receipts retain one entry per Job and source and are removed with Job cleanup.
There is no total
+ordering between different Jobs publishing a source without versioning; an
already-accepted
+request's retry does not overwrite another Job's result. Listener side effects
are not
+transactional and may repeat after a failed transaction.
+
+Publication and inventory request bodies are limited to 16 MiB, and the API
refuses a
+larger body before reading the rest of it. A rejected publication does not
fall back
+to direct database writes, and does not prevent other files from being
processed. Transport
+failures and server errors are retried; exhausting those retries preserves the
last accepted
+result. When the API refuses a result for its content, or the result is over
the limit, the
+processor publishes an import error for the file instead, so the problem is
visible.
+Dags whose definition path matches an import error are marked stale until a
later parse
+publishes them again. Dags missing from an accepted parse are marked stale at
once: the
+publication is atomic, so there is no gap between file and Dag parse times to
wait out. For a
+container-level error, missing member Dags are marked stale only when their
last successful
+parse is older than ``[dag_processor] stale_dag_threshold``, using the
server's clock. Invalid ``access_control`` produces an import error with both
+current and supported older FAB releases. Permission backend failures remain
retryable
+server errors. Empty imports still publish a receipt and clear the file's
earlier import
+errors. A bundle refresh, including one forced by a parse request, waits for
that bundle's
+active imports instead of discarding them; only initializing a bundle again
invalidates and
+requeues older in-flight results from that manager.
+
+Inventories report discovered definitions and sources whose discovery failed
separately.
+Failed sources preserve their existing definitions and diagnostics, including
archive members;
+other sources still undergo deletion reconciliation. Each inventory carries
the previously
+observed server revision; conflicting refreshes must
+rediscover before retrying. Imports carry the accepted revision, so an older
snapshot cannot
+publish after a changed inventory. Only accepted publications establish
missing Dags within
+a file; rejected requests and incomplete discovery do not establish absence.
Every
+``[scheduler] parsing_cleanup_interval``, the scheduler marks the Dags of
inactive bundles stale,
+removes stale warnings, and ends expired processor Jobs, in bounded batches
and in
+both modes.
+
+Callback and priority requests remain stored while claimed. A lost claim
response or
+acknowledgment can be retried. Claims use increasing sequences per Job and
work type; the
+client waits for a claim response before sending the next sequence for that
type.
+Retrying an empty claim remains empty, even if work arrives
+afterward, and retrying an older claim cannot consume newly requested work.
Requests from a
+retired Job become available to another processor. A callback acknowledgment
confirms
+delivery, not successful execution of user code. As with other at-least-once
delivery, a crash
+after a callback's side effect and before acknowledgment can repeat that side
effect. Before the
+processor's Job ends, a graceful shutdown acknowledges callbacks whose
processes have already exited
+cleanly; callbacks that are still running or that the shutdown signal stopped
are delivered again.
+When a callback's bundle
Review Comment:
This still names bundle preparation as the only case that is retried and
then acknowledged without running. Could it also cover callbacks retried after
an Execution API failure, and that only a process where none of the file's
callbacks found its Dag or task is retried?
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -257,8 +261,11 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
)
if msg.callback_requests:
+ targets_loaded = any(_is_callback_target_loaded(bag, request) for
request in msg.callback_requests)
Review Comment:
With `any()`, a batch where one Dag in the file loaded and another didn't
exits 0, and the manager acknowledges both claims. So a static Dag next to Dags
generated from `Variable.get(..., default=[])` still loses the generated Dags'
callbacks on a gateway 503, while the `job_heartbeat_timeout` description in
`config.yml` says a callback whose Dag or task did not load is retried. If
per-request outcomes are out of scope, could the config text say the retry only
happens when none of the file's callbacks found its target, with a mixed-batch
case in `test_parse_file_exits_without_running_callbacks_when_no_target_loaded`?
--
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]
