amoghrajesh commented on code in PR #73135:
URL: https://github.com/apache/airflow/pull/73135#discussion_r4056583820
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1410,6 +1436,17 @@ async def create_triggers(self):
trigger_instance.triggerer_job_id = self.job_id
trigger_instance.timeout_after = workload.timeout_after
+ if (ti := workload.ti) is not None:
Review Comment:
Already checked on line 1391?
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1410,6 +1436,17 @@ async def create_triggers(self):
trigger_instance.triggerer_job_id = self.job_id
trigger_instance.timeout_after = workload.timeout_after
+ if (ti := workload.ti) is not None:
+ trigger_instance.task_state_store = TaskStateStoreAccessor(
+ ti_id=ti.id,
+ scope=TaskScope(
+ dag_id=ti.dag_id,
+ run_id=ti.run_id,
+ task_id=ti.task_id,
+ map_index=ti.map_index,
Review Comment:
The worker guards this: `map_index=self.map_index if self.map_index is not
None else -1`and here there is no guard.
The schema allows `None` and `TaskScope.map_index` is typed int with no
runtime check, so a `None` would land in the scope silently. It cannot happen
today, since the DTO comes from a non-nullable DB column. But the two sites
should match, and a wrong scope would only show up as a custom backend writing
to a different path.
While you are here: `_ti_dto()` only ever uses -1 and 3, so no test covers
this. Parametrizing over -1, 3, None would lock it in.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -124,10 +132,12 @@
handle_put_variable,
handle_set_asset_state_store_by_name,
handle_set_asset_state_store_by_uri,
+ handle_set_task_state_store,
handle_set_xcom,
)
from airflow.sdk.execution_time.supervisor import WatchedSubprocess,
make_buffered_socket_reader
from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
+from airflow.sdk.state import TaskScope
Review Comment:
Can yuo add this line into: `generated/known_sdk_imports_in_core.txt`to fix
the static checks in CI?
##########
airflow-core/docs/core-concepts/task-state-store.rst:
##########
@@ -152,6 +152,13 @@ Async counterparts of ``get``, ``set``, ``delete``, and
``clear`` for use inside
Calling the synchronous ``get``/``set``/``delete``/``clear`` from inside an
``async`` task blocks the event loop and defeats the concurrency the coroutine
was written for. Use the ``a``-prefixed methods there instead.
+Using ``task_state_store`` inside a Trigger
+-------------------------------------------
+
+A trigger belonging to a deferred task can read and write that task's state
store from within ``run()``. The triggerer injects ``self.task_state_store``
before ``run()`` is called, scoped to the task instance that deferred, which is
the same namespace the operator's ``execute()`` and ``execute_complete()`` see.
It is not available during ``__init__`` or ``serialize()``, only from within
``run()``.
Review Comment:
"the same namespace the operator's execute() and execute_complete() see" is
only true for the default DB store. It breaks when `[workers]
state_store_backend` is set.
The accessor reads that config in whatever process it runs in. Many
deployments set `AIRFLOW__WORKERS__STATE_STORE_BACKEND` on worker pods only.
Then the triggerer has no backend.
The bad direction is that the operator writes through the backend, and the
trigger reads. `_extract_get_response` sees backend is None, skips the unwrap,
and returns `{"__external_ref__": "..."}` which is the marker dict, not the
value. No warning fires.
Please say in the docs that `[workers] state_store_backend` has to be set
the same way on the triggerer.
--
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]
