From 25e4462e28ce7164012e57dfeac099a8f718725d Mon Sep 17 00:00:00 2001
From: Eugene Karimov <13220923+eskarimov@users.noreply.github.com>
Date: Wed, 5 Jan 2022 19:19:44 +0100
Subject: [PATCH] Use original task's `start_date` if a task continues after
 deferral (#20062)

GitOrigin-RevId: 5cc5f434d745452651bca76ba6d2406167b7e2b9
---
 airflow/models/taskinstance.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index fba12c14faa..ceead7f4efc 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1207,7 +1207,8 @@ def check_and_change_state_before_execution(
             # Attempt 0 for the first attempt).
             # Set the task start date. In case it was re-scheduled use the initial
             # start date that is recorded in task_reschedule table
-            self.start_date = timezone.utcnow()
+            # If the task continues after being deferred (next_method is set), use the original start_date
+            self.start_date = self.start_date if self.next_method else timezone.utcnow()
             if self.state == State.UP_FOR_RESCHEDULE:
                 task_reschedule: TR = TR.query_for_task_instance(self, session=session).first()
                 if task_reschedule:
@@ -1533,6 +1534,8 @@ def _defer_task(self, session, defer: TaskDeferred):
         session.flush()
 
         # Then, update ourselves so it matches the deferral request
+        # Keep an eye on the logic in `check_and_change_state_before_execution()`
+        # depending on self.next_method semantics
         self.state = State.DEFERRED
         self.trigger_id = trigger_row.id
         self.next_method = defer.method_name