(airflow) branch main updated: Don't update start_date when resuming a TI from deferral (#49407)

[ds...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dstandish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 60d4df1b894 Don't update start_date when resuming a TI from deferral (#49407)
60d4df1b894 is described below

commit 60d4df1b8942813ce47bb60d411026da52ed04ce
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Thu Apr 17 11:56:12 2025 -0700

    Don't update start_date when resuming a TI from deferral (#49407)
    
    This allows duration to reflect the overall time from task start to task completion, which is how the metric was calculated in 2.x.
---
 .../execution_api/routes/task_instances.py         |  4 ++
 .../versions/head/test_task_instances.py           | 61 ++++++++++++++++++++++
 2 files changed, 65 insertions(+)

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index af2bf86c7cd..ca8058bfd3a 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -131,6 +131,10 @@ def ti_run(
     # We exclude_unset to avoid updating fields that are not set in the payload
     data = ti_run_payload.model_dump(exclude_unset=True)
 
+    # don't update start date when resuming from deferral
+    if ti.next_kwargs:
+        data.pop("start_date")
+
     query = update(TI).where(TI.id == ti_id_str).values(data)
 
     previous_state = ti.state
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 4d36a2f5515..49d4f42b9a6 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -268,6 +268,67 @@ class TestTIRunState:
             },
         }
 
+    @pytest.mark.parametrize("resume", [True, False])
+    def test_next_kwargs_determines_start_date_update(self, client, session, create_task_instance, resume):
+        dag_start_time_str = "2024-09-30T12:00:00Z"
+        dag_start_time = timezone.parse(dag_start_time_str)
+        orig_task_start_time = dag_start_time.add(seconds=5)
+
+        ti = create_task_instance(
+            task_id="test_next_kwargs_still_encoded",
+            state=State.QUEUED,
+            session=session,
+            start_date=orig_task_start_time,
+        )
+
+        ti.start_date = orig_task_start_time
+        ti.next_method = "execute_complete"
+
+        second_start_time = orig_task_start_time.add(seconds=30)
+        second_start_time_str = second_start_time.isoformat()
+
+        # ti.next_kwargs under the hood applies the serde encoding for us
+        if resume:
+            ti.next_kwargs = {"moment": second_start_time}
+            expected_start_date = orig_task_start_time
+            expected_next_kwargs = {
+                "__type": "dict",
+                "__var": {"moment": {"__type": "datetime", "__var": second_start_time.timestamp()}},
+            }
+        else:
+            expected_start_date = second_start_time
+            expected_next_kwargs = None
+
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": second_start_time_str,
+            },
+        )
+        session.commit()
+
+        assert response.status_code == 200
+        assert response.json() == {
+            "dag_run": mock.ANY,
+            "task_reschedule_count": 0,
+            "max_tries": 0,
+            "should_retry": False,
+            "variables": [],
+            "connections": [],
+            "xcom_keys_to_clear": [],
+            "next_method": "execute_complete",
+            "next_kwargs": expected_next_kwargs,
+        }
+        session.expunge_all()
+        ti = session.get(TaskInstance, ti.id)
+        assert ti.start_date == expected_start_date
+
     @pytest.mark.parametrize(
         "initial_ti_state",
         [s for s in TaskInstanceState if s not in (TaskInstanceState.QUEUED, TaskInstanceState.RESTARTING)],