Andrushika commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3610810954
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -622,6 +646,95 @@ def _validate_outlet_event_partition_keys(outlet_events:
list[dict[str, Any]]) -
)
+# Lock timeout (seconds) for the opportunistic synchronous asset-event write.
Kept short so a
+# task completion sheds to the durable queue quickly under asset-row
contention rather than
+# holding an API-server worker. This only bounds lock waits; an uncontended
write does not wait
+# on locks and so is unaffected by the timeout.
+_ASSET_SYNC_LOCK_TIMEOUT = 2
+
+
+def _register_asset_changes_or_enqueue(
Review Comment:
Wow this direction looks really clean!
But I traced the code, and I have one concern: if we skip the inline
registration, registration only happens in `_drain_asset_event_queue`, which
runs on the scheduler's main loop thread (am I reading this right?). So all the
registration work would move onto the scheduler loop, and registration can be a
heavy job. Every event would also wait for the next drain tick, even when there
is no contention.
So I think there are two options:
1. Keep the inline registration as the common path, and use the scheduler
drain as a fallback.
2. Find another place to run the registration jobs? But I don't know where
is appropriate.
--
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]
