Introduce RESTARTING state

[yuqian90]

closes: #16680

This PR makes sure that when a user clears a running task, the task does not fail. Instead it is killed and retried gracefully.

This is done by introducing a new State called RESTARTING. As the name suggests, a TaskInstance is set to this state when it's cleared while running. Most of the places handles RESTARTING the same way SHUTDOWN is handled, except in TaskInstance.is_eligible_to_retry, where it is always be treated as eligible for retry.

[ashb]

Is a new state strictly required for this? Cos each state we add needs to be shown in the UI and increases the possible places we need to deal with the new state

[yuqian90]

Is a new state strictly required for this? Cos each state we add needs to be shown in the UI and increases the possible places we need to deal with the new state

I think another way to implement this is to make clear_task_instances increase the max_tries of the running TaskInstance by 1. However, when I tried to do that I realized it's easy. There are more places that need to be changed to do the right thing. One example is is_eligible_to_retry(). It first checks if self.task.retries is zero. If it is, it short-circuits and returns False. But self.task.retries isn't something that can be incremented by clear_task_instances because it's not stored in the db.

    def is_eligible_to_retry(self):
        """Is task instance is eligible for retry"""
        return self.task.retries and self.try_number <= self.max_tries

[yuqian90]

Ready for review. The failing Helm Chart; KubernetesExecutor (pull_request) looks unrelated to this PR.

[potiuk]

Ready for review. The failing Helm Chart; KubernetesExecutor (pull_request) looks unrelated to this PR.

Yep hopefully already fixed by #16750

[ashb]

Code and name looks good now, though this is probably worth documenting somewhere so users can know about this behaviour.

(And you'll have to resolve conflicts obviously.)

[yuqian90]

Code and name looks good now, though this is probably worth documenting somewhere so users can know about this behaviour.

(And you'll have to resolve conflicts obviously.)

Thanks. That's done. I added a paragraph in UPDATING.md about this new behaviour. Please see if that helps.

[ashb]

I mostly also thinking somewhere in the published docs tree.

Ah https://airflow.apache.org/docs/apache-airflow/stable/concepts/tasks.html#task-instances

That diagram needs updating :)

[yuqian90]

Thanks everyone for the suggestions in #6762. @kaxil @potiuk @mik-laj @kaverisharma09 I re-created the diagram with draw.io and here's a draft. Some modifications I made:

• Added "restarting" state introduced in Introduce RESTARTING state #16681
• Named the state according to their definitions in state.py
• Added a choice node to represent the alternative state transition
• Moved the standard lifecycle to the top of the graph. I.e. "none, to scheduled, to queued, to running, and finally to success" is illustrated at the top of the diagram.

Please let me know what you think. I'll include the diagram in my PR and upload to https://cwiki.apache.org/confluence/display/AIRFLOW/File+lists if you think it works.

[kaxil]

Thanks everyone for the suggestions in #6762. @kaxil @potiuk @mik-laj @kaverisharma09 I re-created the diagram with draw.io and here's a draft. Some modifications I made:

• Added "restarting" state introduced in Introduce RESTARTING state #16681
• Named the state according to their definitions in state.py
• Added a choice node to represent the alternative state transition
• Moved the standard lifecycle to the top of the graph. I.e. "none, to scheduled, to queued, to running, and finally to success" is illustrated at the top of the diagram.

Please let me know what you think. I'll include the diagram in my PR and upload to https://cwiki.apache.org/confluence/display/AIRFLOW/File+lists if you think it works.

Sounds good

[potiuk]

Looks good to me :)

[potiuk]

Hey @ashb - any more comments ? I think I would love to merge that one :)

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[potiuk]

Hey @yuqian90 , can you please rebase? We had some master issues recently, but they should be solved already :)