Open aygalinc opened 1 year ago
Hello,
this is not providing the full picture. How is the consumer reading the messages? Is the consumer, for example, acknowledging the message once consumed?
Please provide a working example which reproduces the issue.
The consumer side :
import pika, sys, os
def main():
credentials = pika.PlainCredentials('guest', 'guest')
connection = pika.BlockingConnection(pika.ConnectionParameters(
'localhost',
5672,
'/',
credentials,
client_properties={
'connection_name': 'sub_dedup',
},
))
channel = connection.channel()
args = {
'x-message-deduplication': True
}
channel.queue_declare(queue='refresh_all', arguments=args)
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(queue='refresh_all', on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
print('Interrupted')
try:
sys.exit(0)
except SystemExit:
os._exit(0)
There is indeed something odd with the above code.
When I read the message using Elixir/Erlang client, I get the expected behavior. Same applies if I fetch the message via RabbitMQ Management WebUI. But with the Python pika
client the message seems to be deduplicated forever even if gone from the queue.
I will investigate this further.
I now have a better understanding of the problem.
The issue presents when a consumer is attached to an empty queue with auto_ack
enabled. When such scenario presents, the internal queue discard
callback is called instead of the expected publish_delivered
[1].
Issue with this, is we do not receive the message as parameter of the discard
callback but just its Id. This is not enough to retrieve the x-deduplication-header
and remove it from the cache.
Hence, the observed behaviour: the first message goes through, all subsequent ones are deemed duplicates as the original header is not removed from the cache.
This issue does not present, if auto_ack
is disabled.
The above internal API change contributed by @noxdafox will ship starting with RabbitMQ 3.12.0.
Thanks @michaelklishin, I will make a release of the plugin once RMQ 3.12.0 will be out.
https://github.com/rabbitmq/rabbitmq-server/pull/7802 was reverted because it is not rolling upgrade-safe without a feature flag.
Using a feature flag on the hot path has risks. Perhaps the PR should be adjusted so that all modules support both message IDs and message records, and backported to 3.11.x
.
Then, at some point, we can pass the entire message record along. For now, the plugin would have to require manual acknowledgements (which is almost always a good idea).
Hello ,
On a rabbit 3.9.26, we use a straightforward setting to test message depuplication with pica :
And we have a basic message consumer. We observe that the consumer publish only one message. We expect that once the consumer has finish an individual consumption, a new message will be enqueue.