[hanumanraj07]

What

Catches TaskNotFound in the triggerer supervisor's _create_workload method so that one stale task-associated trigger cannot abort the entire batch of healthy trigger workloads.

Closes: #69841

Why

The triggerer supervisor loads the pinned serialized Dag for every task-associated trigger to check whether the task uses start_from_trigger. If the serialized Dag exists but no longer contains the deferred task, get_task() raises TaskNotFound. The exception escapes the per-batch loop in build_trigger_workloads, so one stale task instance prevents unrelated healthy triggers in the same batch from starting and can repeatedly restart the triggerer supervisor.

How

• Wrapped get_task() in _create_workload with a try/except TaskNotFound block
• When caught, logs a warning and sets task = None, falling through to the context-free RunTrigger return (matching behavior when serialized_dag_model is None)
• Changed task.start_from_trigger check to task is not None and task.start_from_trigger
• Added unit test verifying the fix

[steveahnahn]

Good catch, I verified the test fails on main and passes with the fix. Good use of TaskNotFound rather than a blanket except, LGTM.

[kaxil]

Closing this. #69988 made the same change on main: the triggerer wraps get_task() in try/except TaskNotFound and only treats the task as start_from_trigger when it was found.