Both 'task_publish_retry' and 'task_publish_retry_policy' are ignored when publishing tasks

[ulyssesric]

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

1. Shutdown broker
2. Publish a task (details in "Minimally Reproducible Test Case" section)
3. 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

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

2. 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.

3. 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.

4. 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.

5. 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.

6. 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.