Deferred TI's next_method and next_kwargs not cleared on retries #18146

Activity

1. jedcunningham

   added

   kind:bugThis is a clearly a bugThis is a clearly a bug

   area:core

   on Sep 10, 2021
2. jedcunningham

   added this to the Airflow 2.2 milestone on Sep 10, 2021
3. jedcunningham

   changed the title [-]Deferrable operator not clearing `next_method` and `next_kwargs` on retries[/-] [+]Deferred TI's `next_method` and `next_kwargs` not cleared on retries[/+] on Sep 10, 2021
4. potiuk

   assigned

   jedcunningham

   on Sep 10, 2021
5. andrewgodwin

   added a commit that references this issue on Sep 13, 2021

   Wipe next_kwargs and next_method on task failure

   ...

   This issue will close once commit c93beb5 is merged into the 'main' branch.

   c93beb5
6. kaxil

   closed this as completedin #18210on Sep 13, 2021
7. kaxil

   added a commit that references this issue on Sep 13, 2021

   BugFix: Wipe ``next_kwargs`` and ``next_method`` on task failure (#18210

   ...

   This issue will close once commit 9d49772 is merged into the 'main' branch.

   9d49772
8. jedcunningham commented on Sep 13, 2021

   

   jedcunningham

   on Sep 13, 2021

   Last edited by kaxil

   MemberAuthor

   More actions

   The following DAG can be used to both reproduce (dagruns fail) and verify the issue is fixed (dagruns succeed). Basically, the first run it will fail so it ends up in UP_FOR_RETRY and will succeed on the second try.

   from datetime import datetime, timedelta
   
   from airflow import DAG
   from airflow.exceptions import AirflowException
   from airflow.models import BaseOperator
   from airflow.triggers.testing import SuccessTrigger
   
   
   class RetryOperator(BaseOperator):
       def execute(self, context):
           ti = context["ti"]
           has_next_method = bool(ti.next_method)
           try_number = ti.try_number
           self.log.info(
               f"In `execute`: has_next_method: {has_next_method}, try_number:{try_number}"
           )
   
           self.defer(
               trigger=SuccessTrigger(),
               method_name="next",
               kwargs={"execute_try_number": try_number},
           )
   
       def next(self, context, execute_try_number, event=None):
           self.log.info("In next!")
           ti = context["ti"]
           has_next_method = bool(ti.next_method)
           try_number = ti.try_number
           self.log.info(
               f"In `next`: has_next_method: {has_next_method}, try_number:{try_number}, excute_try_number: {execute_try_number}"
           )
   
           if try_number == 1:
               # Force a retry
               raise AirflowException("Force a retry")
           # Did we run `execute`?
           if execute_try_number != try_number:
               raise AirflowException("`execute` wasn't run during retry!")
           return None  # Success!
   
   
   with DAG(
       "triggerer_retry", schedule_interval=None, start_date=datetime(2021, 9, 13)
   ) as dag:
       RetryOperator(task_id="retry", retries=1, retry_delay=timedelta(seconds=15))