# ES with RabbitMQ river and message lost

**URL:** <https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426>\
**Category:** Elasticsearch\
**Created:** [November 15, 2013, 5:00am UTC](https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426 "2013-11-15T05:00:11Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Guozhu\_Wei](https://avatars.discourse-cdn.com/v4/letter/g/cc9497/32.png) [@Guozhu\_Wei](https://discuss.elastic.co/u/Guozhu_Wei)\
**Post date:** [November 15, 2013, 5:00am UTC](https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426/1 "2013-11-15T05:00:11Z")

</div>

I use ES with RabbitMQ river to index about 120,000 docs. I write a python  
client to publish the docs to RabbitMQ in a speed ~300 docs per second and  
make the messages durable.

If the ES is running, the indexing process seems smoothly and no message is  
lost.

I want to test whether the messages will be lost in a higher publish rate.  
Below is how I tested:

1. Stopped the es, but continues to write all 120,000 messages to the queue;
2. When all the messages are sent to the queue, start the es again;

In this situation, the publish rate is about 5,000 docs per seconds, and  
~70,000 messages are lost. I found exceptions in my log file:

message [EsRejectedExecutionException[rejected execution (queue capacity  
50) on org.elasticsearch.action.support.replication.TransportShardReplica  
tionOperationAction$AsyncShardOperationAction$1@67a0e65]]

I think that may be due to the huge publish rate. So I change some  
parameters(num\_prefetch and num\_consumers, which is new in nightly build)  
refer to  
[https://github.com/elasticsearch/elasticsearch-river-rabbitmq/pull/11](https://github.com/elasticsearch/elasticsearch-river-rabbitmq/pull/11) and  
test again. But still 50,000 messages lost. I also changed the bulk\_size  
and bulk\_timeout, and it doesn't work too.

Any suggestion to solve this problem? Thanks!

Below is the river config:

PUT /\_river/my\_river/\_meta  
{  
"type": "rabbitmq",  
"rabbitmq": {  
"host": "192.168.18.182",  
"port": 5672,  
"user": "guest",  
"pass": "guest",  
"vhost": "/",  
"queue": "elasticsearch",  
"exchange": "elasticsearch",  
"routing\_key": "elasticsearch",  
"exchange\_type": "direct",  
"exchange\_durable": true,  
"queue\_durable": true,  
"queue\_auto\_delete": false,  
"num\_prefetch": 200,  
"num\_consumers": 1  
},  
"index": {  
"bulk\_size": 100,  
"bulk\_timeout": "10s",  
"ordered": false  
}  
}

--  
You received this message because you are subscribed to the Google Groups "elasticsearch" group.  
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).  
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).

---

<div class="post-metadata">

**Author:** ![Matt\_Michalowski](https://avatars.discourse-cdn.com/v4/letter/m/b2d939/32.png) [@Matt\_Michalowski](https://discuss.elastic.co/u/Matt_Michalowski)\
**Post date:** [November 15, 2013, 6:25am UTC](https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426/2 "2013-11-15T06:25:47Z")

</div>

On Friday, November 15, 2013 4:00:11 PM UTC+11, [weig...@gmail.com](mailto:weig...@gmail.com) wrote:

> I use ES with RabbitMQ river to index about 120,000 docs. I write a python  
> client to publish the docs to RabbitMQ in a speed ~300 docs per second and  
> make the messages durable.
> 
> If the ES is running, the indexing process seems smoothly and no message  
> is lost.
> 
> I want to test whether the messages will be lost in a higher publish rate.  
> Below is how I tested:
> 
> 1. Stopped the es, but continues to write all 120,000 messages to the  
> queue;
> 2. When all the messages are sent to the queue, start the es again;
> 
> In this situation, the publish rate is about 5,000 docs per seconds, and  
> ~70,000 messages are lost. I found exceptions in my log file:
> 
> message [EsRejectedExecutionException[rejected execution (queue capacity  
> 50) on org.elasticsearch.action.support.replication.TransportShardReplica  
> tionOperationAction$AsyncShardOperationAction$1@67a0e65]]
> 
> I think that may be due to the huge publish rate. So I change some  
> parameters(num\_prefetch and num\_consumers, which is new in nightly build)  
> refer to  
> [Proposed fix for ES crashing with large indexing job by akroston · Pull Request #11 · elastic/elasticsearch-river-rabbitmq · GitHub](https://github.com/elasticsearch/elasticsearch-river-rabbitmq/pull/11) and  
> test again. But still 50,000 messages lost. I also changed the bulk\_size  
> and bulk\_timeout, and it doesn't work too.
> 
> Any suggestion to solve this problem? Thanks!
> 
> Below is the river config:
> 
> PUT /\_river/my\_river/\_meta  
> {  
> "type": "rabbitmq",  
> "rabbitmq": {  
> "host": "192.168.18.182",  
> "port": 5672,  
> "user": "guest",  
> "pass": "guest",  
> "vhost": "/",  
> "queue": "elasticsearch",  
> "exchange": "elasticsearch",  
> "routing\_key": "elasticsearch",  
> "exchange\_type": "direct",  
> "exchange\_durable": true,  
> "queue\_durable": true,  
> "queue\_auto\_delete": false,  
> "num\_prefetch": 200,  
> "num\_consumers": 1  
> },  
> "index": {  
> "bulk\_size": 100,  
> "bulk\_timeout": "10s",  
> "ordered": false  
> }  
> }

I think you need to increase your 'threadpool.bulk.queue\_size' (and  
possibly 'threadpool.index.queue\_size') setting due to recent defaults.

See:

> <https://github.com/elastic/elasticsearch/issues/3888>
>
> Now that we properly fixed the ability to set the queue size on the index / bulk… thread pool, we should actually set them to a somehow reasonable value to protect from users potentially overflowing our system.
> 
> I suggest defaults to be 50 for bulk, and 200 for indexing.
> 
> Also, set the thread pool for get, which we should set (in a similar value to a "read" queue size we have today).

Matt.

--  
You received this message because you are subscribed to the Google Groups "elasticsearch" group.  
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).  
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).

---

<div class="post-metadata">

**Author:** ![Guozhu\_Wei](https://avatars.discourse-cdn.com/v4/letter/g/cc9497/32.png) [@Guozhu\_Wei](https://discuss.elastic.co/u/Guozhu_Wei)\
**Post date:** [November 15, 2013, 8:03am UTC](https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426/3 "2013-11-15T08:03:39Z")

</div>

Thanks!

I changed the code to test the Qos and build the plugin, now the Qos works  
and nothing lost now. And I think change the thread pool size as you said  
will also take effect.

```
private class Consumer implements Runnable {

    private Connection connection;

    private Channel channel;

    @Override
    public void run() {
        while (true) {
            if (closed) {
                break;
            }
            try {
                connection = 

```

connectionFactory.newConnection(rabbitAddresses);  
channel = connection.createChannel();  
channel.basicQos(200);

I didn’t dig into the code, but I think if the indexer raise an exception,  
the river should not simply ack the message.

Thank you for your response!

Gary.

在 2013年11月15日星期五UTC+8下午2时25分47秒，Matt Michalowski写道：

> On Friday, November 15, 2013 4:00:11 PM UTC+11, [weig...@gmail.com](mailto:weig...@gmail.com) wrote:
> 
> > I use ES with RabbitMQ river to index about 120,000 docs. I write a  
> > python client to publish the docs to RabbitMQ in a speed ~300 docs per  
> > second and make the messages durable.
> > 
> > If the ES is running, the indexing process seems smoothly and no message  
> > is lost.
> > 
> > I want to test whether the messages will be lost in a higher publish  
> > rate. Below is how I tested:
> > 
> > 1. Stopped the es, but continues to write all 120,000 messages to the  
> > queue;
> > 2. When all the messages are sent to the queue, start the es again;
> > 
> > In this situation, the publish rate is about 5,000 docs per seconds, and  
> > ~70,000 messages are lost. I found exceptions in my log file:
> > 
> > message [EsRejectedExecutionException[rejected execution (queue capacity  
> > 50) on org.elasticsearch.action.support.replication.TransportShardReplica  
> > tionOperationAction$AsyncShardOperationAction$1@67a0e65]]
> > 
> > I think that may be due to the huge publish rate. So I change some  
> > parameters(num\_prefetch and num\_consumers, which is new in nightly build)  
> > refer to  
> > [Proposed fix for ES crashing with large indexing job by akroston · Pull Request #11 · elastic/elasticsearch-river-rabbitmq · GitHub](https://github.com/elasticsearch/elasticsearch-river-rabbitmq/pull/11) and  
> > test again. But still 50,000 messages lost. I also changed the bulk\_size  
> > and bulk\_timeout, and it doesn't work too.
> > 
> > Any suggestion to solve this problem? Thanks!
> > 
> > Below is the river config:
> > 
> > PUT /\_river/my\_river/\_meta  
> > {  
> > "type": "rabbitmq",  
> > "rabbitmq": {  
> > "host": "192.168.18.182",  
> > "port": 5672,  
> > "user": "guest",  
> > "pass": "guest",  
> > "vhost": "/",  
> > "queue": "elasticsearch",  
> > "exchange": "elasticsearch",  
> > "routing\_key": "elasticsearch",  
> > "exchange\_type": "direct",  
> > "exchange\_durable": true,  
> > "queue\_durable": true,  
> > "queue\_auto\_delete": false,  
> > "num\_prefetch": 200,  
> > "num\_consumers": 1  
> > },  
> > "index": {  
> > "bulk\_size": 100,  
> > "bulk\_timeout": "10s",  
> > "ordered": false  
> > }  
> > }
> 
> I think you need to increase your 'threadpool.bulk.queue\_size' (and  
> possibly 'threadpool.index.queue\_size') setting due to recent defaults.
> 
> See:  
> [Set queue sizes by default on bulk/index thread pools · Issue #3888 · elastic/elasticsearch · GitHub](https://github.com/elasticsearch/elasticsearch/issues/3888)
> 
> Matt.

--  
You received this message because you are subscribed to the Google Groups "elasticsearch" group.  
To unsubscribe from this group and stop receiving emails from it, send an email to [elasticsearch+unsubscribe@googlegroups.com](mailto:elasticsearch+unsubscribe@googlegroups.com).  
For more options, visit [https://groups.google.com/groups/opt\_out](https://groups.google.com/groups/opt_out).

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 2:06am UTC](https://discuss.elastic.co/t/es-with-rabbitmq-river-and-message-lost/14426/4 "2017-07-06T02:06:48Z")

</div>