pankajastro commented on code in PR #36916:
URL: https://github.com/apache/airflow/pull/36916#discussion_r1463237047
##########
airflow/sensors/external_task.py:
##########
@@ -351,29 +348,30 @@ def execute(self, context: Context) -> None:
super().execute(context)
else:
self.defer(
- trigger=TaskStateTrigger(
- dag_id=self.external_dag_id,
- task_id=self.external_task_id,
+ timeout=self.execution_timeout,
+ trigger=WorkflowTrigger(
+ external_dag_id=self.external_dag_id,
+ external_task_ids=self.external_task_ids,
execution_dates=self._get_dttm_filter(context),
- states=self.allowed_states,
- trigger_start_time=utcnow(),
- poll_interval=self.poll_interval,
+ allowed_states=self.allowed_states,
+ poke_interval=self.poll_interval,
+ soft_fail=self.soft_fail,
),
method_name="execute_complete",
)
def execute_complete(self, context, event=None):
"""Execute when the trigger fires - return immediately."""
if event["status"] == "success":
- self.log.info("External task %s has executed successfully.",
self.external_task_id)
- return None
- elif event["status"] == "timeout":
- raise AirflowException("Dag was not started within 1 minute,
assuming fail.")
+ self.log.info("External tasks %s has executed successfully.",
self.external_task_ids)
else:
- raise AirflowException(
- "Error occurred while trying to retrieve task status. Please,
check the "
- "name of executed task and Dag."
- )
+ if self.soft_fail:
Review Comment:
If an exception occurs in the trigger code then this base implementation
takes care of soft fail. I have this check if handle if trigger has any event
without success then skip is it if soft_fail set instead of fail
--
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: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
