queue worker shutdown: disable restore

fix: #7884
The complete task restored back to the queue causing a task duplication.
Might related to a bug from the celery celery/celery#5663

Follow what in https://github.com/celery/celery/issues/5663#issuecomment-734871162
Disabled the restore_at_shutdown
This commit is contained in:
karajan1001
2022-07-12 21:39:49 +09:00
committed by Peter Rowlands (변기호)
parent d7a917e2ab
commit 16bb31a284
+8 -3
View File
@@ -71,6 +71,11 @@ class LocalCeleryQueue(BaseStashQueue):
@cached_property
def celery(self) -> "FSApp":
from kombu.transport.filesystem import Channel # type: ignore
# related to https://github.com/iterative/dvc-task/issues/61
Channel.QoS.restore_at_shutdown = False
from dvc_task.app import FSApp
app = FSApp(
@@ -181,7 +186,7 @@ class LocalCeleryQueue(BaseStashQueue):
for msg, entry in self._iter_processed():
task_id = msg.headers["id"]
result = AsyncResult(task_id)
result: AsyncResult = AsyncResult(task_id)
if not result.ready():
yield _TaskEntry(task_id, entry)
@@ -190,7 +195,7 @@ class LocalCeleryQueue(BaseStashQueue):
for msg, entry in self._iter_processed():
task_id = msg.headers["id"]
result = AsyncResult(task_id)
result: AsyncResult = AsyncResult(task_id)
if result.ready():
yield _TaskEntry(task_id, entry)
@@ -230,7 +235,7 @@ class LocalCeleryQueue(BaseStashQueue):
if entry.stash_rev == active_entry.stash_rev:
logger.debug("Waiting for exp task '%s' to complete", task_id)
try:
result = AsyncResult(task_id)
result: AsyncResult = AsyncResult(task_id)
result.get(timeout=timeout)
except _CeleryTimeout as exc:
raise DvcException(