clear_task_instances function not clearing tasks

[nicklip]

Apache Airflow version

2.5.2

What happened

The callback function called the clear_task_instances function and it failed to clear the task that was passed to it

What you think should happen instead

clear_task_instances should have cleared the task passed to it but was unable to. Instead the state of that task was set to 'failed' in the task_instance table in the database and never cleared or restarted.

How to reproduce

Run the following DAG and see that failing_task only fails and never gets cleared or restarted / retried.
Look at the XCOM list and you will see that the task object itself was in running state before the clear_task_instances call and restarting state after the call. However, when monitoring the task_instance table in the database while running this task, the task's state only goes from running to failed and never actually gets set to restarting. In version 2.3.1, running this code works as expected and the task state in the task_instance table goes from running to failed to null, successfully clearing it.

@provide_session
def clear_tasks_fn(tis, session=NEW_SESSION, dag=None):

    sleep_time = random.uniform(30.0, 60.0)  # sleep between 30 seconds and 1 minute
    sleep(sleep_time)
    for ti in tis:
        ti.xcom_push(key="before_state", value=ti.state)
    clear_task_instances(tis=tis, session=session, dag=dag)
    for ti in tis:
        session.commit()
        ti.xcom_push(key="after_state", value=ti.state)


def clear_tasks_callback(context):

    all_tasks = context["dag_run"].get_task_instances()
    dag_inner = context["dag"]
    task_ids_to_clear = context["params"].get("task_ids_to_clear", [])

    tasks_to_clear = [ti for ti in all_tasks if ti.task_id in task_ids_to_clear]

    clear_tasks_fn(tasks_to_clear, dag=dag_inner)


def failing_task_fn():

    raise Exception("Forcing task to fail")


def succeeding_task_fn():

    sleep_time = random.uniform(10.0, 20.0)  # sleep between 10 and 20 seconds
    sleep(sleep_time)
    return


default_args = {"owner": "revenge",
                "start_date": datetime(2022, 10, 1),
                }
schedule = "@once"

the_dag = DAG("clear_task_test", schedule_interval=schedule, default_args=default_args,
               catchup=False, is_paused_upon_creation=True)

with the_dag:

    failing_task_t = PythonOperator(
        task_id="failing_task",
        python_callable=failing_task_fn,
        params={"task_ids_to_clear": ["failing_task"]},
        on_failure_callback=clear_tasks_callback
    )

    succeeding_task_t = PythonOperator(
        task_id="succeeding_task",
        python_callable=succeeding_task_fn,
    )

failing_task_t.set_downstream(succeeding_task_t)

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[nicklip]

Sorry about the presentation of my DAG code. I couldn't figure out how to post it as an entire code block

[notatallshaw-gts]

Enclose everything with triple backtick, use the first line to tell GitHub what syntax highlight to do, e.g.

```python
def foo(bar, baz):
    raise ValueError
```

Gets translated to:

def foo(bar, baz):
    raise ValueError

[nicklip]

@notatallshaw-gts Thanks! Fixed it

[potiuk]

I think you are doing something that is somewhat not necessarily supported from Airflow point of view (hence converting it to a discussion). Making it an airflow issue is quite too much. I think rather than posting a code you should rather explain what your loigical intentions are. From what I uderstand you want to clear task from a failure callback - and this has absolutely no guarantees of succeedding, I think - this looks a bit as an abuse of the mechanism - and there are various reasons why it might not work from within the callback - calllback s (I think) is not really foreseen for cases where you would like to clear task that has just failed. That sounds what you try to do.

You have retry mechanisms foreseen for retrying failed tasks and coupled with cluster_policies it sounds like much better way of what you are trying to achieve. I think (but without of your logical explanation what you try to achieve) it is my guess.

[nicklip]

What I've posted is a simplified version of what I'm actually trying to do. Which is, if one particular task fails, then that task and the 2 tasks that precede it need to get cleared out and reran. Multiple sources online state that using clear_task_instances is the way to achieve this.

What is really throwing me off is that the task is never put into a failed state but still the on_failure_callback function gets called. This is a change from version 3.2.1, and could also possibly be the cause of my issue

[derrickrose]

Hello @nicklip ,

Any news about this issue? Is there a way you did to clear up tasks anyway?
Thank you!!

[laazy]

hello, @nicklip:
Did you solve this or have another way to achieve your goal? I'm getting in same trouble, it will be many helpful if let my know your solution.
Thanks~