Issue 1:
Checklist

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 included all related issues and possible duplicate issues in this issue.

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

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

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 workers pool.

I have tried reproducing the issue with retries, ETA/Countdown & rate limits disabled.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Environment & Settings

Celery version:4.2.1

celery report Output:

Steps to Reproduce

Required Dependencies

Minimal Python Version:2.7

Minimal Broker Version: redis 5.0.3

Minimal Result Backend Version:redis 5.0.3

Minimal OS and/or Kernel Version: ubuntu 16.04

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

main.python:

import celery_app_tmp.task0

import celery_app_tmp.task1

import celery_app_tmp.task_download

list_task = [celery_app_tmp.task0,

celery_app_tmp.task1]

addImageInput = {'imageMD5': 'default',

'imageUrl': imageUrl}

result_exc = chain(celery_app_tmp.task_download.download.s(addImageInput),

random.choice(list_task).featureExtract.s())()

while True:

if result_exc.ready():

dict_exc = result_exc.get()

break

task0.py:

from celery import Celery

app = Celery()

app.config_from_object("celery_app_tmp.celeryconfig_gpu")

gpu_id = 0

@app.task

def faceRec(addImageInput, faceSetId):

do somthing()

return result_dict

task1.py is same as task0.py,just gpu_id is different.

task_download.py:

app = Celery()

app.config_from_object("celery_app_tmp.celeryconfig_download")

@app.task

def download(addImageInput):

do somthing()

return addImageInput

celery_config.py:

from kombu import Queue

from kombu import Exchange

result_serializer = 'msgpack'

task_serializer = 'msgpack'

accept_content = ['json', 'msgpack']

broker_url = "redis://:redis@172.16.3.22:7369/1"

result_backend = "redis://:redis@172.16.3.22:7369/1"

worker_concurrency = 8

result_exchange_type = 'direct'

result_expires = 5

task_queues = (

Queue('gpu_0', exchange=Exchange('gpu_0'), routing_key='gpu_0'),

Queue('gpu_1', exchange=Exchange('gpu_1'), routing_key='gpu_1'),

Queue('download', exchange=Exchange('download'), routing_key='download'),

)

task_routes = {

'celery_app_tmp.task0.faceRec': {'queue': 'gpu_0', 'routing_key': 'gpu_0'},

'celery_app_tmp.task1.faceRec': {'queue': 'gpu_1', 'routing_key': 'gpu_1'},

'celery_app_tmp.task_download.download': {'queue': 'download', 'routing_key': 'download'},

}

Expected Behavior

when jmeter stress this program, the redis connected_clients and client_recent_max_output_buffer dont continue to increase

Actual Behavior

Issue 2:
I have a situation where I have a Topic Exchange with two queues bound to the exchange. Given the nature of Topic Exchanges, it's possible for a single messager to be routed to multiple queues (this is desirable for my usage). If I have a worker processing messages on one of the queue and have to stop that worker, it will restore any unacknowledged messages. However, instead of putting the message back into the queue that it retrieved the message from, it puts the message back into the exchange. This is a problem in that it means duplicate messages in other queues.

For example, say Exchange Foo has Queue A and Queue B accepting all messages off of the exchange (routing_key='#'). If we submit Message 123 to the exchange, it will be routed to both queues. If a worker is reading off of Queue A and has to restore the unacknowledged Message 123 (for whatever reason), it will place that message back into Exchange Foo. This will place the Messager 123 back into Queue A (good!) and Queue B (bad!). Queue B now has two instances of Mesage 123.

I believe the correct behaviour should be to simply place the message back into the queue that it came from. Is there a technical limitation that requires us to do this? If not, I'd be happy to make a PR to resolve this.