Issue 1:
Checklist

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

This has already been asked to the discussions forum first.

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

None

Possible Duplicates

None

Environment & Settings

Celery version:

celery report Output:

#celery report

software -> celery:3.1.26.post2 (Cipater) kombu:3.0.37 py:3.6.4
billiard:3.3.0.23 py-amqp:1.4.9
platform -> system:Linux arch:64bit, ELF imp:CPython
loader -> celery.loaders.default.Loader
settings -> transport:amqp results:disabled

Steps to Reproduce

Required Dependencies

Minimal Python Version: 3.6.4

Minimal Celery Version: 3.1.26.post2

Minimal Kombu Version: 3.0.37

Minimal Broker Version: redis5

Minimal Result Backend Version:

Minimal OS and/or Kernel Version: centos7

Minimal Broker Client Version: N/A or Unknown

Minimal Result Backend Client Version: N/A or Unknown

version info

celery 3.1.26.post2

celery-once 3.0.1

django-celery 3.3.0

Python Packages

Output:
```shelll
#python3 -m pip list
Package Version
---------------------------------- ------------
aenum 2.2.3
amqp 1.4.9
anyjson 0.3.3
APScheduler 3.5.3
asn1crypto 0.24.0
Babel 2.7.0
backcall 0.2.0
backports.lzma 0.0.14
bcrypt 3.1.7
billiard 3.3.0.23
bitarray 1.5.3
cassandra-driver 3.19.0
celery 3.1.26.post2
celery-once 3.0.1
certifi 2021.5.30
cffi 1.14.6
chardet 3.0.4
charset-normalizer 2.0.3
clickhouse-driver 0.2.1
coreapi 2.3.3
coreschema 0.0.4
crcmod 1.7
cryptography 3.4.7
cycler 0.10.0
dateparser 1.0.0
decorator 5.0.9
Django 2.0.1
django-cas-ng 3.6.0
django-celery 3.3.0
django-cors-headers 2.1.0
django-extensions 3.1.0
django-redis 4.10.0
django-werkzeug-debugger-runserver 0.3.1
djangorestframework 3.7.7
djangorestframework-jwt 1.11.0
docopt 0.6.2
drf-yasg 1.16.0
elasticsearch 7.13.4
enum34 1.1.6
et-xmlfile 1.1.0
flower 0.9.3
future 0.17.1
gevent 1.3.4
greenlet 1.1.0
gremlinpython 3.4.4
gunicorn 19.8.1
happybase 1.2.0
idna 3.2
importlib-metadata 4.6.1
impyla 0.16.2
inflection 0.5.1
ipython 7.16.1
ipython-genutils 0.2.0
isodate 0.6.0
itypes 1.2.0
jdcal 1.4
jedi 0.17.2
Jinja2 3.0.1
jmespath 0.10.0
joblib 0.16.0
kiwisolver 1.2.0
kombu 3.0.37
ldap3 2.4.1
lxml 4.2.5
MarkupSafe 2.0.1
matplotlib 3.2.1
mybatis-mapper2sql 0.1.8
Naked 0.1.31
numpy 1.19.5
odps 3.5.1
openpyxl 3.0.7
oss2 2.5.0
pandas 1.1.5
paramiko 2.7.2
parso 0.7.1
pexpect 4.8.0
phoenixdb 0.7
pickleshare 0.7.5
Pillow 7.2.0
pip 20.2.3
ply 3.11
prettytable 0.7.2
prompt-toolkit 3.0.19
protobuf 3.17.3
psutil 5.8.0
ptyprocess 0.7.0
pyasn1 0.4.8
pycparser 2.20
pycryptodome 3.10.4
Pygments 2.9.0
PyJWT 1.6.4
pymongo 3.7.2
PyMySQL 0.9.2
PyNaCl 1.4.0
pyodps 0.10.7
pyOpenSSL 19.1.0
pyparsing 2.4.7
pypinyin 0.42.0
python-cas 1.4.0
python-dateutil 2.8.2
python-docx 0.8.6
pytz 2021.1
PyYAML 5.4.1
redis 2.10.6
regex 2021.7.6
requests 2.26.0
ruamel.yaml 0.17.21
ruamel.yaml.clib 0.2.6
scikit-learn 0.23.1
scipy 1.5.1
setuptools 28.8.0
shellescape 3.8.1
simplejson 3.15.0
six 1.16.0
sklearn 0.0
SQLAlchemy 1.4.20
sqlparse 0.2.4
sshtunnel 0.4.0
tablestore 5.1.0
threadpoolctl 2.1.0
thrift 0.13.0
thriftpy2 0.4.5
tornado 4.5.3
traitlets 4.3.3
typing-extensions 3.10.0.0
tzlocal 2.1
uritemplate 4.1.1
urllib3 1.26.6
wcwidth 0.2.5
Werkzeug 1.0.1
xlrd 2.0.1
XlsxWriter 1.2.7
xlwt 1.3.0
xpinyin 0.7.6
zipp 3.5.0
```

Other Dependencies

Redis5.0

Minimally Reproducible Test Case

### my celery task like below
```python
from celery import task

def execuet_job():

result = execute_celery.apply_async(params)

@task(base=QueueOnce, once={'graceful': True})

def execute_celery(params):

do_some_job()

</p>
start celery with the following command
```shell
celery multi start proj -A settings worker -l info --statedb=/data/celery/working1.state --logfile=/data/celery/worker.log --pidfile=/data/celery/worker1.pid -n worker1.%h -Q worker1_queue

### my celery config
```python
import os
from celery import Celery

app = Celery('src')

app.config_from_object('django.conf:settings')

app.conf.ONCE = {

'backend': 'celery_once.backends.Redis',

'settings': {

'url': 'redis://:%s@%s:%s/3' % (

redis_password, redis_address, redis_port),

'default_timeout': 172800

}

}

### my django project settings like below
```python
import os
import datetime
from libs import util
import djcelery
from kombu import Exchange, Queue
import sys

CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://' + redis_conf.redis_address + ':' + redis_conf.redis_port,
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {"max_connections": 100},
"PASSWORD": redis_conf.redis_password,
},
},
}

REST_FRAMEWORK = {
'DEFAULT_PERMISSION_CLASSES': (
'rest_framework.permissions.IsAuthenticated',
),
'DEFAULT_AUTHENTICATION_CLASSES': (
'rest_framework_jwt.authentication.JSONWebTokenAuthentication',
),
}
JWT_AUTH = {
'JWT_RESPONSE_PAYLOAD_HANDLER':
'rest_framework_jwt.utils.jwt_response_payload_handler',
'JWT_EXPIRATION_DELTA': datetime.timedelta(seconds=3000000),
}

djcelery.setup_loader()
CELERYD_FORCE_EXECV = True
CELERYD_CONCURRENCY = 4
CELERY_ACKS_LATE = False
CELERYD_MAX_TASKS_PER_CHILD = 200
CELERYD_TASK_TIME_LIMIT = 172800
BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 31536000}
CELERY_DISABLE_RATE_LIMITS = True

CELERY_IMPORTS = (
'proj.tasks',
'proj.archiver'
)

CELERY_TIMEZONE = 'Asia/Shanghai'

BROKER_BACKEND = 'redis'
BROKER_URL = 'redis://:%s@%s:%s/1' % (
redis_conf.redis_password, redis_conf.redis_address, redis_conf.redis_port)
CELERY_RESULT_BACKEND = 'redis://:%s@%s:%s/2' % (
redis_conf.redis_password, redis_conf.redis_address, redis_conf.redis_port)

CELERY_QUEUES = (
Queue("default", Exchange("default"), routing_key="default"),
Queue("worker1_queue", Exchange("worker1_queue"), routing_key="worker1_router"),
Queue("worker2_queue", Exchange("worker2_queue"), routing_key="worker2_router"),
Queue("worker3_queue", Exchange("worker3_queue"), routing_key="worker3_router"),
Queue("worker4_queue", Exchange("worker4_queue"), routing_key="worker4_router"),
Queue("worker5_queue", Exchange("worker5_queue"), routing_key="worker5_router")
)

CELERY_ROUTES = {
'proj.tasks.mytask_execute_celery1': {"queue": "worker1_queue", "routing_key": "worker1_router"},
'proj.tasks.mytask_execute_celery2': {"queue": "worker2_queue", "routing_key": "worker2_router"},
'proj.tasks.mytask_execute_celery3': {"queue": "worker3_queue", "routing_key": "worker3_router"},
'proj.tasks.mytask_execute_celery4': {"queue": "worker4_queue", "routing_key": "worker4_router"},
'proj.tasks.work_execute_celery': {"queue": "worker5_queue", "routing_key": "worker5_router"},
'proj.archiver.newwork_execute_celery': {"queue": "worker5_queue", "routing_key": "worker5_router"},
'proj.tasks.binlog_analyse_celery': {"queue": "worker5_queue", "routing_key": "worker5_router"}
}

Expected Behavior

except work execute succes

Actual Behavior

I submit task to celery,I saw the celery log,task was received

Received task: tasks.execute_celery[b664cb23-5c80-46c7-9705-28295cd4431f]

however,the task was not execute,with celery state RECEIVED, what's the problem with celery?

Issue 2:
hacked in kombu/asynchronous/hub.py:Hub.close to edit debug log, related code:

https://github.com/celery/kombu/blob/master/kombu/asynchronous/hub.py#L247-L263

def close(self, *args):
[self._unregister(fd) for fd in self.readers]
self.readers.clear()
[self._unregister(fd) for fd in self.writers]
self.writers.clear()
self.consolidate.clear()
self._close_poller()
for callback in self.on_close:
callback(self)

# Complete remaining todo before Hub close
# Eg: Acknowledge message
# To avoid infinite loop where one of the callables adds items
# to self._ready (via call_soon or otherwise).
# we create new list with current self._ready
import inspect
print(inspect.stack())
for item in list(self._ready):
item()

raw log and logic flow

=>start long time running task and send signal to warm shutdown...

smite-ThinkPad-X270-W10DG 2019-11-20 18:40:24,127 5325 INFO tasks[85] in loop 1
smite-ThinkPad-X270-W10DG 2019-11-20 18:40:34,130 5325 INFO tasks[85] in loop 2
smite-ThinkPad-X270-W10DG 2019-11-20 18:40:44,133 5325 INFO tasks[85] in loop 3
smite-ThinkPad-X270-W10DG 2019-11-20 18:40:54,135 5325 INFO trace[124] Task horand.product_audit.tasks.dispatch_request[4136e4b7-5560-4cec-94a9-6e607da7f0db] succeeded in 30.079765408998355s: None

2019-11-20 18:40:54,257 5325 WARNING log[235] [
FrameInfo(frame=<frame object at 0x7f88806ce9d0>, filename='/usr/local/lib/python3.6/dist-packages/kombu/asynchronous/hub.py', lineno=264, function='close', code_context=[' print(inspect.stack())\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88806d6be0>, filename='/usr/local/lib/python3.6/dist-packages/kombu/asynchronous/hub.py', lineno=112, function='reset', code_context=[' self.close()\n'], index=0),
FrameInfo(frame=<frame object at 0x29e50e8>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/loops.py', lineno=96, function='asynloop', code_context=[' hub.reset()\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88807aac50>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/consumer/consumer.py', lineno=596, function='start', code_context=[' c.loop(*c.loop_args())\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88807db220>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=119, function='start', code_context=[' step.start(parent)\n'], index=0),
FrameInfo(frame=<frame object at 0x295e4e8>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/consumer/consumer.py', lineno=318, function='start', code_context=[' blueprint.start(self)\n'], index=0),
FrameInfo(frame=<frame object at 0x7f888082a708>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=369, function='start', code_context=[' return self.obj.start()\n'], index=0),
FrameInfo(frame=<frame object at 0x7f888093ada0>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=119, function='start', code_context=[' step.start(parent)\n'], index=0),
FrameInfo(frame=<frame object at 0x28f0698>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/worker.py', lineno=205, function='start', code_context=[' self.blueprint.start(self)\n'], index=0),
FrameInfo(frame=<frame object at 0x1be46e8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/worker.py', lineno=258, function='run', code_context=[' worker.start()\n'], index=0),
FrameInfo(frame=<frame object at 0x1be22f8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/base.py', lineno=252, function='__call__', code_context=[' ret = self.run(*args, **kwargs)\n'], index=0),
FrameInfo(frame=<frame object at 0x18be4a8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/worker.py', lineno=223, function='run_from_argv', code_context=[' return self(*args, **options)\n'], index=0),
FrameInfo(frame=<frame object at 0x1baa718>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=420, function='execute', code_context=[' ).run_from_argv(self.prog_name, argv[1:], command=argv[0])\n'], index=0),
FrameInfo(frame=<frame object at 0x1b3ed78>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=488, function='handle_argv', code_context=[' return self.execute(command, argv)\n'], index=0),
FrameInfo(frame=<frame object at 0x1a894f8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/base.py', lineno=298, function='execute_from_commandline', code_context=[' return self.handle_argv(self.prog_name, argv[1:])\n'], index=0),
FrameInfo(frame=<frame object at 0x1ac5288>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=496, function='execute_from_commandline', code_context=[' super(CeleryCommand, self).execute_from_commandline(argv)))\n'], index=0),
FrameInfo(frame=<frame object at 0x1a13388>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=322, function='main', code_context=[' cmd.execute_from_commandline(argv)\n'], index=0),
FrameInfo(frame=<frame object at 0x183db88>, filename='/usr/local/lib/python3.6/dist-packages/celery/__main__.py', lineno=16, function='main', code_context=[' _main()\n'], index=0),
FrameInfo(frame=<frame object at 0x1683f88>, filename='/usr/local/bin/celery', lineno=11, function='<module>', code_context=[' sys.exit(main())\n'], index=0)]

2019-11-20 18:40:54,258 5325 DEBUG bootsteps[262] | Worker: Closing Hub...
2019-11-20 18:40:54,258 5325 DEBUG bootsteps[262] | Worker: Closing Pool...
2019-11-20 18:40:54,258 5325 DEBUG bootsteps[262] | Worker: Closing Consumer...
2019-11-20 18:40:54,258 5325 DEBUG bootsteps[262] | Worker: Stopping Consumer...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Closing Connection...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Closing Events...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Closing Tasks...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Closing ThreadControlConsumerStep...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Closing event loop...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Stopping event loop...
2019-11-20 18:40:54,259 5325 DEBUG bootsteps[262] | Consumer: Stopping ThreadControlConsumerStep...
2019-11-20 18:40:55,178 5325 DEBUG bootsteps[262] | Consumer: Stopping Tasks...
2019-11-20 18:40:55,178 5325 DEBUG bootsteps[262] | Consumer: Stopping Events...
2019-11-20 18:40:55,178 5325 DEBUG bootsteps[262] | Consumer: Stopping Connection...
2019-11-20 18:40:55,178 5325 DEBUG bootsteps[262] | Worker: Stopping Pool...
2019-11-20 18:40:55,179 5325 DEBUG bootsteps[262] | Worker: Stopping Hub...

2019-11-20 18:40:55,180 5325 WARNING log[235] [
FrameInfo(frame=<frame object at 0x7f88806ce9d0>, filename='/usr/local/lib/python3.6/dist-packages/kombu/asynchronous/hub.py', lineno=264, function='close', code_context=[' print(inspect.stack())\n'], index=0),
FrameInfo(frame=<frame object at 0x7f8880609dc8>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/components.py', lineno=85, function='stop', code_context=[' w.hub.close()\n'], index=0),
FrameInfo(frame=<frame object at 0x29e53d8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=151, function='send_all', code_context=[' fun(parent, *args)\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88806e1808>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=138, function='restart', code_context=[' self.send_all(parent, method, description, propagate=propagate)\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88806e1618>, filename='/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py', lineno=173, function='stop', code_context=[' description=what, propagate=False,\n'], index=0),
FrameInfo(frame=<frame object at 0x7f888071ea48>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/worker.py', lineno=268, function='_shutdown', code_context=[' self.blueprint.stop(self, terminate=not warm)\n'], index=0),
FrameInfo(frame=<frame object at 0x7f88806e55a0>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/worker.py', lineno=253, function='stop', code_context=[' self._shutdown(warm=True)\n'], index=0),
FrameInfo(frame=<frame object at 0x28f0698>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/worker.py', lineno=212, function='start', code_context=[' self.stop(exitcode=exc.code)\n'], index=0),
FrameInfo(frame=<frame object at 0x1be46e8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/worker.py', lineno=258, function='run', code_context=[' worker.start()\n'], index=0),
FrameInfo(frame=<frame object at 0x1be22f8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/base.py', lineno=252, function='__call__', code_context=[' ret = self.run(*args, **kwargs)\n'], index=0),
FrameInfo(frame=<frame object at 0x18be4a8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/worker.py', lineno=223, function='run_from_argv', code_context=[' return self(*args, **options)\n'], index=0),
FrameInfo(frame=<frame object at 0x1baa718>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=420, function='execute', code_context=[' ).run_from_argv(self.prog_name, argv[1:], command=argv[0])\n'], index=0),
FrameInfo(frame=<frame object at 0x1b3ed78>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=488, function='handle_argv', code_context=[' return self.execute(command, argv)\n'], index=0),
FrameInfo(frame=<frame object at 0x1a894f8>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/base.py', lineno=298, function='execute_from_commandline', code_context=[' return self.handle_argv(self.prog_name, argv[1:])\n'], index=0),
FrameInfo(frame=<frame object at 0x1ac5288>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=496, function='execute_from_commandline', code_context=[' super(CeleryCommand, self).execute_from_commandline(argv)))\n'], index=0),
FrameInfo(frame=<frame object at 0x1a13388>, filename='/usr/local/lib/python3.6/dist-packages/celery/bin/celery.py', lineno=322, function='main', code_context=[' cmd.execute_from_commandline(argv)\n'], index=0),
FrameInfo(frame=<frame object at 0x183db88>, filename='/usr/local/lib/python3.6/dist-packages/celery/__main__.py', lineno=16, function='main', code_context=[' _main()\n'], index=0),
FrameInfo(frame=<frame object at 0x1683f88>, filename='/usr/local/bin/celery', lineno=11, function='<module>', code_context=[' sys.exit(main())\n'], index=0)]

2019-11-20 18:40:55,180 5325 ERROR bootsteps[156] Error on stopping Hub: MessageStateError('Message already acknowledged with state: ACK',)
Traceback (most recent call last):
File "/usr/local/lib/python3.6/dist-packages/celery/worker/worker.py", line 205, in start
self.blueprint.start(self)
File "/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py", line 119, in start
step.start(parent)
File "/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py", line 369, in start
return self.obj.start()
File "/usr/local/lib/python3.6/dist-packages/celery/worker/consumer/consumer.py", line 318, in start
blueprint.start(self)
File "/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py", line 119, in start
step.start(parent)
File "/usr/local/lib/python3.6/dist-packages/celery/worker/consumer/consumer.py", line 596, in start
c.loop(*c.loop_args())
File "/usr/local/lib/python3.6/dist-packages/celery/worker/loops.py", line 80, in asynloop
raise WorkerShutdown(should_stop)
celery.exceptions.WorkerShutdown: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/local/lib/python3.6/dist-packages/celery/bootsteps.py", line 151, in send_all
fun(parent, *args)
File "/usr/local/lib/python3.6/dist-packages/celery/worker/components.py", line 85, in stop
w.hub.close()
File "/usr/local/lib/python3.6/dist-packages/kombu/asynchronous/hub.py", line 267, in close
item()
File "/usr/local/lib/python3.6/dist-packages/vine/promises.py", line 170, in __call__
return self.throw()
File "/usr/local/lib/python3.6/dist-packages/vine/promises.py", line 167, in __call__
retval = fun(*final_args, **final_kwargs)
File "/usr/local/lib/python3.6/dist-packages/kombu/message.py", line 132, in ack_log_error
self.ack(multiple=multiple)
File "/usr/local/lib/python3.6/dist-packages/kombu/message.py", line 125, in ack
self))
kombu.exceptions.MessageStateError: Message already acknowledged with state: ACK
2019-11-20 18:40:55,181 5325 DEBUG bootsteps[262] | Consumer: Shutdown ThreadControlConsumerStep...
2019-11-20 18:40:55,181 5325 DEBUG bootsteps[262] | Consumer: Shutdown Tasks...
2019-11-20 18:40:55,181 5325 DEBUG bootsteps[262] | Consumer: Shutdown Events...
2019-11-20 18:40:55,181 5325 DEBUG bootsteps[262] | Consumer: Shutdown Connection...

inspect

from above log we can see Hub.close called two times from:

case 1
FrameInfo(frame=<frame object at 0x7f88806d6be0>, filename='/usr/local/lib/python3.6/dist-packages/kombu/asynchronous/hub.py', lineno=112, function='reset', code_context=[' self.close()\n'], index=0)
FrameInfo(frame=<frame object at 0x29e50e8>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/loops.py', lineno=96, function='asynloop', code_context=[' hub.reset()\n'], index=0)

the source code: https://github.com/celery/celery/blob/master/celery/worker/loops.py#L88

case 2. FrameInfo(frame=<frame object at 0x7f8880609dc8>, filename='/usr/local/lib/python3.6/dist-packages/celery/worker/components.py', lineno=85, function='stop', code_context=[' w.hub.close()\n'], index=0)

the source code: https://github.com/celery/celery/blob/master/celery/worker/components.py#L85

from source code we can sure the self._ready would not to be reset to empty after execute, so the exception Message already acknowledged with state: ACK was expected by code but not to me.

My question is:

if we should reset the self._ready after execute?

why we need call case 1 ? , it doesn't make any sense in my opinion