Issue 1:
Checklist

I have read the relevant section in the

contribution guide

on reporting bugs.

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 in the master branch.

I have included all related issues and possible duplicate issues

in this issue (If there are none, check this box anyway).

Mandatory Debugging Information

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

(if you are not able to do this, then at least specify the Celery

version affected).

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

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

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

to reproduce this bug.

Optional Debugging Information

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

and/or implementation.

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

result backend.

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

broker and/or result backend.

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

#5358

#4876

Possible Duplicates

None

Environment & Settings

Celery version: 4.3.0 (rhubarb)

celery report Output:

software -> celery:4.3.0 (rhubarb) kombu:4.5.0 py:2.7.13
billiard:3.6.0.0 py-amqp:2.4.2
platform -> system:Linux arch:64bit
kernel version:4.9.0-8-amd64 imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:amqp results:rpc:///

task_queues:
(<unbound Queue gpon_tentacle -> <unbound Exchange celery(direct)> -> >,
<unbound Queue livestatus -> <unbound Exchange celery(direct)> -> >,
<unbound Queue kraken_host_events -> <unbound Exchange celery(direct)> -> >)
broker_heartbeat: 5
broker_connection_timeout: 10
task_default_exchange_type: 'direct'
broker_pool_limit: 1
task_routes: {
'gpon_tentacle.*': { 'queue': 'gpon_tentacle'},
'kraken_host_events.*': { 'queue': 'kraken_host_events'},
'livestatus.*': { 'queue': 'livestatus'}}
broker_connection_max_retries: 0
result_expires: 3600
broker_url: u'amqp://guest:********@localhost:5672//'
result_backend: u'rpc:///'
task_default_exchange: 'celery'

Steps to Reproduce

See this gist: https://gist.github.com/palvarezcordoba/a88e17e79f3efa8f15c0094afb4a2ec2

Required Dependencies

Minimal Python Version: N/A or Unknown

Minimal Celery Version: N/A or Unknown

Minimal Kombu Version: N/A or Unknown

Minimal Broker Version: N/A or Unknown

Minimal Result Backend Version: N/A or Unknown

Minimal OS and/or Kernel Version: N/A or Unknown

Minimal Broker Client Version: N/A or Unknown

Minimal Result Backend Client Version: N/A or Unknown

Python Packages

pip freeze Output:

amqp==2.4.2
astroid==1.6.5
asyncio==3.4.3
atomicwrites==1.3.0
attrs==19.1.0
backports.functools-lru-cache==1.5
backports.shutil-get-terminal-size==1.0.0
billiard==3.6.0.0
bitstring==3.1.5
caniusepython3==7.0.0
celery==4.3.0
celery-slack==0.3.0
certifi==2019.3.9
chardet==3.0.4
Click==7.0
configparser==3.7.1
decorator==4.3.2
dill==0.2.5
distlib==0.2.8
Django==1.11.20
django-appconf==1.0.3
django-betterforms==1.2
django-filter==1.1.0
django-multiselectfield==0.1.8
django-rq==1.3.1
Django-Select2==5.11.1
django-slack==5.13.0
djangorestframework==3.9.2
enum34==1.1.6
ephem==3.7.6.0
filelock==3.0.10
funcsigs==1.0.2
future==0.17.1
futures==3.2.0
googlemaps==3.0.2
gpon-tentacle==0.0.0
idna==2.8
ipython==5.0.0
ipython-genutils==0.2.0
isort==4.3.4
kombu==4.5.0
kraken==0.0.2
krill-gpon==0.5.0
lazy-object-proxy==1.3.1
libkrill==1.1.6
mccabe==0.6.1
mock==2.0.0
more-itertools==5.0.0
MySQL-python==1.2.5
mysqlclient==1.4.2.post1
netaddr==0.7.19
numpy==1.16.1
packaging==19.0
pathlib2==2.3.3
pbr==5.1.3
pexpect==4.6.0
pickleshare==0.7.5
pika==0.13.0
pluggy==0.9.0
ply==3.11
prompt-toolkit==1.0.15
psutil==5.5.0
ptyprocess==0.6.0
py==1.8.0
pyasn1==0.1.9
pycrypto==2.6.1
pycurl==7.43.0.2
pydot==1.4.1
PyFunctional==1.2.0
Pygments==2.3.1
pylint==1.9.4
pyparsing==2.3.1
pysmi==0.3.3
pysnmp==4.3.2
pytest==4.4.0
pytz==2019.1
PyYAML==5.1
recordclass==0.9
recordtype==1.3
redis==3.2.1
redlock==1.2.0
requests==2.21.0
rq==0.13.0
scandir==1.9.0
Shinken==2.4.3
simplegeneric==0.8.1
singledispatch==3.4.0.3
singleton-decorator==1.0.0
six==1.12.0
tabulate==0.8.3
termcolor==1.1.0
toml==0.10.0
tox==3.8.4
traitlets==4.3.2
typing==3.6.6
urllib3==1.24.1
urlparse2==1.1.1
vine==1.3.0
virtualenv==16.4.3
wcwidth==0.1.7
wrapt==1.11.1

Other Dependencies

N/A

Minimally Reproducible Test Case

https://gist.github.com/palvarezcordoba/a88e17e79f3efa8f15c0094afb4a2ec2

Expected Behavior

Celery client should send tasks, and after >60 seconds of inactivity, it should be able to send one more task and get the result.

Actual Behavior

When I send a task by first time, Celery connects to the broker, with a hertbeat conf of 60s.

It declares a random result quue, say QueueX, then consumes it and sends a task, saiyng "eh, reply to QueueX". The worker replies to QueueX and RabbitMQ does a Basic.Deliver as the client is consumer of QueueX.

Celery does not send heartbeats, and after 60 seconds (or configured heartbeat) RabbitMQ closes the connection, and as QueueX has auto-delete active, RabbitMQ deletes it. At this point, if I send other task, - I do not why but... - celery sends the task, the worker receives it, but celery raises error: [Errno 104] Connection reset by peer. If I send another task, celery redeclares QueueX, sends a task which the reply-to being QueueX BUT now celery is not a consumer of QueueX so RabbitMQ does not do a deliver, so celery waits until a timeout.

RabbitMQ send heartbeat frames to celery, it ignores (I guess) those packets, but at TCP level they are replied with an ACK, I do not know if it is correct.

A solution (which I do not like) is setting heartbeat at 0.

In an ideal world, celery would send heartbeats, or if broker closes the connection, it would:

Not send a task whose result can not be received

Detect the network failure

(again) Not send a task whose result can not be received

Reconnect to the broker

Declare QueueX and consume it

Now it can send the task :)

Issue 2:
Currently, the Redis and SQS transports provide a visibility_timeout option in the config. This has an effect on how apply_async() works with either eta or countdown kwargs. The caveats regarding this are buried in:

http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#redis-caveats

http://docs.celeryproject.org/en/latest/getting-started/brokers/sqs.html

Using etas or countdowns greater than the visibility_timeout value will result in tasks being duplicated. There should be some mention of this and/or links to these in:

http://docs.celeryproject.org/en/latest/userguide/calling.html#eta-and-countdown