New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Duplicate callback found for "0:Connection.Unblocked" #1052
Comments
Hello! Please share the code you are using. I will re-open this issue if I can reproduce. Thanks! |
It's not always reproducible. Basically this is what I do: def consume_until_empty_or_expire(self, channel, callback=None, expires_sec=0, echo=False):
"""
consume the messages in the queue until no message available
callback: func(event_data)
expires_sec: quit consuming after certain time. <=0, never expires.
"""
start_time = time.time()
if callback is None:
callback = self._default_consume_until_empty_callback
total = 0
processed = 0
returned = 0
skipped = 0
message_count = 0
while True:
elapsed_time = time.time() - start_time
if expires_sec > 0 and elapsed_time >= expires_sec:
return {
'success': True,
'total': total,
'processed': processed,
'returned': returned,
'skipped': skipped,
'message_count': message_count,
'elapsed_time_sec': elapsed_time
}
method_frame, header_frame, body = channel.basic_get(
queue=self.queue_name)
if method_frame is None or method_frame.NAME == 'Basic.GetEmpty':
return {
'success': True,
'total': total,
'processed': processed,
'returned': returned,
'skipped': skipped,
'message_count': message_count,
'elapsed_time_sec': elapsed_time
}
else:
event_data = {}
event_data['frame'] = vars(method_frame)
event_data['header'] = vars(header_frame)
event_data['body'] = json.loads(body)
message_count = event_data['frame']['message_count']
total += 1
result = callback(event_data)
if result == EventProcessingResult.PROCESSED:
channel.basic_ack(delivery_tag=method_frame.delivery_tag)
processed += 1
if result == EventProcessingResult.SKIPPED:
channel.basic_ack(delivery_tag=method_frame.delivery_tag)
skipped += 1
if result == EventProcessingResult.RETURED:
returned += 1
if echo:
print event_data['frame']['routing_key'] + ': ' + result Nothing really fancy. Probably this is related: if instance.queue_ttl_min > 0:
queue_args = {"x-expires": instance.queue_ttl_min * 60000}
channel.queue_declare(
instance.queue_name,
durable=instance.queue_durable,
arguments=queue_args) I did not really see the problem until started to use "x-expires". Could be a coincidence, though. |
Hi, did you solve this? |
@benizri-ofir - please provide code that can reproduce this, or at the very least, a working set of code or steps that might reproduce the issue. |
Hi |
@nkborisov no, I never received the requested information. There is no evidence at this point of a bug in Pika. |
@lukebakken Thank you for your answer
That class is base for two inheritances - SenderConnection and ReceiverConnection (just dummy handlers of user queues).
Next in RabbitMQ management plugin has been reproducing connection failure situation by force closing. As a result i got following log every time:
"Duplicate callback" log warning appears every time just before expected message about the connection closing event. By debugging i understood that the warning message always appears before any calls of BlockingConnection constructors and destructors. It seems very strange for me. |
The problem occurred also in my case (for both blocked/unblocked callbacks).
The solution would be to call the remove_all method in case the callback_value is not provided
Apparently this problem was introduced with #192 |
We are using
BlockingConnection
andbasic_get
and no callback function. How come we got a lot of Duplicate callback found for "0:Connection.Unblocked" errors? We use pika 0.11.2Any idea? or thoughts?
The text was updated successfully, but these errors were encountered: