Ignore stale defer-exit success once the resumed attempt is running

airflow-core/newsfragments/72767.bugfix.rst

@@ -0,0 +1 @@
+Fixed a race where a deferred task was marked failed when the stale executor success from the pre-deferral worker exit arrived after the trigger had already resumed the attempt and it was running. The scheduler now ignores a same-try executor success for a running task instance that still carries ``next_method``, instead of treating it as an external state mismatch.

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -1385,8 +1385,8 @@ def process_executor_events(
         1. **Normal task completion**: Updates task states for successful/failed tasks
         2. **External termination**: Detects tasks killed outside Airflow and marks them as failed
         3. **Task requeuing**: Handles tasks that were requeued by other schedulers or executors,
-           and tasks moved to ``scheduled`` after a trigger fired so a stale executor success from the
-           pre-deferral worker exit does not fail the task instance
+           and tasks resumed by a trigger after deferral (scheduled, queued, or already running)
+           so a stale executor success from the pre-deferral worker exit does not fail the task instance
         4. **Callback processing**: Sends task callback requests to DAG Processor for execution
         5. **Email notifications**: Sends email notification requests to DAG Processor
 
@@ -1559,9 +1559,11 @@ def process_executor_events(
             # from the worker exit after defer() has not been processed yet - should not fail it.
             # 4) the trigger already put the TI back to queued (resume after defer) but the executor success
             # from the worker exit after defer() has not been processed yet - should not fail it.
+            # 5) the resumed attempt is already running (resume after defer) but the executor success
+            # from the worker exit after defer() has not been processed yet - should not fail it.
 
-            # All of this could also happen if the state is "running",
-            # but that is handled by the scheduler detecting task instances without heartbeats.
+            # A running TI outside case 5 (no next_method set) is still a mismatch here; workers lost
+            # without any executor event are picked up separately by heartbeat detection.
 
             ti_queued = ti.try_number == buffer_key.try_number and ti.state in (
                 TaskInstanceState.SCHEDULED,
@@ -1573,9 +1575,17 @@ def process_executor_events(
                 ti.queued_by_job_id != job_id  # Another scheduler has queued this task again
                 or executor.has_task(ti)  # This scheduler has this task already
                 or (
-                    # Resume-after-defer: trigger moved TI to scheduled or queued (next_method set)
-                    # before we saw the executor success from the defer exit for the same try_number.
-                    ti.state in (TaskInstanceState.SCHEDULED, TaskInstanceState.QUEUED)
+                    # Resume-after-defer: trigger resumed the TI (next_method set) before we saw the
+                    # executor success from the defer exit for the same try_number. The resumed
+                    # attempt may already be running: next_method is only cleared on terminal,
+                    # retry and reschedule updates, and a picked-up workload is no longer tracked
+                    # by the executor, so has_task cannot tell the attempts apart.
+                    ti.state
+                    in (
+                        TaskInstanceState.SCHEDULED,
+                        TaskInstanceState.QUEUED,
+                        TaskInstanceState.RUNNING,
+                    )
                     and state == TaskInstanceState.SUCCESS
                     and ti.next_method is not None
                 )

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -1027,6 +1027,70 @@ def test_process_executor_events_stale_success_when_queued_after_defer(
             tags={"dag_id": dag_id, "task_id": ti1.task_id},
         )
 
+    @mock.patch("airflow.jobs.scheduler_job_runner.TaskCallbackRequest")
+    @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
+    def test_process_executor_events_stale_success_when_running_after_defer(
+        self, mock_get_backend, mock_task_callback, dag_maker
+    ):
+        """
+        Resumed attempt already running before executor success from defer exit arrived.
+
+        Regression for https://github.com/apache/airflow/issues/72716 — must not treat as state mismatch.
+        The fixes for #66374 (#66431) and #67287 (#68741) covered the scheduled/queued variants; the
+        resumed attempt may already be RUNNING (next_method is only cleared on terminal/retry updates,
+        and a picked-up workload is no longer tracked by the executor), so this covers the running variant.
+        """
+        mock_stats = mock.MagicMock(spec=StatsLogger)
+        mock_get_backend.return_value = mock_stats
+        dag_id = "test_process_executor_events_stale_success_running_after_defer"
+        task_id_1 = "dummy_task"
+
+        session = settings.Session()
+        with dag_maker(dag_id=dag_id, fileloc="/test_path1/"):
+            task1 = EmptyOperator(task_id=task_id_1)
+        ti1 = dag_maker.create_dagrun().get_task_instance(task1.task_id)
+
+        executor = MockExecutor(do_update=False)
+        task_callback = mock.MagicMock()
+        mock_task_callback.return_value = task_callback
+        scheduler_job = Job()
+        session.add(scheduler_job)
+        session.flush()
+        self.job_runner = SchedulerJobRunner(scheduler_job, executors=[executor])
+
+        ti1.state = State.RUNNING
+        ti1.next_method = "execute_callback"
+        ti1.queued_by_job_id = scheduler_job.id
+        ti1.try_number = 1
+        session.merge(ti1)
+        session.commit()
+
+        executor.event_buffer[ti1.key] = State.SUCCESS, None
+        executor.has_task = mock.MagicMock(return_value=False)
+        mock_stats.incr.reset_mock()
+
+        self.job_runner._process_executor_events(executor=executor, session=session)
+        ti1.refresh_from_db(session=session)
+        assert ti1.state == State.RUNNING
+        self.job_runner.executor.callback_sink.send.assert_not_called()
+        # Stale success from defer exit must not trigger a mismatch metric —
+        # only the standard processed-events counter should fire.
+        mock_stats.incr.assert_called_once_with("scheduler.executor_events.processed", count=1)
+
+        # Without next_method, running + stale success is still a mismatch (e.g. external kill).
+        ti1.next_method = None
+        session.merge(ti1)
+        session.commit()
+
+        executor.event_buffer[ti1.key] = State.SUCCESS, None
+        mock_stats.incr.reset_mock()
+
+        self.job_runner._process_executor_events(executor=executor, session=session)
+        mock_stats.incr.assert_any_call(
+            "scheduler.tasks.killed_externally",
+            tags={"dag_id": dag_id, "task_id": ti1.task_id},
+        )
+
     @mock.patch("airflow.jobs.scheduler_job_runner.TaskCallbackRequest")
     @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
     def test_process_executor_events_multiple_try_numbers_warns(