Kombu: Kombo + SQS + PyCurl, 'NoneType' object has no attribute 'call_repeatedly'

[jonzlin95]

Hi. I'm running the lastest Kombu master (w/ stable 4.0.2 Celery), SQS Broker, with Django DB Backend, and I run into this error.

[2017-05-31 08:08:47,205: ERROR/MainProcess] Pending callback raised: AttributeError("'NoneType' object has no attribute 'call_repeatedly'",)
Traceback (most recent call last):
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/celery/worker/consumer/consumer.py", line 233, in perform_pending_operations
    self._pending_operations.pop()()
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/vine/promises.py", line 139, in __call__
    return self.throw()
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/vine/promises.py", line 136, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/message.py", line 130, in ack_log_error
    self.ack(multiple=multiple)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/message.py", line 125, in ack
    self.channel.basic_ack(self.delivery_tag, multiple=multiple)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/transport/SQS.py", line 380, in basic_ack
    self.asynsqs.delete_message(message['sqs_queue'],
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/transport/SQS.py", line 436, in asynsqs
    region=self.region
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/aws/sqs/connection.py", line 27, in __init__
    **kwargs
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/aws/connection.py", line 178, in __init__
    **http_client_params)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/aws/connection.py", line 151, in __init__
    self._httpclient = http_client or get_client()
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/http/__init__.py", line 22, in get_client
    client = hub._current_http_client = Client(hub, **kwargs)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/http/__init__.py", line 13, in Client
    return CurlClient(hub, **kwargs)
  File "/home/ubuntu/ies-league/venv/local/lib/python2.7/site-packages/kombu/async/http/curl.py", line 57, in __init__
    self._timeout_check_tref = self.hub.call_repeatedly(
AttributeError: 'NoneType' object has no attribute 'call_repeatedly'
[2017-05-31 08:08:47,207: ERROR/MainProcess] Pending callback raised: AttributeError("'NoneType' object has no attribute 'call_repeatedly'",)

[auvipy]

could you try the master branch

[jonzlin95]

Master branch of which repo. This was master branch of Kombu (circa 16 days ago). (I eventually just gave up on SQS and used Redis instead as my broker).

[alukach]

I am also exeriencing this bug. I am using the Kombu master branch, SQS backend, and am running with a custom consumer (no celery involved):

(env)vagrant@vagrant-ubuntu-trusty-64:/vagrant/cadasta$ ./manage.py process_messages -v 1

Got message [[], {}, {'callbacks': None, 'errbacks': None, 'chord': None, 'chain': None}] <Message object at 0x7fdb7ca0cd38 with details {'delivery_info': {'exchange': 'task_exchange', 'routing_key': 'export'}, 'body_length': 77, 'state': 'RECEIVED', 'content_type': 'application/json', 'properties': {'correlation_id': '63d71bd3-26e9-4ad5-8aad-1b0755d926b1'}, 'delivery_tag': 'AQEBNr1LRk+QljxtqaagKsrUWPSeOcbY13qL3qByOpUi9WbtM3xXRx1T2lg+XVVCQkpf15LBcZIeZxYtIbE+rIN0lvWoGE7Ub1q8XJKrcd9ZOcyI02rD0n16w39sSD4yUl45oZmo/+0NwwrUWjy6/CEqddZrFrrjFjwMON+QwFbiUD7C/l6s4ruQ72ckfvuwiNQV7QRkG/cEhL5m0z7nPRRmAjhTpIxwi2WSEdKfuDHqdiLzFZUprVpIFlYruVJ/g6+r0ftsNw74WFiC8YTMRlfL0Yoh/MMD5Ergn2dJkuTQFAo='}>
Restoring 1 unacknowledged message(s)
Traceback (most recent call last):
  File "/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py", line 20, in get_client
    return hub._current_http_client
AttributeError: 'NoneType' object has no attribute '_current_http_client'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "./manage.py", line 10, in <module>
    execute_from_command_line(sys.argv)
  File "/opt/cadasta/env/lib/python3.5/site-packages/django/core/management/__init__.py", line 367, in execute_from_command_line
    utility.execute()
  File "/opt/cadasta/env/lib/python3.5/site-packages/django/core/management/__init__.py", line 359, in execute
    self.fetch_command(subcommand).run_from_argv(self.argv)
  File "/opt/cadasta/env/lib/python3.5/site-packages/django/core/management/base.py", line 294, in run_from_argv
    self.execute(*args, **cmd_options)
  File "/opt/cadasta/env/lib/python3.5/site-packages/django/core/management/base.py", line 345, in execute
    output = self.handle(*args, **options)
  File "/vagrant/cadasta/tasks/management/commands/process_messages.py", line 76, in handle
    worker.run()
  File "/opt/cadasta/env/src/kombu/kombu/mixins.py", line 170, in run
    for _ in self.consume(limit=None, **kwargs):
  File "/opt/cadasta/env/src/kombu/kombu/mixins.py", line 192, in consume
    conn.drain_events(timeout=safety_interval)
  File "/opt/cadasta/env/src/kombu/kombu/connection.py", line 301, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File "/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py", line 961, in drain_events
    get(self._deliver, timeout=timeout)
  File "/opt/cadasta/env/src/kombu/kombu/utils/scheduling.py", line 56, in get
    return self.fun(resource, callback, **kwargs)
  File "/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py", line 999, in _drain_channel
    return channel.drain_events(callback=callback, timeout=timeout)
  File "/opt/cadasta/env/src/kombu/kombu/transport/SQS.py", line 138, in drain_events
    self._poll(self.cycle, callback, timeout=timeout)
  File "/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py", line 402, in _poll
    return cycle.get(callback)
  File "/opt/cadasta/env/src/kombu/kombu/utils/scheduling.py", line 56, in get
    return self.fun(resource, callback, **kwargs)
  File "/opt/cadasta/env/src/kombu/kombu/transport/SQS.py", line 294, in _get_bulk
    self.connection._deliver(msg, queue)
  File "/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py", line 981, in _deliver
    callback(message)
  File "/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py", line 633, in _callback
    return callback(message)
  File "/opt/cadasta/env/src/kombu/kombu/messaging.py", line 624, in _receive_callback
    return on_m(message) if on_m else self.receive(decoded, message)
  File "/opt/cadasta/env/src/kombu/kombu/messaging.py", line 590, in receive
    [callback(body, message) for callback in callbacks]
  File "/opt/cadasta/env/src/kombu/kombu/messaging.py", line 590, in <listcomp>
    [callback(body, message) for callback in callbacks]
  File "/vagrant/cadasta/tasks/management/commands/process_messages.py", line 55, in process_task
    message.ack()
  File "/opt/cadasta/env/src/kombu/kombu/message.py", line 125, in ack
    self.channel.basic_ack(self.delivery_tag, multiple=multiple)
  File "/opt/cadasta/env/src/kombu/kombu/transport/SQS.py", line 382, in basic_ack
    self.asynsqs.delete_message(message['sqs_queue'],
  File "/opt/cadasta/env/src/kombu/kombu/transport/SQS.py", line 438, in asynsqs
    region=self.region
  File "/opt/cadasta/env/src/kombu/kombu/async/aws/sqs/connection.py", line 27, in __init__
    **kwargs
  File "/opt/cadasta/env/src/kombu/kombu/async/aws/connection.py", line 178, in __init__
    **http_client_params)
  File "/opt/cadasta/env/src/kombu/kombu/async/aws/connection.py", line 151, in __init__
    self._httpclient = http_client or get_client()
  File "/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py", line 22, in get_client
    client = hub._current_http_client = Client(hub, **kwargs)
  File "/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py", line 13, in Client
    return CurlClient(hub, **kwargs)
  File "/opt/cadasta/env/src/kombu/kombu/async/http/curl.py", line 57, in __init__
    self._timeout_check_tref = self.hub.call_repeatedly(
AttributeError: 'NoneType' object has no attribute 'call_repeatedly'

My consumer:

import logging

from django.conf import settings
from django.core.management.base import BaseCommand
from kombu import Consumer, Queue
from kombu.mixins import ConsumerMixin
from kombu.async import Hub

from tasks.celery import app
from tasks.consumer import ConsumeMessage, Consumer


class Worker(ConsumerMixin):

    def __init__(self, connection, queues):
        self.connection = connection
        self.queues = queues

    def get_consumers(self, Consumer, channel):
        return [Consumer(queues=self.queues,
                         accept=['pickle', 'json'],
                         callbacks=[self.process_task])]

    def process_task(self, body, message):
        print("Got message", body, message)
        message.ack()


class Command(BaseCommand):
    help = "Sync task and result messages with database."

    def add_arguments(self, parser):
        parser.add_argument('--queue', '-q', default=settings.PLATFORM_QUEUE)

    def handle(self, queue, *args, **options):
        with app.connection() as conn:
            try:
                worker = Worker(conn, queues=[Queue(queue)])
                worker.run()
            except KeyboardInterrupt:
                print('bye bye')

[alukach]

Okay, digging through the Kombu source code, it appears I needed to add the following near the top of my Consumer file:

from kombu.async import Hub, set_event_loop

set_event_loop(Hub())

This will ensure that a Hub object is returned when get_event_loop() is called in kombu/async/http/__init__.py:

def get_client(hub=None, **kwargs):
    """Get or create HTTP client bound to the current event loop."""
    hub = hub or get_event_loop()
    try:
        return hub._current_http_client
    except AttributeError:
        client = hub._current_http_client = Client(hub, **kwargs)
        return client

What would be the harm in having get_event_loop() call set_event_loop(Hub()) if no event loop currently exists? I'd be happy to make a PR for this.

[alukach]

To append to this, SQS seems to be unusable for RPC backend for the reason mentioned above. Here's an example:

In [24]: res = app.send_task('export.chain')

In [25]: res.get()
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py in get_client(hub, **kwargs)
     19     try:
---> 20         return hub._current_http_client
     21     except AttributeError:

AttributeError: 'NoneType' object has no attribute '_current_http_client'

During handling of the above exception, another exception occurred:

AttributeError                            Traceback (most recent call last)
<ipython-input-25-8bb969b0b8af> in <module>()
----> 1 res.get()

/opt/cadasta/env/lib/python3.5/site-packages/celery/result.py in get(self, timeout, propagate, interval, no_ack, follow_parents, callback, on_message, on_interval, disable_sync_subtasks, EXCEPTION_STATES, PROPAGATE_STATES)
    192             propagate=propagate,
    193             callback=callback,
--> 194             on_message=on_message,
    195         )
    196     wait = get  # deprecated alias to :meth:`get`.

/opt/cadasta/env/lib/python3.5/site-packages/celery/backends/async.py in wait_for_pending(self, result, callback, propagate, **kwargs)
    187                          callback=None, propagate=True, **kwargs):
    188         self._ensure_not_eager()
--> 189         for _ in self._wait_for_pending(result, **kwargs):
    190             pass
    191         return result.maybe_throw(callback=callback, propagate=propagate)

/opt/cadasta/env/lib/python3.5/site-packages/celery/backends/async.py in _wait_for_pending(self, result, timeout, on_interval, on_message, **kwargs)
    254             for _ in self.drain_events_until(
    255                     result.on_ready, timeout=timeout,
--> 256                     on_interval=on_interval):
    257                 yield
    258                 sleep(0)

/opt/cadasta/env/lib/python3.5/site-packages/celery/backends/async.py in drain_events_until(self, p, timeout, on_interval, wait)
     55                 raise socket.timeout()
     56             try:
---> 57                 yield self.wait_for(p, wait, timeout=1)
     58             except socket.timeout:
     59                 pass

/opt/cadasta/env/lib/python3.5/site-packages/celery/backends/async.py in wait_for(self, p, wait, timeout)
     64
     65     def wait_for(self, p, wait, timeout=None):
---> 66         wait(timeout=timeout)
     67
     68

/opt/cadasta/env/lib/python3.5/site-packages/celery/backends/rpc.py in drain_events(self, timeout)
     61     def drain_events(self, timeout=None):
     62         if self._connection:
---> 63             return self._connection.drain_events(timeout=timeout)
     64         elif timeout:
     65             time.sleep(timeout)

/opt/cadasta/env/src/kombu/kombu/connection.py in drain_events(self, **kwargs)
    299             socket.timeout: if the timeout is exceeded.
    300         """
--> 301         return self.transport.drain_events(self.connection, **kwargs)
    302
    303     def maybe_close_channel(self, channel):

/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py in drain_events(self, connection, timeout)
    959         while 1:
    960             try:
--> 961                 get(self._deliver, timeout=timeout)
    962             except Empty:
    963                 if timeout is not None and monotonic() - time_start >= timeout:

/opt/cadasta/env/src/kombu/kombu/utils/scheduling.py in get(self, callback, **kwargs)
     54             resource = self._next()
     55             try:
---> 56                 return self.fun(resource, callback, **kwargs)
     57             except self.predicate:
     58                 # reraise when retries exchausted.

/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py in _drain_channel(self, channel, callback, timeout)
    997
    998     def _drain_channel(self, channel, callback, timeout=None):
--> 999         return channel.drain_events(callback=callback, timeout=timeout)
   1000
   1001     @property

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in drain_events(self, timeout, callback, **kwargs)
    136
    137         # At this point, go and get more messages from SQS
--> 138         self._poll(self.cycle, callback, timeout=timeout)
    139
    140     def _reset_cycle(self):

/opt/cadasta/env/src/kombu/kombu/transport/virtual/base.py in _poll(self, cycle, callback, timeout)
    400     def _poll(self, cycle, callback, timeout=None):
    401         """Poll a list of queues for available messages."""
--> 402         return cycle.get(callback)
    403
    404     def _get_and_deliver(self, queue, callback):

/opt/cadasta/env/src/kombu/kombu/utils/scheduling.py in get(self, callback, **kwargs)
     54             resource = self._next()
     55             try:
---> 56                 return self.fun(resource, callback, **kwargs)
     57             except self.predicate:
     58                 # reraise when retries exchausted.

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in _get_bulk(self, queue, max_if_unlimited, callback)
    291                 for m in resp['Messages']:
    292                     m['Body'] = AsyncMessage(body=m['Body']).decode()
--> 293                 for msg in self._messages_to_python(resp['Messages'], queue):
    294                     self.connection._deliver(msg, queue)
    295                 return

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in _messages_to_python(self, messages, queue)
    252         """
    253         q = self._new_queue(queue)
--> 254         return [self._message_to_python(m, queue, q) for m in messages]
    255
    256     def _get_bulk(self, queue,

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in <listcomp>(.0)
    252         """
    253         q = self._new_queue(queue)
--> 254         return [self._message_to_python(m, queue, q) for m in messages]
    255
    256     def _get_bulk(self, queue,

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in _message_to_python(self, message, queue_name, queue)
    217         if queue_name in self._noack_queues:
    218             queue = self._new_queue(queue_name)
--> 219             self.asynsqs.delete_message(queue, message['ReceiptHandle'])
    220         else:
    221             try:

/opt/cadasta/env/src/kombu/kombu/transport/SQS.py in asynsqs(self)
    436             self._asynsqs = AsyncSQSConnection(
    437                 sqs_connection=self.sqs,
--> 438                 region=self.region
    439             )
    440         return self._asynsqs

/opt/cadasta/env/src/kombu/kombu/async/aws/sqs/connection.py in __init__(self, sqs_connection, debug, region, **kwargs)
     25             sqs_connection,
     26             region_name=region, debug=debug,
---> 27             **kwargs
     28         )
     29

/opt/cadasta/env/src/kombu/kombu/async/aws/connection.py in __init__(self, sqs_connection, http_client, http_client_params, **kwargs)
    176             http_client_params = {}
    177         AsyncConnection.__init__(self, sqs_connection, http_client,
--> 178                                  **http_client_params)
    179
    180     def make_request(self, operation, params_, path, verb, callback=None):  # noqa

/opt/cadasta/env/src/kombu/kombu/async/aws/connection.py in __init__(self, sqs_connection, http_client, **kwargs)
    149     def __init__(self, sqs_connection, http_client=None, **kwargs):  # noqa
    150         self.sqs_connection = sqs_connection
--> 151         self._httpclient = http_client or get_client()
    152
    153     def get_http_connection(self):

/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py in get_client(hub, **kwargs)
     20         return hub._current_http_client
     21     except AttributeError:
---> 22         client = hub._current_http_client = Client(hub, **kwargs)
     23         return client

/opt/cadasta/env/src/kombu/kombu/async/http/__init__.py in Client(hub, **kwargs)
     11     """Create new HTTP client."""
     12     from .curl import CurlClient
---> 13     return CurlClient(hub, **kwargs)
     14
     15

/opt/cadasta/env/src/kombu/kombu/async/http/curl.py in __init__(self, hub, max_clients)
     55
     56         self._socket_action = self._multi.socket_action
---> 57         self._timeout_check_tref = self.hub.call_repeatedly(
     58             1.0, self._timeout_check,
     59         )

AttributeError: 'NoneType' object has no attribute 'call_repeatedly'

[blueyed]

I've seen this also when trying --pool eventlet to work around https://github.com/celery/celery/issues/3759.

Using --pool solo however then fixed https://github.com/celery/celery/issues/3759 for me.

[jendagroovy]

Hi, I am also getting this error with Celery+SQS resulting in messages not being consumed. I have tried @alukach 's proposed solution as a workaround and can confirm it helps.

Using --pool solo did not help.

So can we fix this in Kombu somehow? Is creating the event loop when requesting one the right way, or should that be handled explicitly somewhere else?

[EmersonYe]

Any updates? I am still getting the same AttributeError: 'NoneType' object has no attribute 'call_repeatedly' error, preventing workers from consuming messages from SQS.

I have tried --pool solo and --pool eventlet to no avail.

I can confirm jay-johnson's fix worked for me. Thanks!

[nsonic001]

Is this issue resolved, or which version is stable?
I am experiencing the same issue.

[auvipy]

@blueyed is this still a bug on master?

[melbic]

It seems like it.

[cmberryau]

Seems to be fixed for me in 4.2.1

[austinbutler]

It is still not working on 4.2.1 for me.

[mengjuleu]

Is there any update about this issue? I still get the same exception in v4.2.1.

[sergeyzsg]

It happens because of gevent(may be eventlet).
https://github.com/celery/celery/blob/master/celery/worker/worker.py#L123
https://github.com/celery/celery/blob/master/celery/worker/worker.py#L241

Celery does not start kombu's hub if you are using gevent.
But pycurl wants to use the hub.
Any Ideas?

[rhymiz]

I got this issue resolved by pinning pycurl version: pycurl = "==7.19.3"

[nitrocode]

I'm currently using Flask, Celery with SQS.

celery[sqs]==4.4.2
flask==1.1.2
pycurl==7.43.0.2

I run my flask server, then celery, then I hit an endpoint to run my task and get OP's error. I tried all of these suggestions and none of them worked for me.

• Using set_event_loop(Hub()) throws an exception upon import
• celery worker with --pool solo didn't make a difference
• downgrading pycurl to 7.19.3

I noticed if I don't wait for the task to complete using task.wait() then I won't see the exception.