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

So environment looks like:

C1->R1

C2->R2

and R1<->R2

I then publish a bunch of ETA tasks.

I verify tasks are equally distributed to each celery worker as ETA tasks.

Next I stop R2.

At this point I observer many tasks are Received by C1 as duplicates.

Log of C1 look like this:

[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]

[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]

they are eventually executed by this celery worker twice..

[2015-12-23 23:42:22,633: INFO/MainProcess] Task tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] succeeded in 0.00121066000429s: 9

[2015-12-23 23:42:22,633: INFO/MainProcess] Task tasks.add[5c7e4b80-be18-4d8d-965e-534d75dfffbc] succeeded in 0.00175269600004s: 9

This is my task for reference:

from celery import Celery

import os

app = Celery('tasks', broker='amqp://myuser:mypassword@localhost:{0}/'.format(os.environ['AMQP_PORT']))

app.conf.update(

CELERY_QUEUE_HA_POLICY='all',

CELERY_TRACK_STARTED=True,

CELERY_ACKS_LATE=True,

CELERY_TASK_SERIALIZER='json',

CELERY_EVENT_QUEUE_TTL=20,

CELERY_SEND_TASK_SENT_EVENT=True,

CELERY_EVENT_QUEUE_EXPIRES=900

)

@app.task

def add(x, y):

return x + y

I start up the celery instances as:

AMQP_PORT=5672 celery -A tasks worker --loglevel=info -n celery_5672

AMQP_PORT=6672 celery -A tasks worker --loglevel=info -n celery_6672

Rabbit policy is:

/ ha-all all ^(?!amq\.).* {"ha-mode":"all","ha-sync-mode":"automatic"} 0

I create tasks as:

workflow = (add.s(4,4).set(countdown=300) | add.s(1).set(countdown=300))

print workflow()

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.

Thanks..

Gökhan

Issue 2:
Checklist

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

This has already been asked to the [discussion group]

I have read the relevant section in the [contribution guide].

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.

I have included all related issues and possible duplicate issues

Mandatory Debugging Information

I have included the output of celery -A proj report in the issue.

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

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

I have included all the versions of all the external dependencies.

Optional Debugging Information

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

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

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

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

#2991

#2457

Possible Duplicates

#4296

Environment & Settings

celery report Output:

software -> celery:4.4.2 (cliffs) kombu:4.6.8 py:3.7.7

billiard:3.6.3.0 py-amqp:2.5.2

platform -> system:Darwin arch:64bit

kernel version:19.4.0 imp:CPython

loader -> celery.loaders.app.AppLoader

settings -> transport:pyamqp results:disabled

broker_url: 'amqp://guest:********@localhost:5672//'

task_publish_retry: False

Steps to Reproduce

Shutdown broker

Publish a task (details in "Minimally Reproducible Test Case" section)

Wait indefinitely

Required Dependencies

Minimal Python Version: N/A

Minimal Celery Version: N/A

Minimal Kombu Version: N/A

Minimal Broker Version: N/A

Minimal Result Backend Version: N/A

Minimal OS and/or Kernel Version: N/A

Minimal Broker Client Version: N/A

Minimal Result Backend Client Version: N/A

Python Packages

pip freeze Output:

amqp==2.5.2

billiard==3.6.3.0

celery==4.4.2

docutils==0.16

importlib-metadata==1.6.0

kombu==4.6.8

lockfile==0.12.2

protobuf==3.11.4

python-daemon==2.2.4

pytz==2019.3

six==1.14.0

vine==1.3.0

zipp==3.1.0

Other Dependencies

N/A

Minimally Reproducible Test Case

Step 1: vi tasks.py

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

Step 2: shutdown broker

Step 3: from python prompt:

$ python
>>> from tasks import add
>>> add.delay(1,2)

Expected Behavior

Quit process due to exception:

kombu.exceptions.OperationalError: [Errno 61] Connection refused

Actual Behavior

Wait INDEFINITELY

Analysis

at line 870 of celery/app/base.y, parameter transport_options will only cite calling parameter of calling _connection() & conf.broker_transport_options. Bothtask_publish_retry and task_publish_retry_policy are not included.

at line 1196 of celery/app/base.y, self._pool is generated for the first time by calling connection_for_write() without any parameter. So only conf.broker_transport_options will be used for initiaizing Connection instance of kombu, both task_publish_retry and task_publish_retry_policy will be ignored.

at line 559 of celery/app/amqp.py, both task_publish_retry and task_publish_retry_policy is passed to function publish() of kombu, but these two parameters will be dropped when referencing self.channel to create a connection channel at line 174 of kombu/messaging.py, which will make call to ChannelPromise(lambda: connection.default_channel) at line 224, and get result from Connection.default_channel.

at line 844 of kombu/connection.py we can find that default_channel() takes no parameter at all, and will only take the initialization parameters of Collection instance. As a result, settings from task_publish_retry and task_publish_retry_policy configuration are not included, because they're not part of instance initialization as mentioned above.

at line 383 of kombu/connection.py we can further inspect that the real value of parameter max_retries passed to function ensure_connection() is None. So it will repeat indefinitely.

For the same reason, the retry and retry_policy parameter when calling apply_async() or send_task() does not work, either. Because these parameter will not reach default_channel of kombu at all.

So either the code should be fixed to include task_publish_retry and task_publish_retry_policy in Connection instance initialization, or the instance variables of Connection should be modified when calling Producer.publish(), or the document should be modified to clarify that these two configurations will not work as intended.

Temporary Workaround

Use broker_transport_options instead of task_publish_retry/task_publish_retry_policy:

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)

Didn't find a way to turn off retries, though.

[EDIT]

Add possible duplicate through Googling: #4296

The conclusion is identical to my post, use broker_transport_options instead. This problem (or inconsistency of documentation) was reported back in 2018, but the situation still remains in v4.4.2.

The "Calling Task" section of document should really be modified to clarify that all task_publish_retry, task_publish_retry_policy, retry and retry_policy will not work as intended when connecting to broker for the first time.