Returning `TriggerEvent`'s on exception in `BaseEventTrigger` children

[jroachgolf84]

Apache Airflow version

main (development)

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I'm working on adding the PubsubPullTrigger as a supported queue to be used by the MessageQueueTrigger for event-driven scheduling. However, in the code snippet below, a TriggerEvent is yielded when an exception occurs in the run method. This actually triggers a DAG run when used for Asset-driven scheduling. In my case, the exception was a missing project_id when instantiating the Trigger. This triggered DAGs runs over and over until I provided a project_id.

airflow/providers/google/src/airflow/providers/google/cloud/triggers/pubsub.py

Lines 88 to 108 in 819db98

	async def run(self) -> AsyncIterator[TriggerEvent]:
	try:
	while True:
	if pulled_messages := await self.hook.pull(
	project_id=self.project_id,
	subscription=self.subscription,
	max_messages=self.max_messages,
	return_immediately=True,
	):
	if self.ack_messages:
	await self.message_acknowledgement(pulled_messages)
	messages_json = [ReceivedMessage.to_dict(m) for m in pulled_messages]
	yield TriggerEvent({"status": "success", "message": messages_json})
	return
	self.log.info("Sleeping for %s seconds.", self.poke_interval)
	await asyncio.sleep(self.poke_interval)
	except Exception as e:
	yield TriggerEvent({"status": "error", "message": str(e)})
	return

When a TriggerEvent is not yielded, the Triggerer throws an exception that says something along the lines of "TriggerEvent was not returned".

Not really a bug, more of a question around weird behavior.

What you think should happen instead?

When a child of the BaseEventTrigger class returns a TriggerEvent with "status": "error", an AssetEvent should not be created.

How to reproduce

Please see the existing PR for details around the changes that are needed to reproduce this.

#54684

Operating System

MacOS

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[dejii]

I believe triggers should raise exceptions rather than catch and yield error events. This allows the triggerer framework to handle failures appropriately for different use cases:

• Deferred tasks: Task fails gracefully raising a TaskDeferralError
  

  airflow/task-sdk/src/airflow/sdk/bases/operator.py

  Lines 1622 to 1629 in 3a93f9b

  	if next_method == TRIGGER_FAIL_REPR:
  	next_kwargs = next_kwargs or {}
  	traceback = next_kwargs.get("traceback")
  	if traceback is not None:
  	self.log.error("Trigger failed:\n%s", "\n".join(traceback))
  	if (error := next_kwargs.get("error", "Unknown")) == TriggerFailureReason.TRIGGER_TIMEOUT:
  	raise TaskDeferralTimeout(error)
  	raise TaskDeferralError(error)

  
  

  airflow/airflow-core/src/airflow/models/trigger.py

  Lines 289 to 298 in 3a93f9b

  	# Add the error and set the next_method to the fail state
  	if isinstance(exc, BaseException):
  	traceback = format_exception(type(exc), exc, exc.__traceback__)
  	else:
  	traceback = exc
  	task_instance.next_method = TRIGGER_FAIL_REPR
  	task_instance.next_kwargs = {
  	"error": TriggerFailureReason.TRIGGER_FAILURE,
  	"traceback": traceback,
  	}

• Asset watchers: Trigger stops without creating asset events or scheduling DAG runs.

Same pattern of not yielding TriggerEvent can be observed in SqsSensorTrigger and AwaitMessageTrigger, which are also used as message queues. I've fixed it for PubsubPullTrigger in #54494

When a child of the BaseEventTrigger class returns a TriggerEvent with "status": "error", an AssetEvent should not be created

To achieve this, I think we would need a standardized message format for TriggerEvent, which currently doesn’t exist. Some triggers don’t include a status key in the event dictionary at all, and among those that do, the values are inconsistent—some use error, others use failed