why Dag's on_callback_failure have first taskinstance in dagrun? #24029

eprj453 · 2022-05-30T16:16:18Z

eprj453
May 30, 2022

if i set on_callback_failure in dag and task failed, handle_failure is activated in airflow.models.dags.py.

@provide_session
    def handle_callback(self, dagrun, success=True, reason=None, session=None):
        """
        Triggers the appropriate callback depending on the value of success, namely the
        on_failure_callback or on_success_callback. This method gets the context of a
        single TaskInstance part of this DagRun and passes that to the callable along
        with a 'reason', primarily to differentiate DagRun failures.

        .. note: The logs end up in
            ``$AIRFLOW_HOME/logs/scheduler/latest/PROJECT/DAG_FILE.py.log``



        :param dagrun: DagRun object
        :param success: Flag to specify if failure or success callback should be called
        :param reason: Completion reason
        :param session: Database session
        """

        logger = logging.getLogger("my_logger")
        logger.setLevel(logging.INFO)
        file_handler = logging.FileHandler('/Users/psw/Desktop/repository/dev/airflow2test/logs/TEST/error.log')
        logger.addHandler(file_handler)
        logging.info("dag.py -> handle_callback")

        callback = self.on_success_callback if success else self.on_failure_callback
        if callback:
            self.log.info('Executing dag callback function: %s', callback)
            tis = dagrun.get_task_instances(session=session)
            ti = tis[-1]  # get first TaskInstance of DagRun
            ti.task = self.get_task(ti.task_id)
            context = ti.get_template_context(session=session)
            context.update({'reason': reason})
            try:
                callback(context)
            except Exception:
                self.log.exception("failed to invoke dag state update callback")
                Stats.incr("dag.callback_exceptions")

in this code, callback method is activated with context having first taskinstance not error raised taskinstance. then, is there no way to get error task information unless assigned on_callback_failure in every task?

potiuk · 2022-06-04T21:13:52Z

potiuk
Jun 4, 2022
Collaborator

I completely don't understand what you are asking for . Please be more precise if you would like some help with your problem.

5 replies

MaciejGajewski Jun 28, 2022

@potiuk @eprj453 I've run into the same issue. Basically following the example Callbacks on how to use on_failure_callback at the Dag level one would expect that if the task2 fails context object passed to task_failure_alert function would contain reference to this specific task (in context['task_instance_key_str']).
However this doesn't seem to be the case. Based on the source code provided above first task is always passed:
ti = tis[-1] # get first TaskInstance of DagRun
ti.task = self.get_task(ti.task_id)

In my production code I'm observing a little random behavior - it's first or last task from my Dag definition but not the one which actually failed.

I think this should be treated as a bug or documentation should be corrected to clarify this behavior and enhancement created to return task instance which really failed.

potiuk Jun 28, 2022
Collaborator

@MaciejGajewski - your description is much more understandable. Can you please create an issue describin it or (even better) maybe even PR attempting to fix it?

eprj453 Jun 28, 2022
Author

@potiuk
simply,
on_failure_callback is return DAG instance not TASK instance, so can't find what task raise error in whole task in DAG.
i think it should return Task instance so i'm refactoring & testing.

i'm not good at english so i wanna ask specific reply or PR @MaciejGajewski if have better answer.

thanks!

aitmrza Jan 25, 2023

Hello @potiuk @eprj453, has this problem been solved and in which version of airflow?

josh-fell Jan 25, 2023
Collaborator

Related issue: #26760

Asquator · 2026-02-07T12:11:17Z

Asquator
Feb 7, 2026

Opened a PR to pass a more relevant task to Dag callbacks: #61274

1 reply

Asquator Feb 7, 2026

I support the idea that dag callbacks should not receive any TI at all. It's weird and poorly defined. Maybe it should remain for back comp and that's what I did in the PR while improving the relevancy of the task passed. Anyway, it will be nice to add a CallbackMeta structure and inject it into the context to be able to differentiate between task callbacks and dag callbacks in case the same callback function is used in multiple places. I described the desired behavior here: #61119

why Dag's on_callback_failure have first taskinstance in dagrun? #24029

Uh oh!

Uh oh!

eprj453 May 30, 2022

Replies: 2 comments · 6 replies

Uh oh!

potiuk Jun 4, 2022 Collaborator

Uh oh!

Uh oh!

MaciejGajewski Jun 28, 2022

Uh oh!

potiuk Jun 28, 2022 Collaborator

Uh oh!

eprj453 Jun 28, 2022 Author

Uh oh!

aitmrza Jan 25, 2023

Uh oh!

josh-fell Jan 25, 2023 Collaborator

Uh oh!

Asquator Feb 7, 2026

Uh oh!

Uh oh!

Asquator Feb 7, 2026

eprj453
May 30, 2022

Replies: 2 comments 6 replies

potiuk
Jun 4, 2022
Collaborator

potiuk Jun 28, 2022
Collaborator

eprj453 Jun 28, 2022
Author

josh-fell Jan 25, 2023
Collaborator

Asquator
Feb 7, 2026