title: Celery Custom Queue Exception
body: <p>Im upgrading a configuration from Celery 3.1.25 up to 4.0.2, currently I am seeing an issue when I start the worker and there's a message queued, it raises a 'NotImplementedError'. This doesn't happen when performed the other way around (first the worker and then send the message).</p>
  
  <p>The error:</p>
  
  <pre><code>[2017-01-20 17:12:13,984: INFO/MainProcess] mingle: sync with 11 nodes
  [2017-01-20 17:12:13,985: INFO/MainProcess] mingle: sync complete
  [2017-01-20 17:12:13,993: INFO/MainProcess] Canceling queue celery
  [2017-01-20 17:12:13,998: INFO/MainProcess] Started consuming from 1.my_queue
  [2017-01-20 17:12:14,001: CRITICAL/MainProcess] Unrecoverable error: NotImplementedError(u'Consumer does not have any callbacks',)
  Traceback (most recent call last):
    File "/opt/my-program/lib/python2.7/site-packages/celery/worker/worker.py", line 203, in start
      self.blueprint.start(self)
    File "/opt/my-program/lib/python2.7/site-packages/celery/bootsteps.py", line 119, in start
      step.start(parent)
    File "/opt/my-program/lib/python2.7/site-packages/celery/bootsteps.py", line 370, in start
      return self.obj.start()
    File "/opt/my-program/lib/python2.7/site-packages/celery/worker/consumer/consumer.py", line 318, in start
      blueprint.start(self)
    File "/opt/my-program/lib/python2.7/site-packages/celery/bootsteps.py", line 119, in start
      step.start(parent)
    File "/opt/my-program/lib/python2.7/site-packages/celery/worker/pidbox.py", line 50, in start
      self.node.channel = c.connection.channel()
    File "/opt/my-program/lib/python2.7/site-packages/kombu/connection.py", line 266, in channel
      chan = self.transport.create_channel(self.connection)
    File "/opt/my-program/lib/python2.7/site-packages/kombu/transport/pyamqp.py", line 100, in create_channel
      return connection.channel()
    File "/opt/my-program/lib/python2.7/site-packages/amqp/connection.py", line 456, in channel
      channel.open()
    File "/opt/my-program/lib/python2.7/site-packages/amqp/channel.py", line 448, in open
      spec.Channel.Open, 's', ('',), wait=spec.Channel.OpenOk,
    File "/opt/my-program/lib/python2.7/site-packages/amqp/abstract_channel.py", line 73, in send_method
      return self.wait(wait, returns_tuple=returns_tuple)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/abstract_channel.py", line 93, in wait
      self.connection.drain_events(timeout=timeout)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/connection.py", line 464, in drain_events
      return self.blocking_read(timeout)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/connection.py", line 469, in blocking_read
      return self.on_inbound_frame(frame)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/method_framing.py", line 88, in on_frame
      callback(channel, msg.frame_method, msg.frame_args, msg)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/connection.py", line 473, in on_inbound_method
      method_sig, payload, content,
    File "/opt/my-program/lib/python2.7/site-packages/amqp/abstract_channel.py", line 142, in dispatch_method
      listener(*args)
    File "/opt/my-program/lib/python2.7/site-packages/amqp/channel.py", line 1613, in _on_basic_deliver
      fun(msg)
    File "/opt/my-program/lib/python2.7/site-packages/kombu/messaging.py", line 623, in _receive_callback
      return on_m(message) if on_m else self.receive(decoded, message)
    File "/opt/my-program/lib/python2.7/site-packages/kombu/messaging.py", line 588, in receive
      raise NotImplementedError('Consumer does not have any callbacks')
  NotImplementedError: Consumer does not have any callbacks
  </code></pre>
  
  <p>The definition of the queue:</p>
  
  <pre><code>some_id = 1
  
  class SetQueue(bootsteps.StartStopStep):
      requires = ('celery.worker.consumer:Tasks', )
  
      def __init__(self, parent, **kwargs):
          self.my_queue = kwargs.get('my_queue')
  
      def start(self, parent):
          if self.my_queue:
              parent.cancel_task_queue(app.conf.CELERY_DEFAULT_QUEUE)
              parent.add_task_queue('{}.{}'.format(some_id, self.my_queue))
  
  class SetHostname(bootsteps.StartStopStep):
      def __init__(self, parent, **kwargs):
          self.my_queue = kwargs.get('my_queue')
          parent.hostname = '{}@{}'.format(self.my_queue, some_id)
  
  def my_custom_queue(parser):
      parser.add_argument(
          '--my-queue', help='Task queue base name.', type=str
      )
  
  app.user_options['worker'].add(my_custom_queue)
  app.steps['worker'].add(SetHostname)
  app.steps['consumer'].add(SetQueue)
  </code></pre>
  
  <p>I'm having trouble finding what I'm missing.</p>
  
  <p>Something I found in the whats new <a href="http://docs.celeryproject.org/en/latest/whatsnew-4.0.html" rel="nofollow noreferrer">doc</a> is the following</p>
  
  <p><code>The worker no longer has a Queues bootsteps, as it is now superfluous.</code></p>
  
  <p>and think may be related but dont know what it means.</p>
  
  <p>Thanks in advance for any help</p>
  
  <h3>UPDATE 2017-02-24</h3>
  
  <p>Trying to further debug this issue I've found the error occurs only when it consumes from a remote queue.</p>
  
  <p>This is when we consume from remote</p>
  
  <pre><code>[2017-02-24 16:45:02,059: INFO/MainProcess] Connected to amqp://&lt;user&gt;:**@&lt;remote_ip&gt;:5672//
  [2017-02-24 16:45:02,079: INFO/MainProcess] mingle: searching for neighbors
  [2017-02-24 16:45:03,156: INFO/MainProcess] mingle: sync with 10 nodes
  [2017-02-24 16:45:03,156: INFO/MainProcess] mingle: sync complete
  [2017-02-24 16:45:03,176: INFO/MainProcess] Canceling queue celery
  [2017-02-24 16:45:03,184: INFO/MainProcess] Started consuming from 1.my_queue
  [2017-02-24 16:45:03,192: CRITICAL/MainProcess] Unrecoverable error: NotImplementedError(u'Consumer does not have any callbacks',)
  </code></pre>
  
  <p>This is when we consume from local</p>
  
  <pre><code>[2017-02-24 17:31:35,597: INFO/MainProcess] Connected to amqp://&lt;user&gt;:**@localhost:5672//
  [2017-02-24 17:31:35,607: INFO/MainProcess] mingle: searching for neighbors
  [2017-02-24 17:31:36,623: INFO/MainProcess] mingle: sync with 1 nodes
  [2017-02-24 17:31:36,623: INFO/MainProcess] mingle: sync complete
  [2017-02-24 17:31:36,626: INFO/MainProcess] Canceling queue celery
  [2017-02-24 17:31:36,628: INFO/MainProcess] Started consuming from 1.my_queue
  [2017-02-24 17:31:36,634: INFO/MainProcess] my_queue@1 ready.
  [2017-02-24 17:31:36,638: INFO/MainProcess] Received task: &lt;a_task&gt;[4a25a5d2-933d-4111-9dca-b99187ffa757]  ETA:[2017-02-24 20:31:27.514538+00:00]
  </code></pre>
  
  <p>There's no difference on how I run the consumer worker. It only differs the queue location.</p>