Issue 1:
Using the Redis backend, If you make modifications to your queues/exchanges/routing keys, you can end up with strange behaviour where queues get duplicated, and you end up with tasks being executed multiple times;

Queue Exchange Routing Key
---------------- ---------------- ----------------
worker tasks worker.#
sender tasks sender.#
notification_send tasks dynamic.notification_send.#
notification_build tasks dynamic.notification_build.#
notification_send tasks notification_send
default tasks task.default.#
notification_build tasks notification_build
notification_build tasks task.notification_build.#
default tasks default.#
notification_send tasks notification_send.#
builder tasks builder.#
notification_build tasks task.notification_build
default tasks task.#
notification_build tasks notification_build.#

To clear this out, I have to manually delete the Redis DB or find the keys via CLI and delete them - at which point things turn back to normal

Queue Exchange Routing Key
---------------- ---------------- ----------------
notification_build notification_build notification_build.#
default tasks task.#
worker worker worker.#

There should be the ability to not only purge tasks, but also purge out bindings / routing entries - because queue changes are going to happen and this is very unexpected behaviour.

Cal

Issue 2:
Checklist

[x ] I have read the relevant section in the

contribution guide

on reporting bugs.

[x ] I have checked the issues list

for similar or identical bug reports.

[x ] I have checked the pull requests list

for existing proposed fixes.

[x ] I have checked the commit log

to find out if the bug was already fixed in the master branch.

[x ] I have included all related issues and possible duplicate issues

in this issue (If there are none, check this box anyway).

Steps to Reproduce

Minimally Reproducible Test Case

Inside a task, raise an exception that does something tricky with its arguments. For example, urllib3's MaxRetryError does this:

class PoolError(HTTPError):
"Base exception for errors caused within a pool."
def __init__(self, pool, message):
self.pool = pool
HTTPError.__init__(self, "%s: %s" % (pool, message))

def __reduce__(self):
# For pickling purposes.
return self.__class__, (None, None)

class RequestError(PoolError):
"Base exception for PoolErrors that have associated URLs."
def __init__(self, pool, url, message):
self.url = url
PoolError.__init__(self, pool, message)

def __reduce__(self):
# For pickling purposes.
return self.__class__, (None, self.url, None)

class MaxRetryError(RequestError):
"""Raised when the maximum number of retries is exceeded.

:param pool: The connection pool
:type pool: :class:`~urllib3.connectionpool.HTTPConnectionPool`
:param string url: The requested Url
:param exceptions.Exception reason: The underlying error

"""

def __init__(self, pool, url, reason=None):
self.reason = reason

message = "Max retries exceeded with url: %s (Caused by %r)" % (
url, reason)

RequestError.__init__(self, pool, url, message)

If you use this exception inside a celery task, like this:

from urllib3.exceptions import MaxRetryError
raise MaxRetryError(None, 'http://fakeurl')

What this results in is an exception whose args contain only a single string, instead of the minimum 2 arguments that MaxRetry requires.

This will make it impossible to use its AsyncResult, because when it tries to rebuild the exception it doesn't have the right number of arguments.

The problem is in celery/backends/base.py in Backend.prepare_exception, which is

def prepare_exception(self, exc, serializer=None):
"""Prepare exception for serialization."""
serializer = self.serializer if serializer is None else serializer
if serializer in EXCEPTION_ABLE_CODECS:
return get_pickleable_exception(exc)
return {'exc_type': type(exc).__name__,
'exc_message': ensure_serializable(exc.args, self.encode),
'exc_module': type(exc).__module__}

The issue is that exc.args can't be relied upon to reinstantiate this exception. Instead, I think we should use the result of exc.__reduce__() which exists to preserve, more-or-less, the constructor arguments for pickling. I think the fix is something like:

def prepare_exception(self, exc, serializer=None):
"""Prepare exception for serialization."""
serializer = self.serializer if serializer is None else serializer
if serializer in EXCEPTION_ABLE_CODECS:
return get_pickleable_exception(exc)
exc_class, exc_args = exc.__reduce__()
return {'exc_type': exc_class.__name__,
'exc_message': ensure_serializable(exc_args, self.encode),
'exc_module': exc_class.__module__}