FrankYang0529 commented on code in PR #72057:
URL: https://github.com/apache/airflow/pull/72057#discussion_r4172298119
##########
providers/edge3/src/airflow/providers/edge3/worker_api/routes/jobs.py:
##########
@@ -44,11 +45,19 @@
jobs_router = AirflowRouter(tags=["Jobs"], prefix="/jobs")
+# ``ExecuteCallback`` is also a valid Dag id, so checking ``dag_id`` alone
would mistake that Dag's
+# tasks for callbacks. ``EdgeExecutor.queue_workload()`` gives a callback this
whole identity.
+_IS_CALLBACK_JOB = and_(
+ EdgeJobModel.dag_id == EXECUTE_CALLBACK_TAG,
+ EdgeJobModel.run_id == literal(f"{EXECUTE_CALLBACK_TAG}-") +
EdgeJobModel.task_id,
+ EdgeJobModel.map_index == -1,
+ EdgeJobModel.try_number == 0,
+)
+
def parse_command(command: str, dag_id: str, run_id: str) -> ExecuteTypeBody:
if AIRFLOW_V_3_3_PLUS:
from airflow.executors.workloads import ExecuteCallback
- from airflow.providers.edge3.models.types import EXECUTE_CALLBACK_TAG
if dag_id == EXECUTE_CALLBACK_TAG and
run_id.startswith(EXECUTE_CALLBACK_TAG):
Review Comment:
`parse_command()` now calls `is_callback_job()` with the job's full
identity, so the parser and `fetch()` agree on what a callback is.
--
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]
