cancel_consumer might lead into 'Unrecoverable error': Message for queue 'xxxx' without consumers. #2953

alekibango · Dec 7, 2015

i really need some workers to be able to get out of most queues in some cases.

When i use some prefetch and concurrency,
it is really possible to crash worker by using cancel_consumer to stop listening to a queue.

When such task from cancelled queue comes in,
this problem should be recoverable, not unrecoverable one.
Just warning in log should be ok, task should be returned into its queue (with late_ack).

Possibly related (affected) issues: #2564, #1452

[2015-12-07 04:52:41,490: INFO/MainProcess] Cancelling queue extra2
..
[2015-12-07 04:52:41,493: INFO/MainProcess] Task tasks.extraq[481dd03c-f91d-496e-8e8f-444c2a3b1420] succeeded in 0.0125239877962s: None
[2015-12-07 04:52:41,494: ERROR/MainProcess] Unrecoverable error: KeyError("Message for queue 'extra2' without consumers: {u'body': u'eyJldGEiOiBudWxsLCAiaWQiOiAiMGQ3Nzk3NWQtYjNmNS00YWViLTg5NzQtZjA5ZmMyZjE3MmM4IiwgInRhc2siOiAidGFza3MuZXh0cmFxIiwgInRhc2tzZXQiOiBudWxsLCAidGltZWxpbWl0IjogW251bGwsIG51bGxdLCAicmV0cmllcyI6IDAsICJrd2FyZ3MiOiB7fSwgImV4cGlyZXMiOiBudWxsLCAiY2hvcmQiOiBudWxsLCAidXRjIjogdHJ1ZSwgImNhbGxiYWNrcyI6IG51bGwsICJhcmdzIjogWyJmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZiJdLCAiZXJyYmFja3MiOiBudWxsfQ==', u'headers': {}, u'content-type': u'application/json', u'properties': {u'body_encoding': u'base64', u'delivery_info': {u'priority': 0, u'routing_key': u'extra2', u'exchange': u'extra2'}, u'delivery_mode': 2, u'correlation_id': u'0d77975d-b3f5-4aeb-8974-f09fc2f172c8', u'reply_to': u'7a4016f2-625d-3add-9cc2-a316b182811d', u'delivery_tag': u'3d2d33bc-a9d1-44e6-960a-d00226a504d5'}, u'content-encoding': u'utf-8'}",)
Traceback (most recent call last):
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/worker/__init__.py", line 206, in start
    self.blueprint.start(self)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/bootsteps.py", line 123, in start
    step.start(parent)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/bootsteps.py", line 374, in start
    return self.obj.start()
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/worker/consumer.py", line 278, in start
    blueprint.start(self)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/bootsteps.py", line 123, in start
    step.start(parent)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/worker/consumer.py", line 821, in start
    c.loop(*c.loop_args())
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/celery/worker/loops.py", line 76, in asynloop
    next(loop)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/kombu/async/hub.py", line 324, in create_loop
    cb(*cbargs)
  File "/home/dp/test/celery3.19/local/lib/python2.7/site-packages/kombu/transport/redis.py", line 960, in on_readable
    queue, message))
KeyError: "Message for queue 'extra2' without consumers: {u'body': u'eyJldGEiOiBudWxsLCAiaWQiOiAiMGQ3Nzk3NWQtYjNmNS00YWViLTg5NzQtZjA5ZmMyZjE3MmM4IiwgInRhc2siOiAidGFza3MuZXh0cmFxIiwgInRhc2tzZXQiOiBudWxsLCAidGltZWxpbWl0IjogW251bGwsIG51bGxdLCAicmV0cmllcyI6IDAsICJrd2FyZ3MiOiB7fSwgImV4cGlyZXMiOiBudWxsLCAiY2hvcmQiOiBudWxsLCAidXRjIjogdHJ1ZSwgImNhbGxiYWNrcyI6IG51bGwsICJhcmdzIjogWyJmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZiJdLCAiZXJyYmFja3MiOiBudWxsfQ==', u'headers': {}, u'content-type': u'application/json', u'properties': {u'body_encoding': u'base64', u'delivery_info': {u'priority': 0, u'routing_key': u'extra2', u'exchange': u'extra2'}, u'delivery_mode': 2, u'correlation_id': u'0d77975d-b3f5-4aeb-8974-f09fc2f172c8', u'reply_to': u'7a4016f2-625d-3add-9cc2-a316b182811d', u'delivery_tag': u'3d2d33bc-a9d1-44e6-960a-d00226a504d5'}, u'content-encoding': u'utf-8'}"
[2015-12-07 04:52:41,499: INFO/MainProcess] Task tasks.extraq[23cd584b-4fe7-46e5-b6b8-1b055cc6cb17] succeeded in 0.0156317800283s: None
[2015-12-07 04:52:41,499: INFO/MainProcess] Task tasks.extraq[f58fb897-a467-4f57-95a5-2b2da41ee140] succeeded in 0.00987975113094s: None
[2015-12-07 04:52:41,500: DEBUG/MainProcess] | Worker: Closing Hub...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Worker: Closing Pool...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Worker: Closing Consumer...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Worker: Stopping Consumer...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Consumer: Closing Connection...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Consumer: Closing Events...
[2015-12-07 04:52:41,501: DEBUG/MainProcess] | Consumer: Closing Tasks...
[2015-12-07 04:52:41,502: DEBUG/MainProcess] | Consumer: Closing Control...
[2015-12-07 04:52:41,502: DEBUG/MainProcess] | Consumer: Closing event loop...
[2015-12-07 04:52:41,502: DEBUG/MainProcess] | Consumer: Stopping event loop...
[2015-12-07 04:52:41,502: DEBUG/MainProcess] | Consumer: Stopping Control...
[2015-12-07 04:52:41,504: DEBUG/MainProcess] | Consumer: Stopping Tasks...
[2015-12-07 04:52:41,504: DEBUG/MainProcess] Cancelling task consumer...
[2015-12-07 04:52:41,504: DEBUG/MainProcess] | Consumer: Stopping Events...
[2015-12-07 04:52:41,505: DEBUG/MainProcess] | Consumer: Stopping Connection...
[2015-12-07 04:52:41,505: DEBUG/MainProcess] | Worker: Stopping Pool...
[2015-12-07 04:52:46,501: DEBUG/MainProcess] | Worker: Stopping Hub...
[2015-12-07 04:52:46,501: DEBUG/MainProcess] | Consumer: Shutdown Control...
[2015-12-07 04:52:46,502: DEBUG/MainProcess] | Consumer: Shutdown Tasks...
[2015-12-07 04:52:46,502: DEBUG/MainProcess] Cancelling task consumer...
[2015-12-07 04:52:46,502: DEBUG/MainProcess] Closing consumer channel...
[2015-12-07 04:52:46,502: DEBUG/MainProcess] | Consumer: Shutdown Events...
[2015-12-07 04:52:46,503: DEBUG/MainProcess] | Consumer: Shutdown Connection...

also depends on celery/py-amqp@b6d2567

ask · Dec 8, 2015

also depends on celery/py-amqp@b6d2567

unsubscribe

Sent from my iPhone

On 9 Dec 2015, at 5:03 am, Ask Solem Hoel notifications@github.com wrote:

also depends on b6d2567d9b9e9449fec9d5a3950b1d4ed55b214c

—
Reply to this email directly or view it on GitHub.

matkeng · Dec 8, 2015

unsubscribe

Sent from my iPhone

On 9 Dec 2015, at 5:03 am, Ask Solem Hoel notifications@github.com wrote:

also depends on b6d2567d9b9e9449fec9d5a3950b1d4ed55b214c

—
Reply to this email directly or view it on GitHub.

@ask

@ask is it possible to backport these changes or publish a new package by bumping incremental version?

Context:
The fix doesn't seem to be available with kombu package 3.0.35.
I was able to get it though by installing the latest snapshot

pip install --upgrade https://github.com/celery/kombu/zipball/master#egg=kombu
[amqp-2.0.2 kombu-4.0.0a1 vine-1.1.0]

our current celery version is 3.1.17 and it has dependency on kombu

Requires-Dist: kombu (>=3.0.24,<3.1) so installing a major version release --- kombu-4.0.0a1 might not be a good idea at this point, thoughts?

kheraankit · Jun 23, 2016

@ask is it possible to backport these changes or publish a new package by bumping incremental version?

Context:
The fix doesn't seem to be available with kombu package 3.0.35.
I was able to get it though by installing the latest snapshot

pip install --upgrade https://github.com/celery/kombu/zipball/master#egg=kombu
[amqp-2.0.2 kombu-4.0.0a1 vine-1.1.0]

our current celery version is 3.1.17 and it has dependency on kombu

Requires-Dist: kombu (>=3.0.24,<3.1) so installing a major version release --- kombu-4.0.0a1 might not be a good idea at this point, thoughts?

ask closed this in celery/kombu@c53bed2 Dec 8, 2015

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 17, 2016

Reject+requeue messages that have no consumers. Closes celery/celery#…
…2564. Closes celery/celery#2953

6ed7a00

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 19, 2016

Reject+requeue messages that have no consumers. Closes celery/celery#…
…2564. Closes celery/celery#2953

58226d5

celery/celery

cancel_consumer might lead into 'Unrecoverable error': Message for queue 'xxxx' without consumers. #2953

alekibango commented Dec 7, 2015

ask closed this in `celery/kombu@c53bed2` Dec 8, 2015

This comment has been minimized.

ask commented Dec 8, 2015

This comment has been minimized.

matkeng commented Dec 8, 2015

This comment has been minimized.

kheraankit commented Jun 23, 2016 •

edited

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 17, 2016

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 19, 2016

celery/celery

Join GitHub today

cancel_consumer might lead into 'Unrecoverable error': Message for queue 'xxxx' without consumers. #2953

Comments

alekibango commented Dec 7, 2015

ask closed this in celery/kombu@c53bed2 Dec 8, 2015

This comment has been minimized.

ask Dec 8, 2015

ask commented Dec 8, 2015

This comment has been minimized.

matkeng Dec 8, 2015

matkeng commented Dec 8, 2015

This comment has been minimized.

kheraankit Jun 23, 2016

kheraankit commented Jun 23, 2016 • edited

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 17, 2016

arcivanov pushed a commit to karellen/karellen-kombu-ext that referenced this issue Dec 19, 2016

ask closed this in `celery/kombu@c53bed2` Dec 8, 2015

kheraankit commented Jun 23, 2016 •

edited