<p dir="auto">In summary I have two rabbitmq instances running on the same host different ports (e.g. R1:5672,R2:6672). They are set up as a cluster and configured to mirror all queues.  I run two celery workers on the same host, each configured to connect to one of these rabbitmq instances.<br>
So environment looks like:<br>
C1-&gt;R1<br>
C2-&gt;R2<br>
and R1&lt;-&gt;R2</p>
<p dir="auto">I then publish a bunch of ETA tasks.<br>
I verify tasks are equally distributed to each celery worker as ETA tasks.<br>
Next I stop R2.</p>
<p dir="auto">At this point I observer many tasks are Received by C1 as duplicates.<br>
Log of C1 look like this:</p>
<p dir="auto">[2015-12-23 23:37:21,993: INFO/MainProcess] Received task: tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] eta:[2015-12-24 04:42:21.966493+00:00]<br>
[2015-12-23 23:37:21,995: INFO/MainProcess] Received task: tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] eta:[2015-12-24 04:42:21.966325+00:00]</p>
<p dir="auto">they are eventually executed by this celery worker twice..<br>
[2015-12-23 23:42:22,633: INFO/MainProcess] Task tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] succeeded in 0.00121066000429s: 9<br>
[2015-12-23 23:42:22,633: INFO/MainProcess] Task tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] succeeded in 0.00175269600004s: 9</p>
<p dir="auto">This is my task for reference:<br>
from celery import Celery<br>
import os<br>
app = Celery('tasks', broker='amqp://myuser:mypassword@localhost:{0}/'.format(os.environ['AMQP_PORT']))<br>
app.conf.update(<br>
CELERY_QUEUE_HA_POLICY='all',<br>
CELERY_TRACK_STARTED=True,<br>
CELERY_ACKS_LATE=True,<br>
CELERY_TASK_SERIALIZER='json',<br>
CELERY_EVENT_QUEUE_TTL=20,<br>
CELERY_SEND_TASK_SENT_EVENT=True,<br>
CELERY_EVENT_QUEUE_EXPIRES=900<br>
)</p>
<p dir="auto">@app.task<br>
def add(x, y):<br>
return x + y</p>
<p dir="auto">I start up the celery instances as:<br>
AMQP_PORT=5672 celery -A tasks worker --loglevel=info -n celery_5672<br>
AMQP_PORT=6672 celery -A tasks worker --loglevel=info -n celery_6672</p>
<p dir="auto">Rabbit policy is:<br>
/   ha-all  all ^(?!amq\.).* {"ha-mode":"all","ha-sync-mode":"automatic"}    0</p>
<p dir="auto">I create tasks as:<br>
workflow = (add.s(4,4).set(countdown=300) | add.s(1).set(countdown=300))<br>
print workflow()</p>
<p dir="auto">Wouldn't it make sense for Celery to discard duplicates when there is an ETA task scheduled already instead of scheduling a new one? My tasks are mostly(!) idempotent (we use ACKS_LATE), but still it seems that it should be easy for Celery to prevent executing these tasks multiple times.<br>
Thanks..<br>
Gökhan</p>
<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" checked=""> I have verified that the issue exists against the <code class="notranslate">master</code> branch.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> This has already been asked to the [discussion group]</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have read the relevant section in the [contribution guide].</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have checked the [issues list] for similar or identical bug reports.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have checked the [pull requests list]for existing proposed fixes.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have checked the [commit log] to find out if the bug was already fixed.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included all related issues and possible duplicate issues</li>
</ul>
<h2 dir="auto">Mandatory Debugging Information</h2>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included the output of <code class="notranslate">celery -A proj report</code> in the issue.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have verified that the issue exists against the <code class="notranslate">master</code> branch.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> 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" checked=""> I have included all the versions of all the external dependencies.</li>
</ul>
<h2 dir="auto">Optional Debugging Information</h2>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have tried reproducing the issue on more than one Python version</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</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 version of broker</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have tried reproducing the issue on more than one operating system.</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 autoscaling, retries,<br>
ETA/Countdown &amp; rate limits disabled.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue after downgrading<br>
and/or upgrading Celery and its dependencies.</li>
</ul>
<h2 dir="auto">Related Issues and Possible Duplicates</h2>
<h4 dir="auto">Related Issues</h4>
<p dir="auto"><a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="125573898" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/2991" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/2991/hovercard" href="https://github.com/celery/celery/issues/2991">#2991</a><br>
<a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="54503831" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/2457" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/2457/hovercard" href="https://github.com/celery/celery/issues/2457">#2457</a></p>
<h4 dir="auto">Possible Duplicates</h4>
<p dir="auto"><a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="261165563" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/4296" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/4296/hovercard" href="https://github.com/celery/celery/issues/4296">#4296</a></p>
<h2 dir="auto">Environment &amp; Settings</h2>
<details>
<summary><b><code class="notranslate">celery report</code> Output:</b></summary>
<p dir="auto">
</p><p dir="auto">software -&gt; celery:4.4.2 (cliffs) kombu:4.6.8 py:3.7.7<br>
billiard:3.6.3.0 py-amqp:2.5.2<br>
platform -&gt; system:Darwin arch:64bit<br>
kernel version:19.4.0 imp:CPython<br>
loader   -&gt; celery.loaders.app.AppLoader<br>
settings -&gt; transport:pyamqp results:disabled</p>
<p dir="auto">broker_url: 'amqp://guest:********<a class="user-mention notranslate" data-hovercard-type="user" data-hovercard-url="/users/localhost/hovercard" data-octo-click="hovercard-link-click" data-octo-dimensions="link_type:self" href="https://github.com/localhost">@localhost</a>:5672//'<br>
task_publish_retry: False</p>
<p dir="auto"></p>
</details>
<h1 dir="auto">Steps to Reproduce</h1>
<ol dir="auto">
<li>Shutdown broker</li>
<li>Publish a task (details in "Minimally Reproducible Test Case" section)</li>
<li>Wait indefinitely</li>
</ol>
<h2 dir="auto">Required Dependencies</h2>
<ul dir="auto">
<li><strong>Minimal Python Version</strong>: N/A</li>
<li><strong>Minimal Celery Version</strong>: N/A</li>
<li><strong>Minimal Kombu Version</strong>: N/A</li>
<li><strong>Minimal Broker Version</strong>: N/A</li>
<li><strong>Minimal Result Backend Version</strong>: N/A</li>
<li><strong>Minimal OS and/or Kernel Version</strong>: N/A</li>
<li><strong>Minimal Broker Client Version</strong>: N/A</li>
<li><strong>Minimal Result Backend Client Version</strong>: N/A</li>
</ul>
<h3 dir="auto">Python Packages</h3>
<details>
<summary><b><code class="notranslate">pip freeze</code> Output:</b></summary>
<p dir="auto">
</p><p dir="auto">amqp==2.5.2<br>
billiard==3.6.3.0<br>
celery==4.4.2<br>
docutils==0.16<br>
importlib-metadata==1.6.0<br>
kombu==4.6.8<br>
lockfile==0.12.2<br>
protobuf==3.11.4<br>
python-daemon==2.2.4<br>
pytz==2019.3<br>
six==1.14.0<br>
vine==1.3.0<br>
zipp==3.1.0</p>
<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>
<p dir="auto">Step 1: vi tasks.py</p>
<details>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from celery import Celery

class celeryconfig:
	broker_url = 'pyamqp://guest@localhost//'
	task_publish_retry = False

app = Celery('tasks')
app.config_from_object(celeryconfig)

@app.task
def add(x, y):
	return x + y"><pre class="notranslate"><code class="notranslate">from celery import Celery

class celeryconfig:
	broker_url = 'pyamqp://guest@localhost//'
	task_publish_retry = False

app = Celery('tasks')
app.config_from_object(celeryconfig)

@app.task
def add(x, y):
	return x + y
</code></pre></div>
<p dir="auto"></p>
</details>
<p dir="auto">Step 2: shutdown broker<br>
Step 3: from python prompt:</p>
<details>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="$ python
&gt;&gt;&gt; from tasks import add
&gt;&gt;&gt; add.delay(1,2)"><pre class="notranslate"><code class="notranslate">$ python
&gt;&gt;&gt; from tasks import add
&gt;&gt;&gt; add.delay(1,2)
</code></pre></div>
<p dir="auto"></p>
</details>
<h1 dir="auto">Expected Behavior</h1>
<p dir="auto">Quit process due to exception:<br>
<code class="notranslate">kombu.exceptions.OperationalError: [Errno 61] Connection refused</code></p>
<h1 dir="auto">Actual Behavior</h1>
<p dir="auto">Wait <strong>INDEFINITELY</strong></p>
<h1 dir="auto">Analysis</h1>
<ol dir="auto">
<li>
<p dir="auto">at line 870 of <code class="notranslate">celery/app/base.y</code>, parameter <code class="notranslate">transport_options</code> will only cite calling parameter of <code class="notranslate">calling _connection()</code> &amp; <code class="notranslate">conf.broker_transport_options</code>. Both<code class="notranslate">task_publish_retry</code> and <code class="notranslate">task_publish_retry_policy</code> are not included.</p>
</li>
<li>
<p dir="auto">at line 1196 of <code class="notranslate">celery/app/base.y</code>, <code class="notranslate">self._pool</code> is generated for the first time by calling <code class="notranslate">connection_for_write()</code> <strong>without any parameter</strong>. So only <code class="notranslate">conf.broker_transport_options</code> will be used for initiaizing <code class="notranslate">Connection</code> instance of <code class="notranslate">kombu</code>, both <code class="notranslate">task_publish_retry</code> and <code class="notranslate">task_publish_retry_policy</code> will be ignored.</p>
</li>
<li>
<p dir="auto">at line 559 of <code class="notranslate">celery/app/amqp.py</code>, both <code class="notranslate">task_publish_retry</code> and <code class="notranslate">task_publish_retry_policy</code> is passed to function <code class="notranslate">publish()</code> of <code class="notranslate">kombu</code>, but these two parameters will be dropped when referencing <code class="notranslate">self.channel</code> to create a connection channel at line 174 of <code class="notranslate">kombu/messaging.py</code>, which will make call to <code class="notranslate">ChannelPromise(lambda: connection.default_channel)</code> at line 224, and get result from <code class="notranslate">Connection.default_channel</code>.</p>
</li>
<li>
<p dir="auto">at line 844 of <code class="notranslate">kombu/connection.py</code> we can find that <code class="notranslate">default_channel()</code> takes no parameter at all, and will only take the initialization parameters of Collection instance. As a result, settings from <code class="notranslate">task_publish_retry</code> and <code class="notranslate">task_publish_retry_policy</code> configuration are not included, because they're not part of instance initialization as mentioned above.</p>
</li>
<li>
<p dir="auto">at line 383 of <code class="notranslate">kombu/connection.py</code> we can further inspect that the real value of parameter <code class="notranslate">max_retries</code> passed to function <code class="notranslate">ensure_connection()</code> is <strong><code class="notranslate">None</code></strong>. So it will repeat indefinitely.</p>
</li>
<li>
<p dir="auto">For the same reason, the <code class="notranslate">retry</code> and <code class="notranslate">retry_policy</code> parameter when calling <code class="notranslate">apply_async()</code> or <code class="notranslate">send_task()</code> does not work, either. Because these parameter will not reach <code class="notranslate">default_channel</code> of <code class="notranslate">kombu</code> at all.</p>
</li>
</ol>
<p dir="auto">So either the code should be fixed to include <code class="notranslate">task_publish_retry</code> and <code class="notranslate">task_publish_retry_policy</code> in <code class="notranslate">Connection</code> instance initialization, or the instance variables of <code class="notranslate">Connection</code> should be modified when calling <code class="notranslate">Producer.publish()</code>, or the document should be modified to clarify that these two configurations will not work as intended.</p>
<h1 dir="auto">Temporary Workaround</h1>
<p dir="auto">Use <code class="notranslate">broker_transport_options</code> instead of <code class="notranslate">task_publish_retry</code>/<code class="notranslate">task_publish_retry_policy</code>:</p>
<details>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="class celeryconfig:
	broker_url = 'pyamqp://guest@localhost//'
	broker_transport_options = {
		'max_retries': 1,
		'interval_start': 0,
		'interval_step': 0.2,
		'interval_max': 0.2,
	}

app.config_from_object(celeryconfig)"><pre class="notranslate"><code class="notranslate">class celeryconfig:
	broker_url = 'pyamqp://guest@localhost//'
	broker_transport_options = {
		'max_retries': 1,
		'interval_start': 0,
		'interval_step': 0.2,
		'interval_max': 0.2,
	}

app.config_from_object(celeryconfig)
</code></pre></div>
<p dir="auto"></p>
</details>
<p dir="auto">Didn't find a way to turn off retries, though.</p>
<h2 dir="auto">[EDIT]</h2>
<p dir="auto">Add possible duplicate through Googling: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="261165563" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/4296" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/4296/hovercard" href="https://github.com/celery/celery/issues/4296">#4296</a></p>
<p dir="auto">The conclusion is identical to my post, use <code class="notranslate">broker_transport_options</code> instead. This problem (or inconsistency of documentation) was reported back in 2018, but the situation still remains in v4.4.2.</p>
<p dir="auto">The "Calling Task" section of document should really be modified to clarify that all <code class="notranslate">task_publish_retry</code>, <code class="notranslate">task_publish_retry_policy</code>, <code class="notranslate">retry</code> and <code class="notranslate">retry_policy</code> will not work as intended when connecting to broker for the first time.</p>