Clearing a defered task instance in deferred state does not clear its next_method

[Greetlist]

Apache Airflow version

2.2.1

Operating System

Ubuntu 16.04

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

What happened

When I mark a defferd task to Failed-State then clear it or clear a defferd task which in Defferd-State, this task is marked success immediately.
I check the task instance table, this task's next_method is not null when I clear this task.This cause task's execute function is replaced by next_method function.

What you expected to happen

When I clear a defferd task, I want to re-judge it's trigger condition, re-execute the execute function.

How to reproduce

Sensor

class ExternalTaskFromOtherAirflowSensor(BaseSensorOperator):
    def execute(self, context):
        trigger = ExternalTaskFromOtherAirflowTrigger(args)
        self.defer(trigger=trigger, method_name="execute_complete")

    def execute_complete(self, context, event=None):
        return

Trigger

class ExternalTaskFromOtherAirflowTrigger(BaseTrigger):
    def _get_task_instance_state(self):
        sql_query = "SELECT state FROM {}.task_instance WHERE dag_id = '{}' and run_id like '%{}%';".format(
            self.old_airflow_db_name,
            self.old_external_dag_id,
            self.old_external_task_id,
            self.old_external_execution_date_str
        )
    async def run(self):
        while self._get_task_instance_state() == False:
            await asyncio.sleep(5)
        yield TriggerEvent('success')

1. Clear this task when it is in defferd State.
2. Mark this task as failed and clear this task.

Both operations above will cause this task is marked as success instead of re-execute the execute function.

Anything else

Similar issue:
#18146
#19120

I try to add this code in models/taskinstance.py clear_task_instances function

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]

Thanks for opening your first issue here! Be sure to follow the issue template!

[uranusjr]

Your added code makes sense. Feel free to start a pull request for this.

[malthe]

We can close this now then right?

[jochenott]

I believe this can be closed, as the according code change was made via #23929

[uranusjr]

Thanks for catching this

[park-peter]

Can we reopen this? I just tested the fix in astronomer image 2.3.3-2 and clearing a deferred task will still mark the task as success immediately without executing. The other case of marking the task as failed first then clearing the task does re-execute the task.

[Greetlist]

After adding this code, In my env, this problem is disappear. of course, I do not use astronomer image 2.3.3-2, I just deploy airflow on host machine. Try to attach into container and check source code sudo docker exec -ti $container_id /bin/bash @park-peter

[uranusjr]

Should probably do it in a new issue instead with fresh reproduction steps.