Repository: apache/airflow
File: airflow-core/src/airflow/jobs/triggerer_job_runner.py (Python)

Code before:
                        "Trigger %s exited with error %s", details["name"], e, trigger_id=trigger_id
                    )
                    saved_exc = e
                else:
                    # See if they foolishly returned a TriggerEvent
                    if isinstance(result, TriggerEvent):
                        self.log.error(
                            "Trigger returned a TriggerEvent rather than yielding it",
                            trigger=details["name"],
                            trigger_id=trigger_id,
                        )
                # See if this exited without sending an event, in which case
                # any task instances depending on it need to be failed
                if details["events"] == 0:
                    self.log.error(
                        "Trigger exited without sending an event. Dependent tasks will be failed.",
                        name=details["name"],
                        trigger_id=trigger_id,
                    )
                    # TODO: better formatting of the exception?
                    self.failed_triggers.append((trigger_id, saved_exc))
                del self.triggers[trigger_id]
            await asyncio.sleep(0)
        return finished_ids

    def validate_state_changes(self, finished_ids: list[int]) -> messages.TriggerStateChanges:
        # Copy out of our dequeues in threadsafe manner to sync state with parent

        req_encoder = _new_encoder()
        events_to_send: list[tuple[int, DiscrimatedTriggerEvent]] = []
        failures_to_send: list[tuple[int, list[str] | None]] = []

        while self.events:
            trigger_id, trigger_event = self.events.popleft()

            try:
                req_encoder.encode(trigger_event)
                events_to_send.append((trigger_id, trigger_event))
            except NotImplementedError as e:
                logger.error(
                    "Trigger %s returned non-serializable result %r. Cancelling trigger.",
                    trigger_id,
                    trigger_event,
                )
                self.failed_triggers.append((trigger_id, e))

        while self.failed_triggers:
            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))


Review comment (suggestion):
This method name is a bit off -- given it returns the state changes, I'd suggest calling it something like process_ or serialize_ etc.

Code after:
                    # If we don't, then the system requesting a trigger be removed -
                    # which turns into CancelledError - results in a failure.
                    del self.triggers[trigger_id]
                    continue
                except BaseException as e:
                    # This is potentially bad, so log it.
                    self.log.exception(
                        "Trigger %s exited with error %s", details["name"], e, trigger_id=trigger_id
                    )
                    saved_exc = e
                else:
                    # See if they foolishly returned a TriggerEvent
                    if isinstance(result, TriggerEvent):
                        self.log.error(
                            "Trigger returned a TriggerEvent rather than yielding it",
                            trigger=details["name"],
                            trigger_id=trigger_id,
                        )
                # See if this exited without sending an event, in which case
                # any task instances depending on it need to be failed
                if details["events"] == 0:
                    self.log.error(
                        "Trigger exited without sending an event. Dependent tasks will be failed.",
                        name=details["name"],
                        trigger_id=trigger_id,
                    )
                    # TODO: better formatting of the exception?
                    self.failed_triggers.append((trigger_id, saved_exc))
                del self.triggers[trigger_id]
            await asyncio.sleep(0)
        return finished_ids

    def process_trigger_events(self, finished_ids: list[int]) -> messages.TriggerStateChanges:
        # Copy out of our dequeues in threadsafe manner to sync state with parent
        events_to_send: list[tuple[int, DiscrimatedTriggerEvent]] = []
        failures_to_send: list[tuple[int, list[str] | None]] = []

        while self.events:
            trigger_id, trigger_event = self.events.popleft()
            events_to_send.append((trigger_id, trigger_event))

        while self.failed_triggers:
            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))

        return messages.TriggerStateChanges(
            events=events_to_send if events_to_send else None,
            finished=finished_ids if finished_ids else None,
            failures=failures_to_send if failures_to_send else None,
        )