hrithiksaini99 commented on code in PR #76322:
URL: https://github.com/apache/airflow/pull/76322#discussion_r4454721470
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1749,8 +1762,13 @@ async def cancel_triggers(self):
):
while queue:
trigger_id = queue.popleft()
- if trigger_id in self.triggers:
- self.triggers[trigger_id]["task"].cancel(cancel_msg)
+ details = self.triggers.get(trigger_id)
+ if details is None or details.get("on_kill_started"):
+ # The supervisor resends every trigger it is still waiting
on once per state
+ # sync (~1s). Once on_kill() is running, another cancel
would interrupt it
+ # before ``[triggerer] on_kill_timeout`` expires.
+ continue
+ details["task"].cancel(cancel_msg)
await asyncio.sleep(0)
Review Comment:
Great point! Generalized to `cancellation_started`, set directly inside
`cancel_triggers()` when the initial cancel is dispatched. This cleanly
protects both `self.to_cancel` (user action -> `on_kill()`) and
`self.to_release` (reassignment -> `cleanup()`) from repeated cancel signals
across state syncs while the asynchronous handlers are in flight.
--
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]
