<p dir="auto">Currently using Celery 4.1.0</p>
<h2 dir="auto">Steps to reproduce</h2>
<p dir="auto">Start a new project using RabbitMQ and register the following task:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from django.core.cache import cache

@shared_task(bind=True)
def test_1(self):
    if not cache.add(self.request.id, 1):
        raise Exception('Duplicated task {}'.format(self.request.id))"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">django</span>.<span class="pl-s1">core</span>.<span class="pl-s1">cache</span> <span class="pl-k">import</span> <span class="pl-s1">cache</span>

<span class="pl-en">@<span class="pl-en">shared_task</span>(<span class="pl-s1">bind</span><span class="pl-c1">=</span><span class="pl-c1">True</span>)</span>
<span class="pl-k">def</span> <span class="pl-en">test_1</span>(<span class="pl-s1">self</span>):
    <span class="pl-k">if</span> <span class="pl-c1">not</span> <span class="pl-s1">cache</span>.<span class="pl-en">add</span>(<span class="pl-s1">self</span>.<span class="pl-s1">request</span>.<span class="pl-s1">id</span>, <span class="pl-c1">1</span>):
        <span class="pl-k">raise</span> <span class="pl-v">Exception</span>(<span class="pl-s">'Duplicated task {}'</span>.<span class="pl-en">format</span>(<span class="pl-s1">self</span>.<span class="pl-s1">request</span>.<span class="pl-s1">id</span>))</pre></div>
<p dir="auto">Now start 2 workers. I used gevent with a concurrency of 25 for this test:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="celery worker -A my_proj -Q my_queue -P gevent -c 25"><pre class="notranslate"><code class="notranslate">celery worker -A my_proj -Q my_queue -P gevent -c 25
</code></pre></div>
<p dir="auto">Open a python shell and fire a a bunch of tasks:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from myproj.tasks import test_1

for i in range(10000):
    test_1.apply_async()"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">myproj</span>.<span class="pl-s1">tasks</span> <span class="pl-k">import</span> <span class="pl-s1">test_1</span>

<span class="pl-k">for</span> <span class="pl-s1">i</span> <span class="pl-c1">in</span> <span class="pl-en">range</span>(<span class="pl-c1">10000</span>):
    <span class="pl-s1">test_1</span>.<span class="pl-en">apply_async</span>()</pre></div>
<p dir="auto">Now quickly do a warm shutdown (Ctrl+c) in one of the workers while it's still processing the tasks, you should see the errors popping in the second worker:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="ERROR    Task my_proj.tasks.test_1[e28e6760-1371-49c9-af87-d196c59375e9] raised unexpected: Exception('Duplicated task e28e6760-1371-49c9-af87-d196c59375e9',)
Traceback (most recent call last):
  File &quot;/code/virtualenv/CURRENT/lib/python3.5/site-packages/celery/app/trace.py&quot;, line 374, in trace_task
    R = retval = fun(*args, **kwargs)
  File &quot;/code/virtualenv/CURRENT/lib/python3.5/site-packages/celery/app/trace.py&quot;, line 629, in __protected_call__
    return self.run(*args, **kwargs)
  File &quot;/code/scp/python/my_proj/tasks.py&quot;, line 33, in test_1
    raise Exception('Duplicated task {}'.format(self.request.id))
Exception: Duplicated task e28e6760-1371-49c9-af87-d196c59375e9"><pre class="notranslate"><code class="notranslate">ERROR    Task my_proj.tasks.test_1[e28e6760-1371-49c9-af87-d196c59375e9] raised unexpected: Exception('Duplicated task e28e6760-1371-49c9-af87-d196c59375e9',)
Traceback (most recent call last):
  File "/code/virtualenv/CURRENT/lib/python3.5/site-packages/celery/app/trace.py", line 374, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/code/virtualenv/CURRENT/lib/python3.5/site-packages/celery/app/trace.py", line 629, in __protected_call__
    return self.run(*args, **kwargs)
  File "/code/scp/python/my_proj/tasks.py", line 33, in test_1
    raise Exception('Duplicated task {}'.format(self.request.id))
Exception: Duplicated task e28e6760-1371-49c9-af87-d196c59375e9
</code></pre></div>
<h2 dir="auto">Expected behavior</h2>
<p dir="auto">Since I am not using late acknowledgment and I am not killing the workers I wasn't expecting the tasks to execute again.</p>
<h2 dir="auto">Actual behavior</h2>
<p dir="auto">The tasking are being executed twice, this is causing some problems in our servers because we restart our works every 15 minutes or so in order to avoid memory leaks.</p>
<p dir="auto">def reset_connection(self):<br>
"""Re-establish the broker connection and set up consumers,<br>
heartbeat and the event dispatcher."""<br>
######################  Added code   ###########################################<br>
listening_queues = self.task_consumer.queues and self.task_consumer.queues or None</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    .......

    self.task_consumer = self.app.amqp.TaskConsumer(self.connection, queues=listening_queues,
                                on_decode_error=self.on_decode_error)"><pre class="notranslate"><code class="notranslate">    .......

    self.task_consumer = self.app.amqp.TaskConsumer(self.connection, queues=listening_queues,
                                on_decode_error=self.on_decode_error)
</code></pre></div>
<p dir="auto">I read the logic of reset_connection, find out the consumer only recover the queue sent to it when initializing.<br>
But the queue added with celery.control.broadcast("add_task_queue") will be lost forever.</p>