diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py index 2a74bde0807c0..ca02969a79ed0 100644 --- a/airflow/models/taskinstance.py +++ b/airflow/models/taskinstance.py @@ -1693,6 +1693,11 @@ def handle_failure( # Log failure duration session.add(TaskFail(task, self.execution_date, self.start_date, self.end_date)) + # Ensure we unset next_method and next_kwargs to ensure that any + # retries don't re-use them. + self.next_method = None + self.next_kwargs = None + # Set state correctly and figure out how to log it and decide whether # to email diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py index e22a44748e602..28afaa9918b3c 100644 --- a/tests/models/test_taskinstance.py +++ b/tests/models/test_taskinstance.py @@ -459,6 +459,36 @@ def task_function(ti): ti.refresh_from_db() assert ti.state == State.UP_FOR_RETRY + def test_task_retry_wipes_next_fields(self, session, dag_maker): + """ + Test that ensures that tasks wipe their next_method and next_kwargs + fields when they are queued for retry after a failure. + """ + + with dag_maker('test_mark_failure_2'): + task = BashOperator( + task_id='test_retry_handling_op', + bash_command='exit 1', + retries=1, + retry_delay=datetime.timedelta(seconds=2), + ) + + dr = dag_maker.create_dagrun() + ti = dr.task_instances[0] + ti.next_method = "execute" + ti.next_kwargs = {} + session.merge(ti) + session.commit() + + ti.task = task + with pytest.raises(AirflowException): + ti.run() + ti.refresh_from_db() + + assert ti.next_method is None + assert ti.next_kwargs is None + assert ti.state == State.UP_FOR_RETRY + def test_retry_delay(self, dag_maker): """ Test that retry delays are respected