jason810496 commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3610203618
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3676,6 +3682,88 @@ def _remove_unreferenced_triggers(self, *, session:
Session = NEW_SESSION) -> No
.execution_options(synchronize_session="fetch")
)
+ @provide_session
+ def _drain_asset_event_queue(self, *, session: Session = NEW_SESSION) ->
None:
+ """
+ Process pending asset-event registrations enqueued by the execution
API.
+
+ The task-success path enqueues an
:class:`~airflow.models.asset.AssetEventQueue` row when
+ synchronous asset-event registration is skipped or fails under
database lock contention.
+ Here we claim pending rows one at a time (``FOR UPDATE SKIP LOCKED``,
so multiple schedulers
+ do not collide), (re)run ``register_asset_changes_in_db`` to create
the ``AssetEvent`` and
+ ``AssetDagRunQueue`` rows, and delete the queue row in the same
transaction so it is
+ processed exactly once. A row that keeps failing is retried on later
passes until it exceeds
+ ``asset_event_queue_max_attempts``, after which it is left in place
(parked) and surfaced via
+ the ``asset.event_queue.failures`` metric for an operator to inspect.
+ """
+ from airflow.api_fastapi.execution_api.datamodels.asset import
AssetProfile
+
+ batch_size = conf.getint("scheduler", "asset_event_queue_batch_size",
fallback=100)
+ max_attempts = conf.getint("scheduler",
"asset_event_queue_max_attempts", fallback=5)
Review Comment:
The config should be set at class attribute level so that we won't retrieve
the conf every time when the `_drain_asset_event_queue` tick (get `conf` might
access the secret backend, we should cache the value on scheduler init stage
instead of re-compute).
--
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]
