kombu for RabbitMQ cancel_by_queue method raise ChannelError - PreconditionFailed: Basic.ack

Viewed 170

Using amqp==2.6.1 with kombu==4.6.11 - when trying to remove some queue using the method:

self.consumer.cancel_by_queue(queue_name)

the kombu library does the following:

    def cancel_by_queue(self, queue):
        """Cancel consumer by queue name."""
        qname = queue.name if isinstance(queue, Queue) else queue
        try:
            tag = self._active_tags.pop(qname)
        except KeyError:
            pass
        else:
            self.channel.basic_cancel(tag)
        finally:
            self._queues.pop(qname, None)

and amqp library does the following:

    def basic_cancel(self, consumer_tag, nowait=False, argsig='sb'):
        if self.connection is not None:
            self.no_ack_consumers.discard(consumer_tag)
            return self.send_method(
                spec.Basic.Cancel, argsig, (consumer_tag, nowait),
                wait=None if nowait else spec.Basic.CancelOk,
            )

which resulting in:

  File "/usr/local/lib/python3.7/site-packages/amqp/channel.py", line 1440, in basic_cancel
wait=None if nowait else spec.Basic.CancelOk,
  File "/usr/local/lib/python3.7/site-packages/amqp/abstract_channel.py", line 68, in send_method
return self.wait(wait, returns_tuple=returns_tuple)
  File "/usr/local/lib/python3.7/site-packages/amqp/abstract_channel.py", line 88, in wait
self.connection.drain_events(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/amqp/connection.py", line 508, in drain_events
while not self.blocking_read(timeout):
  File "/usr/local/lib/python3.7/site-packages/amqp/connection.py", line 514, in blocking_read
return self.on_inbound_frame(frame)
  File "/usr/local/lib/python3.7/site-packages/amqp/method_framing.py", line 55, in on_frame
callback(channel, method_sig, buf, None)
  File "/usr/local/lib/python3.7/site-packages/amqp/connection.py", line 521, in on_inbound_method
method_sig, payload, content,
  File "/usr/local/lib/python3.7/site-packages/amqp/abstract_channel.py", line 145, in dispatch_method
listener(*args)
  File "/usr/local/lib/python3.7/site-packages/amqp/channel.py", line 280, in _on_close
reply_code, reply_text, (class_id, method_id), ChannelError,
 amqp.exceptions.PreconditionFailed: Basic.ack: (406) PRECONDITION_FAILED - unknown delivery tag 1234

reading the documentation I saw that:

The client should not wait for a reply method, and If the server could not complete the method it will raise a channel or connection exception.

and

The client may receive an abitrary number of messages in between sending the cancel method and receiving the cancel-ok reply.

So What is the right way to use cancel_by_queue and why this exception has even been raised - and how to prevent it?

0 Answers
Related