Allow CeleryExecutor to "adopt" an orphaned queued or running task #10949
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
boring-cyborg
bot
added
area:docs
area:Scheduler
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
b58fac4 to
c1a3f9f
Compare
kaxil
force-pushed
the
adopt-dont-reset-celery-tasks
branch
4 times, most recently
from
4f3cf22 to
627207a
Compare
kaxil
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
627207a to
2871b48
Compare
8 tasks
XD-DENG
reviewed
Sep 15, 2020
XD-DENG
reviewed
Sep 15, 2020
airflow/migrations/versions/e1a11ece99cc_add_external_executor_id_to_ti.py
Outdated
Show resolved
Hide resolved
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
2871b48 to
12debd5
Compare
10 tasks
Member
Author
|
The doc tests are now failing with:
But I didn't change the imports in that file :/ Oh I missed Kaxil's earlier fix in my rebase+force-push |
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
2 times, most recently
from
bbb6f5f to
03a9290
Compare
XD-DENG
requested changes
Sep 16, 2020
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
03a9290 to
7d13295
Compare
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
2 times, most recently
from
64c2215 to
7c05b37
Compare
turbaszek
reviewed
Sep 16, 2020
turbaszek
reviewed
Sep 16, 2020
turbaszek
reviewed
Sep 16, 2020
turbaszek
reviewed
Sep 16, 2020
turbaszek
reviewed
Sep 16, 2020
turbaszek
reviewed
Sep 16, 2020
XD-DENG
approved these changes
Sep 16, 2020
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
3f4b2ea to
f1967f6
Compare
Member
Author
|
Right, final rebase done, will merge once tests are green. (I wish Github had a button for that) |
This can happen when a task is enqueued by one executor, and then that scheduler dies/exits. The default fallback behaviour is unchanged -- that queued tasks are cleared and then and then later rescheduled. But for Celery we can do better -- if we record the Celery-generated task_id, we can then re-create the AsyncResult objects for orphaned tasks at a later date. However since Celery just reports all AsyncResult as "PENDING", even if they aren't tasks currently in the broker queue, we need to apply a timeout to "unblock" these tasks in case they never actually made it to the Celery broker. This all means that we can adopt tasks that have been enqueued another CeleryExecutor if it dies, without having to clear the task and slow down. This is especially useful as the task may have already started running, and while clearing it would stop it, it's better if we don't have to reset it! Co-authored-by: Kaxil Naik <[email protected]>
ashb
force-pushed
the
adopt-dont-reset-celery-tasks
branch
from
f1967f6 to
092b9c0
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This can happen when a task is enqueued by one executor, and then that
scheduler dies/exits.
The default fallback behaviour is unchanged -- that queued tasks are
cleared and then and then later rescheduled.
But for Celery we can do better -- if we record the Celery-generated
task_id, we can then re-create the AsyncResult objects for orphaned
tasks at a later date.
However since Celery just reports all AsyncResult as "PENDING", even if
they aren't tasks currently in the broker queue, we need to apply a
timeout to "unblock" these tasks in case they never actually made it to
the Celery broker.
This all means that we can adopt tasks that have been enqueued another
CeleryExecutor if it dies, without having to clear the task and slow
down. This is especially useful as the task may have already started
running, and while clearing it would stop it, it's better if we don't
have to reset it!
Part of #9630
^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.