Revoke a task from celery

Question

I want to explicitly revoke a task from celery. This is how I'm currently doing:-

from celery.task.control import revoke

revoke(task_id, terminate=True)

where task_id is string(have also tried converting it into UUID uuid.UUID(task_id).hex).

After the above procedure, when I start celery again celery worker -A proj it still consumes the same message and starts processing it. Why?

When viewed via flower, the message is still there in the broker section. how do I delete the message so that it cant be consumed again?

Solution

How does revoke works?

When calling the revoke method the task doesn't get deleted from the queue immediately, all it does is tell celery(not your broker!) to save the task_id in a in-memory set(look here if you like reading source code like me).

When the task gets to the top of the queue, Celery will check if is it in the revoked set, if it does, it won't execute it.

It works this way to prevent O(n) search for each revoke call, where checking if the task_id is in the in-memory set is just O(1)

Why after restarting celery, your revoked tasks executed?

Understanding how things works, you now know that the set is just a normal python set, that being saved in-memory - that means when you restart, you lose this set, but the task is(of course) persistence and when the tasks turn comes, it will be executed as normal.

What can you do?

You will need to have a persistence set, this is done by initial your worker like this:

celery worker -A proj --statedb=/var/run/celery/worker.state

This will save the set on the filesystem.

References:

• Celery source code of the in-memory set
• Revoke doc
• Persistent revokes docs

Comments

[e.arbitrio]

How many workers are you using? Revoking tasks works by sending a broadcast message to all the workers, the workers then keep a list of revoked tasks in memory. When a worker starts up it will synchronize revoked tasks with other workers in the cluster. docs.celeryproject.org/en/latest/userguide/…

[Jens]

I’ve been trying to revoke() a task but it actually executes despite being marked as revoked: github.com/celery/celery/issues/4300 Any ideas what I might miss?

[A.Raouf]

revoke also require state parameter which I don't know what is? did u know it ?