# Consume data from sharded rabbitmq queue #2 **URL:** \ **Category:** Logstash\ **Created:** [March 13, 2019, 3:46pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178 "2019-03-13T15:46:34Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![lsdev14](https://avatars.discourse-cdn.com/v4/letter/l/e9c0ed/32.png) [@lsdev14](https://discuss.elastic.co/u/lsdev14)\ **Post date:** [March 13, 2019, 3:46pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/1 "2019-03-13T15:46:34Z")
I'm testing the rabbitmq-sharding plugin [https://github.com/rabbitmq/rabbitmq-sharding](https://github.com/rabbitmq/rabbitmq-sharding) I created a test scenario based in this question [Consume data from sharded RabbitMQ queue](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248) In resume I have these instances 2 producers (logstash 6.6.3) 1 rabbitmq 3.6.11 2 consumers (logstash 6.6.3) Pipeline of the **producer 1** input { heartbeat { interval =\> 1 type =\> "heartbeat" message =\> "%{@timestamp}" } } output { rabbitmq { host =\> 'rabbitmq:5672' exchange =\> 'events.sharded' exchange\_type =\> 'x-modulus-hash' key =\> 'producer-1' persistent =\> true workers =\> 1 durable =\> true password =\> '123456' user =\> 'logstash-producer' } } Pipeline of the **producer 2** input { heartbeat { interval =\> 1 type =\> "heartbeat" message =\> "%{@timestamp}" } } output { rabbitmq { host =\> 'rabbitmq:5672' exchange =\> 'events.sharded' exchange\_type =\> 'x-modulus-hash' key =\> 'producer-2' persistent =\> true workers =\> 1 durable =\> true password =\> '123456' user =\> 'logstash-producer' } } In the **rabbitmq** I added a policy `rabbitmqctl set_policy events-shard "^events.sharded$" '{"shards-per-node": 2}'` So, when I added an exchange of type x-modulus-hash, the rabbitmq create 2 queues ![exchanges](https://us1.discourse-cdn.com/elastic/original/3X/9/4/943a493582c61a831fee7df2dd93596a9b76a76b.jpeg) ![queues](https://us1.discourse-cdn.com/elastic/original/3X/a/7/a79d0d35f5341e317cdbd04b470a2e8f8d3df315.jpeg) Pipeline for the **consumer 1** and **consumer 2** input { rabbitmq { host =\> 'rabbitmq:5672' queue =\> 'events.sharded' passive =\> true durable =\> true prefetch\_count =\> 2 password =\> '123456' user =\> 'logstash-consumer' threads =\> 10 } } output { stdout {} } When I start the producers the behavior is ok, there are messages in both queues. ![connections-producers](https://us1.discourse-cdn.com/elastic/original/3X/1/9/198d1079a212104f4181558e84f28d6db7a81b72.jpeg) ![queues-incoming](https://us1.discourse-cdn.com/elastic/original/3X/1/6/167c40fc56761f3babf0a2b959908791fc9b5e15.jpeg) But, when I start the consumers, they only connect to the **first queue**. After start the first consumer ![connections-consumer-1](https://us1.discourse-cdn.com/elastic/original/3X/2/5/25a7e2ee9f5d303e971909ebe5ccc94440c3f380.jpeg) ![queues-delivery](https://us1.discourse-cdn.com/elastic/original/3X/d/4/d482ce26dd84bbffe0f4065016220888e26f41d2.jpeg) ![queue-0-consumer-1](https://us1.discourse-cdn.com/elastic/original/3X/2/a/2ae59b0d28a2a7c089876fd2bf2e34ce5e4a1033.jpeg) ![queue-1-whitout-consumer](https://us1.discourse-cdn.com/elastic/original/3X/a/e/aeaaea51a60d28047d32445127cc635f37adb5fd.jpeg) After start the second consumer ![connections-consumer-2](https://us1.discourse-cdn.com/elastic/original/3X/d/f/dfec2fd61ade9061dc20dbad37de94d1d10fe701.jpeg) ![queues-delivery-2](https://us1.discourse-cdn.com/elastic/original/3X/c/3/c3c63496f8517d8aa4126d357733bbfcaf2c0ed2.jpeg) ![queue-0-consumer-2](https://us1.discourse-cdn.com/elastic/original/3X/7/6/763b1c800a27dabbe8338c97b080a47ecd411607.jpeg) ![queue-1-whitout-consumer-2](https://us1.discourse-cdn.com/elastic/original/3X/c/c/cc76b9a6cb977715152458bcb22f35950e9a8419.jpeg) Any suggestion to do the consumers connect in both queues? ---
**Author:** ![yaauie](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yaauie/32/23363_2.png) [@yaauie](https://discuss.elastic.co/u/yaauie)\ **Post date:** [March 13, 2019, 11:45pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/2 "2019-03-13T23:45:37Z")
I am not an expert on RabbitMQ, but some of the caveats that are called out on the Sharding Plugin's docs are disconcerting: > How does it work? The plugin will chose the queue from the shard with the _least amount of consumers_ , provided the queue contents are local to the broker you are connected to. > > **NOTE: there's a small race condition between RabbitMQ updating the queue's internal stats about consumers and when clients issue `basic.consume` commands.** The problem with this is that if your client issue many `basic.consume` commands without too much time in between, it might happen that the plugin assigns the consumers to queues in an uneven way. > > -- [RabbitMQ Sharding Plugin](https://github.com/rabbitmq/rabbitmq-sharding#consuming-from-a-sharded-pseudo-queue) Essentially, the plugin does not guarantee that 100% of your pseudo-queues will be consumed, even if you create a sufficient quantity of consumers. ---
**Author:** ![lsdev14](https://avatars.discourse-cdn.com/v4/letter/l/e9c0ed/32.png) [@lsdev14](https://discuss.elastic.co/u/lsdev14)\ **Post date:** [March 16, 2019, 12:23am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/3 "2019-03-16T00:23:55Z")
Hey Biesemeyer thanks for answering. I read this comment of the rabbitmq plugin and I made other simple test. I started only one logstash, so, in the rabbitmq management, I can see 10 consumers in the first queue. ![logstash-consumer1](https://us1.discourse-cdn.com/elastic/original/3X/6/9/6906d9463590ba7cd9309959fb602e7187d4bd2d.jpeg) I waited some minutes and I started the second consumer. But it continues to connect only in the first queue. ![logstash-consumer2](https://us1.discourse-cdn.com/elastic/original/3X/f/c/fc8c752afd7317959c30a9dadd6cf7d04afdd30d.jpeg) So, I made another test with nodejs consumers. ``` var amqp = require('amqplib/callback_api'); amqp.connect('amqp://nodejs-consumer:1234@rabbitmq:5672', function (err, conn) { if (err) { console.error(err); return; } conn.createChannel(function (err, ch) { var q = 'events.sharded'; ch.assertQueue(q, { durable: true }); ch.prefetch(1); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q); ch.consume(q, function (msg) { console.log(" [x] Received %s", msg.content.toString()); }, { noAck: true }); }); }); ``` In this case, I can see the consumers distributed for each queue. ![nodejs-consumers](https://us1.discourse-cdn.com/elastic/original/3X/c/1/c1128ca1f069ea616cb8eea861f21e3201d1c9cd.jpeg) I tried to understand if there anything to settings in the logstash input-plugin, but I didn't find. I only found this open issue in the plugin repository - [https://github.com/logstash-plugins/logstash-input-rabbitmq/issues/29](https://github.com/logstash-plugins/logstash-input-rabbitmq/issues/29) ---
**Author:** ![yaauie](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yaauie/32/23363_2.png) [@yaauie](https://discuss.elastic.co/u/yaauie)\ **Post date:** [March 16, 2019, 12:31am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/4 "2019-03-16T00:31:08Z")
Again, this plugin is a RabbitMQ-internal thing, that is not supposed to expose itself in any way to consumers. It does not add any options to the APIs that the consumers use to connect. I don't know why it is failing to route consumers, but it is outside of the control of the Logstash RabbitMQ Input. The caveats in the documentation about the RabbitMQ Sharding plugin indicating that it does not guarantee that consumers will be routed to all queues would be enough for me to not use it in any production environment. ---
**Author:** ![lsdev14](https://avatars.discourse-cdn.com/v4/letter/l/e9c0ed/32.png) [@lsdev14](https://discuss.elastic.co/u/lsdev14)\ **Post date:** [March 16, 2019, 1:18am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/5 "2019-03-16T01:18:11Z")
Hey Biesemeyer thanks again for answering. I'm testing this rabbitmq plugin because I need to share the same type of messages between some queues. The idea is to increase the throughput and use more then one CPU for each rabbitmq node. I did other test, I created rabbitmq's queues manually. The problem is to setup a logstash pipeline for each queue, I need to specify only one queue name in the input settings. Do you have any suggestion to minimize this? ---
**Author:** ![lsdev14](https://avatars.discourse-cdn.com/v4/letter/l/e9c0ed/32.png) [@lsdev14](https://discuss.elastic.co/u/lsdev14)\ **Post date:** [March 16, 2019, 1:27am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/6 "2019-03-16T01:27:44Z")
I did another test to minimize the effort to keep some many pipelines. I added more then one entry for rabbitmq plugin in the same pipeline. input { rabbitmq { host =\> 'rabbitmq:5672' queue =\> 'events.sharded. **1**' passive =\> true durable =\> true password =\> '123456' user =\> 'logstash-consumer' threads =\> 10 } rabbitmq { host =\> 'rabbitmq:5672' queue =\> 'events.sharded. **2**' passive =\> true durable =\> true password =\> '123456' user =\> 'logstash-consumer' threads =\> 10 } rabbitmq { host =\> 'rabbitmq:5672' queue =\> 'events.sharded. **n**' passive =\> true durable =\> true password =\> '123456' user =\> 'logstash-consumer' threads =\> 10 } } output { stdout {} } Why do you think about it? ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 13, 2019, 1:27am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue-2/172178/7 "2019-04-13T01:27:48Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.