Issue 1:
Checklist

I have verified that the issue exists against the master branch of Celery.

This has already been asked to the discussion group first.

I have read the relevant section in the

contribution guide

on reporting bugs.

I have checked the issues list

for similar or identical bug reports.

I have checked the pull requests list

for existing proposed fixes.

I have checked the commit log

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

I have included all related issues and possible duplicate issues

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

Mandatory Debugging Information

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have verified that the issue exists against the master branch of Celery.

I have included the contents of pip freeze in the issue.

I have included all the versions of all the external dependencies required

to reproduce this bug.

Environment & Settings

Celery version: 4.4.7

celery report Output:

celery -A backend.celery_app:app report

software -> celery:4.4.7 (cliffs) kombu:4.6.11 py:3.8.5
billiard:3.6.3.0 redis:3.5.3
platform -> system:Linux arch:64bit
kernel version:4.15.0-112-generic imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:redis results:redis://redis:6379/2

broker_url: 'redis://redis:6379/1'
result_backend: 'redis://redis:6379/2'
accpet_content: ['application/json']
task_serializer: 'json'
result_serializer: 'json'
task_acks_late: True
task_default_queue: 'default'
worker_send_task_events: True
worker_prefetch_multiplier: 1
task_queues:
(<unbound Queue default -> <unbound Exchange default(direct)> -> default>,
<unbound Queue other -> <unbound Exchange other(direct)> -> other>)
task_routes: {
'core.tasks.debug_task': { 'exchange': 'default',
'queue': 'default',
'routing_key': '********'},
'core.tasks.debug_task_other': { 'exchange': 'other',
'queue': 'other',
'routing_key': '********'}}
task_default_exchange_type: 'direct'

Python Packages

pip freeze Output:

amqp==2.6.1
appdirs==1.4.4
argh==0.26.2
argon2-cffi==20.1.0
asgiref==3.2.10
atomicwrites==1.4.0
attrs==20.1.0
backcall==0.2.0
billiard==3.6.3.0
black==19.10b0
bleach==3.1.5
celery==4.4.7
cffi==1.14.2
click==7.1.2
decorator==4.4.2
defusedxml==0.6.0
Django==3.1
django-debug-toolbar==2.2
django-extensions==3.0.5
entrypoints==0.3
factory-boy==2.11.1
Faker==4.1.2
gunicorn==20.0.4
ipykernel==5.3.4
ipython==7.17.0
ipython-genutils==0.2.0
ipywidgets==7.5.1
jedi==0.17.2
Jinja2==2.11.2
jsonschema==3.2.0
jupyter==1.0.0
jupyter-client==6.1.6
jupyter-console==6.1.0
jupyter-core==4.6.3
kombu==4.6.11
MarkupSafe==1.1.1
mistune==0.8.4
more-itertools==8.4.0
nbconvert==5.6.1
nbformat==5.0.7
notebook==6.1.3
packaging==20.4
pandocfilters==1.4.2
parso==0.7.1
pathspec==0.8.0
pathtools==0.1.2
pexpect==4.8.0
pickleshare==0.7.5
pluggy==0.13.1
prometheus-client==0.8.0
prompt-toolkit==3.0.6
psycopg2-binary==2.8.5
ptyprocess==0.6.0
py==1.9.0
pycparser==2.20
Pygments==2.6.1
pyparsing==2.4.7
pyrsistent==0.16.0
pytest==4.4.1
pytest-django==3.4.8
python-dateutil==2.8.1
pytz==2020.1
PyYAML==5.3.1
pyzmq==19.0.2
qtconsole==4.7.6
QtPy==1.9.0
redis==3.5.3
regex==2020.7.14
Send2Trash==1.5.0
six==1.15.0
sqlparse==0.3.1
terminado==0.8.3
testpath==0.4.4
text-unidecode==1.3
toml==0.10.1
tornado==6.0.4
traitlets==4.3.3
typed-ast==1.4.1
vine==1.3.0
watchdog==0.10.3
wcwidth==0.2.5
webencodings==0.5.1
widgetsnbextension==3.5.1

Minimally Reproducible Test Case

Here is a fully reproducible example: https://gitlab.com/verbose-equals-true/digital-ocean-docker-swarm that can be setup with docker-compose. Run docker-compose up to start it.

I am using celery with Django and redis as the broker. I'm trying to setup two queues: default and other. My tasks are working, but the settings I have configured are not working have I am expecting them to work.

Actual Behavior

I'm having two related issues:

celery tasks are not respecting the task_routes setting (see below).

all of the celery tasks (no matter how they are defined) are registered to each of the two queues when they are started

Here are the relevant parts of my code:

the celery app definition file

task definitions/declarations

commands to start workers

celery app definition:

from celery import Celery
from django.conf import settings

from kombu import Exchange, Queue

CELERY_QUEUE_DEFAULT = 'default'
CELERY_QUEUE_OTHER = 'other'

app = Celery('backend')
app.conf["broker_url"] = f"redis://{settings.REDIS_SERVICE_HOST}:6379/1"
app.conf["result_backend"] = f"redis://{settings.REDIS_SERVICE_HOST}:6379/2"
app.conf["accpet_content"] = ['application/json']
app.conf["task_serializer"] = 'json'
app.conf["result_serializer"] = 'json'
app.conf["task_acks_late"] = True
app.conf["task_default_queue"] = CELERY_QUEUE_DEFAULT
app.conf["worker_send_task_events"] = True
app.conf["worker_prefetch_multiplier"] = 1
app.conf["task_queues"] = (
Queue(
CELERY_QUEUE_DEFAULT,
Exchange(CELERY_QUEUE_DEFAULT),
routing_key=CELERY_QUEUE_DEFAULT,
),
Queue(
CELERY_QUEUE_OTHER,
Exchange(CELERY_QUEUE_OTHER),
routing_key=CELERY_QUEUE_OTHER,
),
)
app.conf["task_routes"] = {
'backend.core.tasks.debug_task': {
'queue': 'default',
'routing_key': 'default',
'exchange': 'default',
},
'backend.core.tasks.debug_task_other': {
'queue': 'other',
'routing_key': 'other',
'exchange': 'other',
},
}
app.conf["task_default_exchange_type"] = 'direct'

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

Task definitions (defined in a file called tasks.py in an app called core:

import time

import celery

# from backend import celery_app as app
# from celery import shared_task
from celery.task import task
from django.conf import settings

# @celery.task <-- I have seen these decorators in other example
# @app.task <-- neither of these result in the tasks being sent to the correct queue
# @shared_task
@task(queue="default", exchange="default")
def debug_task():
time.sleep(10)
return "Task is done."

@task(queue="other", exchange="other")
def debug_task_other():
time.sleep(10)
return "Task is done for other queue."

Here's how I'm starting my workers in docker-compose locally:

celery:
<<: *backend
container_name: celery
command:
- "watchmedo"
- "auto-restart"
- "--directory=./"
- "--pattern=*.py"
- "--recursive"
- "--"
- "celery"
- "worker"
- "--app=backend.celery_app:app"
- "-Q"
- "default"
- "--concurrency=1"
- "--loglevel=INFO"
ports: []

celery_other:
<<: *backend
container_name: celery_other
command:
- "watchmedo"
- "auto-restart"
- "--directory=./"
- "--pattern=*.py"
- "--recursive"
- "--"
- "celery"
- "worker"
- "--app=backend.celery_app:app"
- "-Q"
- "other"
- "--concurrency=1"
- "--loglevel=INFO"
ports: []

Here are the logs from docker-compose that show that the two tasks are both registered to each worker:

celery |
celery | -------------- celery@c8f0ed3f97df v4.4.7 (cliffs)
celery | --- ***** -----
celery | -- ******* ---- Linux-4.15.0-112-generic-x86_64-with-glibc2.2.5 2020-08-21 20:35:47
celery | - *** --- * ---
celery | - ** ---------- [config]
celery | - ** ---------- .> app: backend:0x7fb327c2e6a0
celery | - ** ---------- .> transport: redis://redis:6379/1
celery | - ** ---------- .> results: redis://redis:6379/2
celery | - *** --- * --- .> concurrency: 1 (prefork)
celery | -- ******* ---- .> task events: ON
celery | --- ***** -----
celery | -------------- [queues]
celery | .> default exchange=default(direct) key=default
celery |
celery |
celery | [tasks]
celery | . core.tasks.debug_task
celery | . core.tasks.debug_task_other
celery |
celery_other |
celery_other | -------------- celery@3dd99b1ed32e v4.4.7 (cliffs)
celery_other | --- ***** -----
celery_other | -- ******* ---- Linux-4.15.0-112-generic-x86_64-with-glibc2.2.5 2020-08-21 20:35:47
celery_other | - *** --- * ---
celery_other | - ** ---------- [config]
celery_other | - ** ---------- .> app: backend:0x7f54fa89e6a0
celery_other | - ** ---------- .> transport: redis://redis:6379/1
celery_other | - ** ---------- .> results: redis://redis:6379/2
celery_other | - *** --- * --- .> concurrency: 1 (prefork)
celery_other | -- ******* ---- .> task events: ON
celery_other | --- ***** -----
celery_other | -------------- [queues]
celery_other | .> other exchange=other(direct) key=other
celery_other |
celery_other |
celery_other | [tasks]
celery_other | . core.tasks.debug_task
celery_other | . core.tasks.debug_task_other

I was thinking that defining task_routes would mean that I don't have to specify the tasks's queue in the task decorator. If I don't specify the queue, the tasks are all picked up by the default worker.

If it helps, here is my Django directory structure:

tree -L 3 backend
backend
├── backend
│ ├── asgi.py
│ ├── celery_app.py <- this is where I define my celery app
│ ├── __init__.py
│ ├── settings
│ │ ├── base.py
│ │ ├── development.py
│ │ ├── __init__.py
│ │ └── production.py
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
├── core
│ ├── admin.py
│ ├── apps.py
│ ├── __init__.py
│ ├── migrations
│ │ └── __init__.py
│ ├── models.py
│ ├── tasks.py <- this is where I define the tasks shown above
│ ├── tests.py
│ ├── urls.py
│ └── views.py
├── docker
│ ├── Dockerfile.dev
│ └── Dockerfile.prod
├── manage.py
└── requirements
├── base.txt
├── dev.txt
└── test.txt

I have tried to follow the Routing Tasks page from the celery documentation to get everything setup correctly: https://docs.celeryproject.org/en/stable/userguide/routing.html

Issue 2:
it's duplicate of celery/kombu#870, but in order to fix 4.1 I suggest creating a 4.1.1 release with kombu dependency set to >=4,1,<4.2