Skip to content

Increment try_number while clearing deferred tasks. #38984

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

Closed
tirkarthi wants to merge 2 commits into from
Closed

Increment try_number while clearing deferred tasks. #38984

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c8f0902
Increment try_number while clearing deferred tasks.
tirkarthi Apr 13, 2024
f450542
Fix test since using example_bash_operator fails for some reason.
tirkarthi Apr 13, 2024
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
6 changes: 6 additions & 0 deletions airflow/models/taskinstance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,12 @@ def clear_task_instances(
ti.state = TaskInstanceState.RESTARTING
job_ids.append(ti.job_id)
else:
# When the task is deferred the try_number is decremented so that the same try
# number is used when the task resumes execution to process the event. But in case of
# clearing the try number should be incremented so that the next run doesn't reuse the same try_number
if ti.state == TaskInstanceState.DEFERRED:

@dirrao dirrao Apr 14, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe clear action is meant to quickly restart things. However, here we are increasing the try_number nothing else.

@tirkarthi tirkarthi Apr 14, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dirrao Sorry, I don't understand your comment. When the task moves from running to deferred the try number is decremented so that when task resumes from the trigger to running the try number is incremented so that same try number is used and helps with logging etc. When the task is cleared then the try number is incremented but since it's in deferred state with try number decremented it causes issues where same log file is used, retry count mismatch etc. for different attempts.

ti._try_number += 1

ti_dag = dag if dag and dag.dag_id == ti.dag_id else dag_bag.get_dag(ti.dag_id, session=session)
task_id = ti.task_id
if ti_dag and ti_dag.has_task(task_id):
Expand Down
36 changes: 36 additions & 0 deletions tests/www/views/test_views_tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -856,6 +856,42 @@ def test_task_instance_clear(session, request, client_fixture, should_succeed):
assert state == (State.NONE if should_succeed else initial_state)


def test_task_instance_clear_deferred(session, admin_client, create_task_instance):
"""Ensures clearing a task instance in deferred state increments _try_number for next execution."""
task_instance = create_task_instance(
dag_id="example_bash_decorator",
task_id="runme_0",
execution_date=timezone.utcnow(),
state=State.DEFERRED,
)
task_id = task_instance.task_id
run_id = task_instance.run_id
old_try_number = task_instance._try_number

data = {
"dag_id": task_instance.dag_id,
"dag_run_id": task_instance.dag_run.run_id,
"execution_date": DEFAULT_DATE,
"task_id": task_instance.task_id,
"upstream": "false",
"downstream": "false",
"future": "true",
"past": "true",
"only_failed": "false",
"confirmed": "true",
}

resp = admin_client.post("clear", data=data, follow_redirects=True)
assert resp.status_code == 200

try_number = (
session.query(TaskInstance._try_number)
.filter(TaskInstance.task_id == task_id, TaskInstance.run_id == run_id)
.scalar()
)
assert try_number == old_try_number + 1


def test_task_instance_clear_downstream(session, admin_client, dag_maker):
"""Ensures clearing a task instance clears its downstream dependencies exclusively"""
with dag_maker(
Expand Down