Issue 1:
Checklist

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).

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Description

I have a fairly basic setup of celery running on 4.4.6 most of the task look like

@app.task(bind=True)
def get_some_data_from_a_server_and_update_records(self, ...):
# Make a request using requests via a Session that is maintained by the instance of the SDK
my_sdk_client = Client()

response = my_sdk_client.session.get(...)

# Update my database

Django celery settings

# Celery
CELERY_BROKER_URL = REDIS_URL
CELERY_RESULT_BACKEND = REDIS_URL

CELERY_TASK_QUEUES = (
Queue('critical', Exchange('critical'), routing_key='critical'),
Queue('high', Exchange('high'), routing_key='high'),
Queue('medium', Exchange('medium'), routing_key='medium'),
Queue('low', Exchange('low'), routing_key='low'),
)
CELERY_TASK_DEFAULT_QUEUE = 'medium'
CELERY_TASK_IGNORE_RESULT = True
CELERY_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_WORKER_HIJACK_ROOT_LOGGER = False
CELERY_WORKER_POOL = 'threads'

Which runs on GKE using multiple pods i.e 8 - 10 pods with each worker reading from a dedicated queue

Unfortunately this results in a number of ConnectionErrors as shown below

File /venv/lib/python3.6/site-packages/requests/sessions.py line 578 in post
return self.request('POST', url, data=data, json=json, **kwargs)
File /venv/lib/python3.6/site-packages/requests/sessions.py line 530 in request
resp = self.send(prep, **send_kwargs)
File /venv/lib/python3.6/site-packages/requests/sessions.py line 643 in send
r = adapter.send(request, **kwargs)
File /venv/lib/python3.6/site-packages/requests/adapters.py line 498 in send args locals
raise ConnectionError(err, request=request)
ConnectionError: ('Connection aborted.', OSError(0, 'Error'))
ProtocolError: ('Connection aborted.', OSError(0, 'Error')) (Most recent call last)

The current docs don't highlight this which seems like a common use case and is currently a pain point.

Based: psf/requests#4323

Most debate around this has been about the lack of multiprocess safety using requests or ensuring you use a single Session per thread/process which doesn't seem feasible when you interact with more than one endpoint or initialize multiple Sessions in different tasks.

We switched our execution pool to use threads but this doesn't solve the issue.

Suggestions

Expect a better setup documentation that clearly states how to configure celery on GKE.

How to work with celery and requests or urllib3

A Recommended execution pool based on how celery is intended to be used. i.e prefork vs threads or gevent / eventlet.

Package Versions:

celery[redis]==4.4.6 # via coherentpath (setup.py), django-celery-beat, flower
certifi==2019.11.28 # via requests, urllib3
cffi==1.13.2 # via bcrypt, cryptography, pynacl
chardet==3.0.4 # via aiohttp, requests
cryptography==3.3.2 # via django-cryptography, paramiko, pyopenssl, requests, urllib3
django-celery-beat==2.0.0 # via coherentpath (setup.py)
django-timezone-field==4.1.1 # via coherentpath (setup.py), django-celery-beat
future==0.18.2 # via celery, django-json-widget, python-jose
humanize==0.5.1 # via flower
idna-ssl==1.1.0 # via aiohttp
idna==2.6 # via idna-ssl, jsonschema, requests, urllib3, yarl
importlib-metadata==1.6.0 # via kombu, scout-apm
kombu==4.6.11 # via celery
oauthlib==3.1.0 # via django-oauth-toolkit, requests-oauthlib
psycopg2==2.7.3.1 # via coherentpath (setup.py)
pycparser==2.19 # via cffi
pyopenssl==19.1.0 # via requests, urllib3
pyparsing==2.4.7 # via httplib2, packaging
python-crontab==2.5.1 # via django-celery-beat
python-dateutil==2.8.1 # via jsonmodels, pandas, python-crontab
redis==3.5.3 # via celery
requests-toolbelt==0.9.1 # via zeep
requests[security]==2.24.0
urllib3[secure]==1.25.11 # via pdpyras, requests, scout-apm, selenium
vine==1.3.0 # via amqp, celery

Issue 2:
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.

Optional Debugging Information

I have tried reproducing the issue on more than one Python version

and/or implementation.

I have tried reproducing the issue on more than one message broker and/or

result backend.

I have tried reproducing the issue on more than one version of the message

broker and/or result backend.

I have tried reproducing the issue on more than one operating system.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with autoscaling, retries,

ETA/Countdown & rate limits disabled.

I have tried reproducing the issue after downgrading

and/or upgrading Celery and its dependencies.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

#4337 -- closed duplicate, partially fixed with #4448 for manual routing (explicitly passing queue name in signature parameters).

Environment & Settings

Celery version:

celery report Output:

celery -A myapp report

software -> celery:5.1.2 (sun-harmonics) kombu:5.1.0 py:3.9.5
billiard:3.6.4.0 py-amqp:5.0.6
platform -> system:Darwin arch:64bit
kernel version:20.6.0 imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:amqp results:db+postgresql://postgres:**@postgres:5432/postgres

broker_url: 'amqp://guest:********@localhost:5672//'
result_backend: 'db+postgresql://postgres:********@postgres:5432/postgres'
task_routes:
({'*': {'queue': 'low_priority_queue'},
'myapp.add_list': {'queue': 'high_priority_queue'}},)
deprecated_settings: None

Steps to Reproduce

Required Dependencies

Minimal Python Version: N/A or Unknown

Minimal Celery Version: N/A or Unknown

Minimal Kombu Version: N/A or Unknown

Minimal Broker Version: N/A or Unknown

Minimal Result Backend Version: N/A or Unknown

Minimal OS and/or Kernel Version: N/A or Unknown

Minimal Broker Client Version: N/A or Unknown

Minimal Result Backend Client Version: N/A or Unknown

Python Packages

pip freeze Output:

Other Dependencies

N/A

Minimally Reproducible Test Case

celeryconfig.py

broker_url = "amqp://guest@localhost//"
result_backend = 'db+postgresql://{user}:{password}@{host}:{port}/{db}'.format(
user='postgres',
password='secretsecret',
host='postgres',
db='postgres',
port=5432,
)

task_routes = ({
"myapp.add_list": {"queue": "high_priority_queue"},
"*": {"queue": "low_priority_queue"},
},)

myapp.py

from celery import Celery

app = Celery('myapp')
app.config_from_object("celeryconfig")

@app.task
def add_list(args):
return sum(args)

if __name__ == '__main__':
app.start()

main.py

from myapp import add_list, app
from celery.canvas import chord

app.config_from_object("celeryconfig")
promise = chord([add_list.s([2, 3]), add_list.s([4, 5])])(add_list.s())
print(promise.get())

Steps to reproduce:

Execute docker run -d -p 5672:5672 rabbitmq to start broker

Execute docker run -d -p 5432:5432 postgres:11.4 for backend.

Run first worker with celery -A myapp worker -l debug -Q high_priority_queue in first terminal window

Run second worker with celery -A myapp worker -l debug -Q low_priority_queue in second terminal window

Run python3 -m main in third terminal window

Expected Behavior

All the tasks are routed to high_priority_queue and executed by the first worker.

Actual Behavior

celery.chord_unlock task is routed to the default low_priority_queue and executed by the second worker.