diederikperdok commented on code in PR #76322:
URL: https://github.com/apache/airflow/pull/76322#discussion_r4453601376
##########
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:
It seems like a repeated cancellation would also be problematic for triggers
in `self.to_release`, if the call to `.cleanup()` for that trigger takes >1s.
Perhaps we need a more general flag "cancellation_started" instead of
"on_kill_started", applying to all cancellations?
##########
airflow-core/tests/unit/jobs/test_triggerer_job.py:
##########
@@ -1741,6 +1742,23 @@ async def
test_cancel_triggers_uses_distinct_messages_for_user_action_and_reassi
assert not trigger_runner.to_cancel
assert not trigger_runner.to_release
+ @pytest.mark.asyncio
+ async def test_cancel_triggers_is_idempotent_across_state_syncs(self) ->
None:
+ """A repeated cancel from the next state sync must not interrupt a
running on_kill()."""
+ trigger_runner = TriggerRunner()
+ task = MagicMock(spec=asyncio.Task)
+ trigger_runner.triggers = {1: {"task": task, "is_watcher": False,
"name": "user", "events": 0}}
+
+ trigger_runner.to_cancel.append(1)
+ await trigger_runner.cancel_triggers()
+ # run_trigger() marks the trigger as soon as it enters the on_kill()
branch.
+ trigger_runner.triggers[1]["on_kill_started"] = True
Review Comment:
Ideally, instead of setting `on_kill_started` inside the test, the test
would verify that the triggerer sets `on_kill_started`. Now the test itself
implements (part of) the logic it claims to test, if I understand correctly.
--
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]
