Celery broken after release 5.0.8

[wowi42]

Hi all,

We found a bug inside py-amqp on celery: when we updated to amqp==5.0.8, we are facing the following issue:

[2021-12-19 07:48:41,093: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 510, in channel
    return self.channels[channel_id]
KeyError: None

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/bas/venv/lib/python3.10/site-packages/celery/worker/consumer/consumer.py", line 326, in start
    blueprint.start(self)
  File "/home/bas/venv/lib/python3.10/site-packages/celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/home/bas/venv/lib/python3.10/site-packages/celery/worker/consumer/gossip.py", line 107, in start
    super().start(c)
  File "/home/bas/venv/lib/python3.10/site-packages/celery/bootsteps.py", line 396, in start
    channel = c.connection.channel()
  File "/home/bas/venv/lib/python3.10/site-packages/kombu/connection.py", line 282, in channel
    chan = self.transport.create_channel(self.connection)
  File "/home/bas/venv/lib/python3.10/site-packages/kombu/transport/pyamqp.py", line 166, in create_channel
    return connection.channel()
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 513, in channel
    channel.open()
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/channel.py", line 436, in open
    return self.send_method(
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/abstract_channel.py", line 66, in send_method
    return self.wait(wait, returns_tuple=returns_tuple)
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/abstract_channel.py", line 86, in wait
    self.connection.drain_events(timeout=timeout)
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 521, in drain_events
    while not self.blocking_read(timeout):
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 527, in blocking_read
    return self.on_inbound_frame(frame)
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/method_framing.py", line 53, in on_frame
    callback(channel, method_sig, buf, None)
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 533, in on_inbound_method
    return self.channels[channel_id].dispatch_method(
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/abstract_channel.py", line 143, in dispatch_method
    listener(*args)
  File "/home/bas/venv/lib/python3.10/site-packages/amqp/connection.py", line 663, in _on_close
    raise error_for_code(reply_code, reply_text,
amqp.exceptions.ChannelNotOpen: Channel.open: (504) CHANNEL_ERROR - second 'channel.open' seen

We reverted to 5.0.7 and it's working perfectly.

We are going to investigate the issue later today/tomorrow!

[auvipy]

yes kombu integration tests are also failing celery/kombu#1462

[auvipy]

@pawl @michael-lazar if you can check please when you are available.

[99ACA]

Having the same issue

[Tobeyforce]

same here, using django 3.2.9, same thing with 3.2.10. Tried both using celery 5.0.5 and 5.2.1. Tried multiple rabbitmq versions. What really ticks me off is that I've been gooling all day long, and still this github issue didn't get indexed by google :(
Took a while to figure out amqp was the problem.

[2021-12-19 19:26:43,892: INFO/MainProcess] Connected to amqp://celery:**@127.0.0.1:5672/soklagenhet_vhost
[2021-12-19 19:26:43,896: INFO/MainProcess] mingle: searching for neighbors
[2021-12-19 19:26:44,913: INFO/MainProcess] mingle: all alone
[2021-12-19 19:26:44,924: WARNING/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 510, in channel
    return self.channels[channel_id]
KeyError: None

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/celery/worker/consumer/consumer.py", line 326, in start
    blueprint.start(self)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/celery/bootsteps.py", line 116, in start
    step.start(parent)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/celery/worker/pidbox.py", line 52, in start
    self.node.channel = c.connection.channel()
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/kombu/connection.py", line 282, in channel
    chan = self.transport.create_channel(self.connection)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/kombu/transport/pyamqp.py", line 166, in create_channel
    return connection.channel()
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 513, in channel
    channel.open()
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/channel.py", line 436, in open
    return self.send_method(
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/abstract_channel.py", line 66, in send_method
    return self.wait(wait, returns_tuple=returns_tuple)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/abstract_channel.py", line 86, in wait
    self.connection.drain_events(timeout=timeout)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 521, in drain_events
    while not self.blocking_read(timeout):
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 527, in blocking_read
    return self.on_inbound_frame(frame)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/method_framing.py", line 53, in on_frame
    callback(channel, method_sig, buf, None)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 533, in on_inbound_method
    return self.channels[channel_id].dispatch_method(
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/abstract_channel.py", line 143, in dispatch_method
    listener(*args)
  File "/home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/amqp/connection.py", line 663, in _on_close
    raise error_for_code(reply_code, reply_text,
amqp.exceptions.ChannelNotOpen: Channel.open: (504) CHANNEL_ERROR - second 'channel.open' seen
[2021-12-19 19:26:44,926: WARNING/MainProcess] /home/tobias/projects/soklagenhet_api/venv/lib/python3.8/site-packages/celery/worker/consumer/consumer.py:361: CPendingDeprecationWarning: 
In Celery 5.1 we introduced an optional breaking change which
on connection loss cancels all currently executed tasks with late acknowledgement enabled.
These tasks cannot be acknowledged as the connection is gone, and the tasks are automatically redelivered back to the queue.
You can enable this behavior using the worker_cancel_long_running_tasks_on_connection_loss setting.
In Celery 5.1 it is set to False by default. The setting will be set to True by default in Celery 6.0.

  warnings.warn(CANCEL_TASKS_BY_DEFAULT, CPendingDeprecationWarning)

[amthorn]

I'm seeing the same thing here. I confirmed the issue with two identical containers where the only difference is one uses amqp==5.0.7 and the other uses amqp==5.0.8

5.0.7 works but 5.0.8 does not. Downgrading fixes it.

[pawl]

I have a fix for this issue in this open PR: #383