potiuk commented on code in PR #68792:
URL: https://github.com/apache/airflow/pull/68792#discussion_r3962286373
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1506,11 +1528,14 @@ def process_trigger_events(self, finished_ids:
list[int]) -> messages.TriggerSta
trigger_id, exc = self.failed_triggers.popleft()
tb = format_exception(type(exc), exc, exc.__traceback__) if exc
else None
failures_to_send.append((trigger_id, tb))
+ if trigger_id not in self.triggers:
+ finished_to_send.append(trigger_id)
Review Comment:
This double-counts on the ordinary failure path, so please fix it here
rather than leaving it for later. `cleanup_finished_triggers` appends
`trigger_id` to `finished_ids` at the top of the loop, and then — when
`details["events"] == 0` — also appends to `self.failed_triggers` before `del
self.triggers[trigger_id]`. So a trigger that errors out or returns without
yielding an event arrives here already in `finished_to_send`, is no longer in
`self.triggers`, and gets appended a second time.
Benign today: the supervisor's `running_triggers.discard(id)` is idempotent
and `logger_cache.pop(id, None)` returns `None` the second time. But `finished`
now routinely carries duplicates where it didn't before, and anything that
later counts it rather than iterating it would be wrong.
```suggestion
if trigger_id not in self.triggers and trigger_id not in
finished_to_send:
finished_to_send.append(trigger_id)
```
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
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]
