# Consume data from sharded RabbitMQ queue **URL:** \ **Category:** Logstash\ **Created:** [May 2, 2018, 12:27pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248 "2018-05-02T12:27:33Z")\ **Posts on this page:** 8\ **Page:** 1
**Author:** ![peter.ridzi](https://avatars.discourse-cdn.com/v4/letter/p/6de8d8/32.png) [@peter.ridzi](https://discuss.elastic.co/u/peter.ridzi)\ **Post date:** [May 2, 2018, 12:27pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/1 "2018-05-02T12:27:33Z")
Hi, I would like to set up a Logstash instance to consume data from a sharded RabbitMQ queue. I am using RabbitMQ 3.3.5 and Logstash 5.2 right now. I installed the rabbitmq-sharding plugin ([https://github.com/rabbitmq/rabbitmq-sharding](https://github.com/rabbitmq/rabbitmq-sharding)) as it is not included in this version. For RabbitMQ I set the policy to define the sharding on my test exchange: `rabbitmqctl set_policy shard_test_exchange-shard "^shard_test_exchange$" '{"shards-per-node": 7, "routing-key": "logstash"}' --apply-to exchanges` I loaded the data into the queue using the following Logstash config: `/usr/share/logstash/bin/logstash -e "input { file { start_position => 'beginning' path => '/home/myuser/testinput.txt' codec => 'plain' sincedb_path => '/opt/myapp/rabbitmqsharding' } } output{ rabbitmq { host => 'myhostname' exchange => 'shard_test_exchange' exchange_type => 'x-modulus-hash' key => '%{@timestamp}' persistent => true workers => 1 durable => true password => 'admin_password' user => 'admin_user' } }"` The exchange was created and I could see the sharded queue in RabbitMQ. Then I tried to run another Logstash instance and set up its input plugin to consume data from this sharded queue: `/usr/share/logstash/bin/logstash -e "input { rabbitmq { host => 'myhostname' queue => 'shard_test_exchange' passive => true prefetch_count => 4000 password => 'admin_password' user => 'admin_user' debug =>true } } output{ stdout{} }"` (The rabbitmq documentation sais that the queue name should be the name of the exchange.) I always get the following error message: `13:22:42.160 [[main]"#method(reply-code=541, reply-text=INTERNAL_ERROR - Interceptor: rabbit_sharding_interceptor failed with reason: \"Can't declare sharded queue: <<\\\"shard_test_exchange\\\">>\", class-id=50, method-id=10)", :class=>"MarchHare::ChannelAlreadyClosed", :location=>"/usr/share/logstash/vendor/bundle/jruby/1.9/gems/march_hare-2.22.0-java/lib/march_hare/exceptions.rb:121:in`convert\_and\_reraise'"}` It seems that the logstash rabbitmq input plugin tries to define the queue, but it can't, because it already exists. How can I configure the logstash to connect to the sharded queue? Thanks, Peter ---
**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:** [May 3, 2018, 1:02am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/2 "2018-05-03T01:02:47Z")
I believe that your input should mirror some of the settings from the output, namely `exchange` and `exchange_type`; I believe that the `queue` directive is a bit of a misnomer, in that it only names the local queue and is optional. ---
**Author:** ![peter.ridzi](https://avatars.discourse-cdn.com/v4/letter/p/6de8d8/32.png) [@peter.ridzi](https://discuss.elastic.co/u/peter.ridzi)\ **Post date:** [May 3, 2018, 7:23am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/3 "2018-05-03T07:23:07Z")
Hi @yaauie, Thank you for the idea, now I tried to add the `exchange => 'shard_test_exchange' exchange_type => 'x-modulus-hash'` parameters, but the results are the same. (I tried to delete the `queue`, and the `passive` too, but no luck.) It seems that the plugin tries to declare the queue (in active or passive mode) and somehow it doesn't compatible with sharded queues, can it be avoided somehow? In this case during the creation of the exchage the queue is created too, so when the input plugin (consumer) starts, it can use the existing queue. Due to performance issues I need to use sharding in rabbitmq, because it can't handle the velocity of incoming/outgoing data on 1 CPU (as rabbitmq uses 1 CPU per queue). I use multiple threads in the input plugin (consumers to get data from the queue), but without sharding it doesn't help me much. I don't think it is a unique problem, I hope that I just miss something in the paramterization. Do you have any idea, how can I solve this issue, and connect to the queue? ---
**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:** [May 4, 2018, 4:53pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/4 "2018-05-04T16:53:00Z")
Unfortunately, I do not know much about RabbitMQ, but the documentation for the plugin seems to indicate that the sharding is an _internal_ feature, and that consumers attach to the exchange in the normal way with a notable exception: you need at least as many consumers as shards, or else you risk messages from one partition falling behind. There is a bold note in the docs about _not_ declaring the same queue name in your consumer as in your producer as well: > ---
**Author:** ![peter.ridzi](https://avatars.discourse-cdn.com/v4/letter/p/6de8d8/32.png) [@peter.ridzi](https://discuss.elastic.co/u/peter.ridzi)\ **Post date:** [May 10, 2018, 8:04am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/5 "2018-05-10T08:04:42Z")
This is exacltly what I found, and thats why I thought that the logstash input plugin should handle it without any issues (if in the `queue` parameter I give it the name of the exchange). However it seems that the rabbitmq input plugin tries to create the queue in the `setup!` (in source code). The only option what I found was the `passive`, but it doesn't change anything. Do you know how can I modify the parameters, is there any way to avoid the queue creation? Or should I create a custom modified plugin to handle that case (ie. delete the call of `declare_queue!` from the code)? ---
**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:** [May 10, 2018, 8:11am UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/6 "2018-05-10T08:11:04Z")
I would at least open up an issue on the project indicating that it appears that its use of the RabbitMQ APIs is preventing it from working with queues that are sharded. If you can also hack on the project to figure out what bit is interfering, I'm sure our Logstash Integrations team will be able to assist in getting a pull-request into shape that solves your use-case without breaking everyone else's. That way, you won't have to maintain a diverged fork of the project or have the overhead of installing custom plugins on your production machines. ---
**Author:** ![Thomas\_Davis](https://avatars.discourse-cdn.com/v4/letter/t/90ced4/32.png) [@Thomas\_Davis](https://discuss.elastic.co/u/Thomas_Davis)\ **Post date:** [May 15, 2018, 3:06pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/7 "2018-05-15T15:06:49Z")
This works with logstash6.x. I just got it working. create the exchange, set the policy. For every queue that is created, create the following logstash entry: `input { rabbitmq { host => "rabbit" port => 5672 exchange => "sharded-exchange" queue => "sharding: sharded-exchange - rabbit@rabbit-1 - 0" passive => true durable => true user => "logstash" password => "password" codec => "json" } }` You change the 'queue' name to match exactly the sharded queue you want to pull from. If you have 2 rabbitmq hosts, and you've sharded 4 times, you need to repeat this 4 times, with each rabbit host & queue shard # in the queue name. ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [June 12, 2018, 3:06pm UTC](https://discuss.elastic.co/t/consume-data-from-sharded-rabbitmq-queue/130248/8 "2018-06-12T15:06:50Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.