Kombu: Kombu 4.1 seems to break celery's task_publish_retry

[peinkofer]

Dear valued Kombu developers,

when using Celery with RabbitMQ I stumbled across a problem where celery's task_publish_retry gets not honored, when I use Kombu 4.1.0. Please see below my Bug report. If you have any further questions, please let me know.

Many thanks in advance.
Best Regards,
Stephan

Environment:

• Debian 8.9 with python 3.4.2
• amqp==2.2.1
• billiard==3.5.0.3
• celery==4.1.0
• kombu==4.1.0
• pytz==2017.2
• vine==1.1.4

Steps to reproduce:

1. Make sure Broker is NOT running
2. Create tasks.py

from celery import Celery
app = Celery('tasks', broker='pyamqp://celery:secret@localhost/myvhost')
@app.task
def add(x, y):
    return x + y

1. Create test.py

from tasks import add
add.apply_async((4,4), retry=False)

1. Run test.py

• Expected result: kombu.exceptions.OperationalError raised because connection to broker cannot be established
• Observed result: test.py blocks/hangs.

1. Hit CTRL-C to get the stack trace
   Traceback (most recent call last):
   File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/utils/functional.py", line 36, in __call__
   return self.__value__
   AttributeError: 'ChannelPromise' object has no attribute '__value__'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/utils/functional.py", line 333, in retry_over_time
    return fun(*args, **kwargs)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/connection.py", line 261, in connect
    return self.connection
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/connection.py", line 802, in connection
    self._connection = self._establish_connection()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/connection.py", line 757, in _establish_connection
    conn = self.transport.establish_connection()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/transport/pyamqp.py", line 130, in establish_connection
    conn.connect()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/amqp/connection.py", line 296, in connect
    self.transport.connect()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/amqp/transport.py", line 123, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/amqp/transport.py", line 164, in _connect
    self.sock.connect(sa)
ConnectionRefusedError: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "test.py", line 2, in <module>
    add.apply_async((4,4), retry=False)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/celery/app/task.py", line 536, in apply_async
    **options
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/celery/app/base.py", line 737, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/celery/app/amqp.py", line 554, in send_task_message
    **properties
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/messaging.py", line 181, in publish
    exchange_name, declare,
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/messaging.py", line 187, in _publish
    channel = self.channel
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/messaging.py", line 209, in _get_channel
    channel = self._channel = channel()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/utils/functional.py", line 38, in __call__
    value = self.__value__ = self.__contract__()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/messaging.py", line 224, in <lambda>
    channel = ChannelPromise(lambda: connection.default_channel)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/connection.py", line 819, in default_channel
    self.ensure_connection()
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/connection.py", line 405, in ensure_connection
    callback)
  File "/home/a2822bp/.virtualenvs/dssweb/lib/python3.4/site-packages/kombu/utils/functional.py", line 345, in retry_over_time
    sleep(1.0)
KeyboardInterrupt

1. Install Kombu 4.0.2 using pip install --upgrade kombu==4.0.2

2. Rerun test.py

3. Exception is raised as expected:

kombu.exceptions.OperationalError: [Errno 111] Connection refused

[sh4nks]

Having the same issue.

Also celery status, celery inspect ping & co isn't working because of that.

[thedrow]

I'm confused. The reproduction instructions talk about 4.0.2 and the initial report talks about 4.1.

[peinkofer]

You start with 4.1 and see how it fails and then downgrade to 4.0.2 to see how it works. (And how 4.1 presumably schould also behave)

[MickaelBergem]

I can confirm the issue and I am also affected. The stacktrace in https://github.com/celery/celery/issues/4296 looks similar to what I have when I am debugging the issue.

I also confirm that downgrading to 4.0.2 works.

[auvipy]

https://github.com/celery/kombu/pull/769 should fix it

[hans2520]

Is the expected behavior for default conf to retry forever?

If not, this issue still applies (or was regressed) in 4.6.8 - 4.6.11, haven't checked other versions.

Had to debug to find out why tests were hanging -- /lib/python3.6/site-packages/kombu/utils/functional.py(344)retry_over_time()

If default parameters are passed into retry_over_time, as the code below shows, then with repeatLast=True as it is in fxrange, it will just keep retrying from 2 seconds up to 32 seconds, and will just keep retrying the 32 seconds interval and never fail, just stuck in an infinite loop.

def retry_over_time(fun, catch, args=None, kwargs=None, errback=None,
                    max_retries=None, interval_start=2, interval_step=2,
                    interval_max=30, callback=None, timeout=None):
...
    kwargs = {} if not kwargs else kwargs
    args = [] if not args else args
    interval_range = fxrange(interval_start,
                             interval_max + interval_start,
                             interval_step, repeatlast=True)

[auvipy]

is this issue related can you check please https://github.com/celery/kombu/issues/857