<h1 dir="auto">Checklist</h1>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have included the output of <code class="notranslate">celery -A proj report</code> in the issue.<br>
(if you are not able to do this, then at least specify the Celery<br>
version affected).</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have included all related issues and possible duplicate issues in this issue.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have included the contents of <code class="notranslate">pip freeze</code> in the issue.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have verified that the issue exists against the <code class="notranslate">master</code> branch of Celery.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one message broker and/or result backend.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one workers pool.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue with retries, ETA/Countdown &amp; rate limits disabled.</li>
</ul>
<h2 dir="auto">Related Issues and Possible Duplicates</h2>

<h4 dir="auto">Related Issues</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h4 dir="auto">Possible Duplicates</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h2 dir="auto">Environment &amp; Settings</h2>

<p dir="auto"><strong>Celery version</strong>:4.2.1</p>

<details>
<summary><b><code class="notranslate">celery report</code> Output:</b></summary>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=""><pre class="notranslate"><code class="notranslate"></code></pre></div>
<p dir="auto"></p>
</details>
<h1 dir="auto">Steps to Reproduce</h1>
<h2 dir="auto">Required Dependencies</h2>

<ul dir="auto">
<li><strong>Minimal Python Version</strong>:2.7</li>
<li><strong>Minimal Broker Version</strong>: redis 5.0.3</li>
<li><strong>Minimal Result Backend Version</strong>:redis 5.0.3</li>
<li><strong>Minimal OS and/or Kernel Version</strong>: ubuntu 16.04</li>
<li><strong>Minimal Broker Client Version</strong>: N/A or Unknown</li>
<li><strong>Minimal Result Backend Client Version</strong>: N/A or Unknown</li>
</ul>
<h3 dir="auto">Python Packages</h3>

<details>
<summary><b><code class="notranslate">pip freeze</code> Output:</b></summary>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=""><pre class="notranslate"><code class="notranslate"></code></pre></div>
<p dir="auto"></p>
</details>
<h3 dir="auto">Other Dependencies</h3>

<details>
<p dir="auto">
N/A
</p>
</details>
<h2 dir="auto">Minimally Reproducible Test Case</h2>

<details>
<p dir="auto">
main.python:
</p><p dir="auto">import celery_app_tmp.task0<br>
import celery_app_tmp.task1<br>
import celery_app_tmp.task_download</p>
<p dir="auto">list_task = [celery_app_tmp.task0,<br>
celery_app_tmp.task1]<br>
addImageInput = {'imageMD5': 'default',<br>
'imageUrl': imageUrl}<br>
result_exc = chain(celery_app_tmp.task_download.download.s(addImageInput),<br>
random.choice(list_task).featureExtract.s())()<br>
while True:<br>
if result_exc.ready():<br>
dict_exc = result_exc.get()<br>
break</p>
<p dir="auto">task0.py:</p>
<p dir="auto">from celery import Celery</p>
<p dir="auto">app = Celery()<br>
app.config_from_object("celery_app_tmp.celeryconfig_gpu")<br>
gpu_id = 0<br>
@app.task<br>
def faceRec(addImageInput, faceSetId):<br>
do somthing()<br>
return result_dict</p>
<p dir="auto">task1.py is same as task0.py,just gpu_id is different.</p>
<p dir="auto">task_download.py:<br>
app = Celery()<br>
app.config_from_object("celery_app_tmp.celeryconfig_download")</p>
<p dir="auto">@app.task<br>
def download(addImageInput):<br>
do somthing()<br>
return addImageInput</p>
<p dir="auto">celery_config.py:<br>
from kombu import Queue<br>
from kombu import Exchange</p>
<p dir="auto">result_serializer = 'msgpack'<br>
task_serializer = 'msgpack'<br>
accept_content = ['json', 'msgpack']</p>
<p dir="auto">broker_url = "redis://:redis@172.16.3.22:7369/1"<br>
result_backend = "redis://:redis@172.16.3.22:7369/1"</p>
<p dir="auto">worker_concurrency = 8<br>
result_exchange_type = 'direct'<br>
result_expires = 5</p>
<p dir="auto">task_queues = (<br>
Queue('gpu_0', exchange=Exchange('gpu_0'), routing_key='gpu_0'),<br>
Queue('gpu_1', exchange=Exchange('gpu_1'), routing_key='gpu_1'),<br>
Queue('download', exchange=Exchange('download'), routing_key='download'),<br>
)</p>
<p dir="auto">task_routes = {<br>
'celery_app_tmp.task0.faceRec': {'queue': 'gpu_0', 'routing_key': 'gpu_0'},<br>
'celery_app_tmp.task1.faceRec': {'queue': 'gpu_1', 'routing_key': 'gpu_1'},<br>
'celery_app_tmp.task_download.download': {'queue': 'download', 'routing_key': 'download'},</p>
<p dir="auto">}</p>
<p dir="auto"></p>
</details>
<h1 dir="auto">Expected Behavior</h1>
<p dir="auto">when jmeter stress this program, the redis connected_clients and client_recent_max_output_buffer dont continue to increase</p>
<h1 dir="auto">Actual Behavior</h1>
<p dir="auto">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.</p>
<p dir="auto">For example, say Exchange Foo has Queue A and Queue B accepting all messages off of the exchange (<code class="notranslate">routing_key='#'</code>). 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.</p>
<p dir="auto">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.</p>