Skip to content

Ignore stale executor success when a deferred task has resumed to running #71579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
raghavan-arvind wants to merge 1 commit into
base: main
Choose a base branch
from
+20 −73
Open

Ignore stale executor success when a deferred task has resumed to running #71579

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions airflow-core/src/airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1534,6 +1534,8 @@ 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 trigger fired fast enough that the resumed TI is already running again (next_method
# set) when the executor success from the defer exit arrives - 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.
Expand All @@ -1548,9 +1550,11 @@ 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 moved TI to scheduled, queued or already running
# (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, TaskInstanceState.RUNNING)
and state == TaskInstanceState.SUCCESS
and ti.next_method is not None
)
Expand Down
83 changes: 13 additions & 70 deletions airflow-core/tests/unit/jobs/test_scheduler_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -891,19 +891,24 @@ def test_process_executor_events_ti_requeued(
# no killed_externally mismatch metric should appear.
assert all(c.args[0] == "scheduler.executor_events.processed" for c in mock_stats.incr.call_args_list)

@pytest.mark.parametrize("resumed_state", [State.SCHEDULED, State.QUEUED, State.RUNNING])
@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_scheduled_after_defer(
self, mock_get_backend, mock_task_callback, dag_maker
def test_process_executor_events_stale_success_when_resumed_after_defer(
self, mock_get_backend, mock_task_callback, dag_maker, resumed_state
):
"""
Trigger moved TI to scheduled (resume after defer) before executor success from defer exit arrived.
Trigger resumed the TI (next_method set) before executor success from defer exit arrived.

Regression for https://github.com/apache/airflow/issues/66374 — must not treat as state mismatch.
Regression for https://github.com/apache/airflow/issues/66374 (scheduled, fixed by #66431),
https://github.com/apache/airflow/issues/67287 (queued, fixed by #68741), and
https://github.com/apache/airflow/issues/71580 (running, fixed by #71579) — must not treat as
state mismatch. A trigger that fires immediately after deferral can beat executor event
processing all the way to running.
"""
mock_stats = mock.MagicMock(spec=StatsLogger)
mock_get_backend.return_value = mock_stats
dag_id = "test_process_executor_events_stale_success_scheduled_after_defer"
dag_id = "test_process_executor_events_stale_success_resumed_after_defer"
task_id_1 = "dummy_task"

session = settings.Session()
Expand All @@ -919,7 +924,7 @@ def test_process_executor_events_stale_success_when_scheduled_after_defer(
session.flush()
self.job_runner = SchedulerJobRunner(scheduler_job, executors=[executor])

ti1.state = State.SCHEDULED
ti1.state = resumed_state
ti1.next_method = "execute_callback"
ti1.queued_by_job_id = scheduler_job.id
ti1.try_number = 1
Expand All @@ -932,75 +937,13 @@ def test_process_executor_events_stale_success_when_scheduled_after_defer(

self.job_runner._process_executor_events(executor=executor, session=session)
ti1.refresh_from_db(session=session)
assert ti1.state == State.SCHEDULED
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, scheduled + 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_stale_success_when_queued_after_defer(
self, mock_get_backend, mock_task_callback, dag_maker
):
"""
Trigger moved TI to queued (resume after defer) before executor success from defer exit arrived.

Regression for https://github.com/apache/airflow/issues/67287 — must not treat as state mismatch.
The fix for #66374 (#66431) covered the scheduled-state variant; this covers the queued-state variant.
"""
mock_stats = mock.MagicMock(spec=StatsLogger)
mock_get_backend.return_value = mock_stats
dag_id = "test_process_executor_events_stale_success_queued_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.QUEUED
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.QUEUED
assert ti1.state == resumed_state
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, queued + stale success is still a mismatch (e.g. external kill).
# Without next_method, a stale success is still a mismatch (e.g. external kill).
ti1.next_method = None
session.merge(ti1)
session.commit()
Expand Down