potiuk commented on code in PR #68792:
URL: https://github.com/apache/airflow/pull/68792#discussion_r3680753521
##########
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:
This exits the supervisor loop via `should_stop()`, terminating the
Triggerer and every trigger it is running. The tasks get picked up by another
Triggerer eventually, so it's recoverable — but it's a disruptive,
whole-process response to what may be a transient accounting difference.
The docstring concedes the invariant is only valid "between finished-removal
and to_create-addition", which is a timing-dependent window: `num_running` is
computed in the subprocess when it builds the message, and compared against
supervisor state some time later. I'd want to see the argument for why no
legitimate interleaving can produce a one-off difference before wiring it to a
shutdown.
Softer options that still surface the bug: log at ERROR and emit a metric,
or require N consecutive mismatches before stopping so a single transient blip
doesn't cost an outage. If shutdown really is the right call, a comment
explaining why the window is airtight would help the next reader.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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(
Review Comment:
Uses the module-level `log`, but `_handle_request` already receives a bound
`log: FilteringBoundLogger` that carries the request context. Passing that
through would keep this message correlated with the rest of the request's
logging.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1399,11 +1417,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_ids.append(trigger_id)
Review Comment:
Mutating the caller's list in place. I traced it and it's harmless today —
`finished_ids` is freshly built in `cleanup_finished_triggers()` and isn't read
again after `sync_state_to_supervisor` returns — so this is style, not a bug.
Still, `process_trigger_events` reads as a pure "build the message"
function, and a caller that later reused its list would get a surprise.
Building the combined list locally and leaving the parameter untouched keeps
that property.
---
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]
