JH0917 commented on code in PR #68792:
URL: https://github.com/apache/airflow/pull/68792#discussion_r3698277600
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -720,6 +723,21 @@ def clean_unused(self) -> None:
"""Remove triggers that are no longer needed."""
Trigger.clean_unused()
+ def check_for_unhandled_triggers(self, num_running: int) -> None:
+ """
+ Shut down if the subprocess trigger count disagrees with the
supervisor.
+
+ Only valid between finished-removal and to_create-addition in
``_handle_request``.
+ """
+ expected = len(self.running_triggers)
+ if expected != num_running:
+ log.error(
+ "Trigger count mismatch: expected %d, subprocess reports %d.
Shutting down.",
+ expected,
+ num_running,
+ )
+ self.stop = True
Review Comment:
Thanks -- the shutdown is gone. The check now recovers instead: dropping the
stranded ids from running_triggers / cancelling_triggers makes them fall out of
known_trigger_ids in update_triggers, so the next loop rebuilds their workloads
and nothing else running is disturbed. I added the ERROR log and the
triggers.state_mismatch counter you suggested.
On the timing window: both sides only change at the message boundary, and
the only code that adds a trigger runs in the runner's main loop -- which is
blocked waiting for the reply while the supervisor compares. So the snapshot
can't go stale and one disagreement is enough. That's why I skipped the
N-consecutive debounce.
Drafted-by: Claude Code (Opus 5); reviewed by @JH0917 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]
