shahar1 commented on code in PR #68792: URL: https://github.com/apache/airflow/pull/68792#discussion_r4560690175
########## airflow-core/newsfragments/68792.bugfix.rst: ########## @@ -0,0 +1 @@ +Recover Triggerer triggers that the supervisor tracked as running but for which the runner had no live coroutine. Their task instances previously stayed in ``deferred`` indefinitely; the trigger is now re-created on the next sync so the task can resume. A new ``triggers.state_mismatch`` counter is emitted when this is detected. Review Comment: **nit** — [`contributing-docs/18_contribution_workflow.rst`](https://github.com/apache/airflow/blob/main/contributing-docs/18_contribution_workflow.rst) now says: > For ``airflow-core/``, use only ``significant`` newsfragments, and only for a change users must read before upgrading: [...] > > [...] Do not add a newsfragment for an ordinary bug fix, feature, improvement or documentation change, or for internal refactors, CI, tooling and tests. That rule landed on 2026-11-10, after this newsfragment was requested in review, so nothing to fault here — flagging it so whoever merges can decide whether to keep it or drop it. ########## airflow-core/src/airflow/jobs/triggerer_job_runner.py: ########## @@ -933,6 +937,24 @@ def clean_unused(self) -> None: with attempt, create_session() as session: Trigger.clean_unused(session=session) + def check_for_unhandled_triggers(self, running_ids: set[int] | None) -> None: + """ + Re-create triggers we track as running that the runner has no coroutine for. + + Only valid between finished-removal and to_create-addition in ``_handle_request``, where the + two sides agree. Dropping the leftovers lets :meth:`update_triggers` rebuild them. + """ + if running_ids is None: + return + unhandled = self.running_triggers - running_ids + if not unhandled: + return + log.error("Triggers have no coroutine in the runner; re-creating", trigger_ids=sorted(unhandled)) + self.running_triggers -= unhandled Review Comment: **minor** — the recovered ids are dropped from the three tracking sets, but their `logger_cache` entries stay. When `update_triggers` re-creates them, `build_trigger_workloads` overwrites `self.logger_cache[trigger.id]` (`:1101` / `:1137`) without calling `close()` or `upload_to_remote()`. If the lost trigger had logged anything, its file handle leaks, and the new `TriggerLoggingFactory` opens the same path (same TI try, same job id) with `"wb"`, truncating the lost run's log before it was ever uploaded — and that log is the one most useful for working out why the coroutine vanished. The `finished` branch in `_handle_request` (`:739-751`) already does pop + upload + close; doing the same for each unhandled id here would cover it, e.g.: ```python for trigger_id in unhandled: if factory := self.logger_cache.pop(trigger_id, None): try: factory.upload_to_remote() except Exception: log.exception("Failed to upload trigger logs to remote", trigger_id=trigger_id) finally: factory.close() ``` A small helper shared with the `finished` loop would avoid duplicating it, and a test asserting the factory is closed after recovery would pin the behaviour. -- 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]
