BulkResponse throws EOFException

jason · April 12, 2011, 6:30pm

Hi everyone,

I am trying to send a bulk request to the elastic search, and I am
getting an exception. Please see the code and the stacktrace below:

                LinkedList<XContentBuilder> queue = .....//assume

the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet();  //throws

exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.

kimchy · April 12, 2011, 6:55pm

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.
On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.

jason · April 13, 2011, 12:03am

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -

kimchy · April 13, 2011, 7:53am

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.
On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(Â))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.Âjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂs.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂs.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.Âjava:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedÂKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.jÂava:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamiÂngRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -

jason · April 13, 2011, 9:22pm

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

kimchy · April 13, 2011, 9:48pm

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?
On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂs.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂs.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.ÂÂjava:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedÂÂKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.jÂÂava:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamiÂÂngRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

jason · April 13, 2011, 10:29pm

Here is the gist:

gist.github.com

https://gist.github.com/efurman/c1400655e0d8728cac49

BulkProcessorTest.java (version that fails)

package com.edmunds.training.real_time_logger;

import java.io.IOException;

import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

This file has been truncated. show original

BulkProcessorTest.java (working version)

package com.edmunds.training.real_time_logger;

import java.io.IOException;

import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

This file has been truncated. show original

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

kimchy · April 13, 2011, 10:46pm

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?
On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂÂÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂs.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂs.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.ÂÂÂjava:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedÂÂÂKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.jÂÂÂava:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamiÂÂÂngRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

jason · April 13, 2011, 10:50pm

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

jason · April 13, 2011, 10:52pm

Oops, I made a mistake. I fixed it and updated the gist, which still
didn't eliminate the exception but the logic should be correct now.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

kimchy · April 13, 2011, 10:58pm

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.
On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂÂÂÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂs.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂs.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.ÂÂÂÂjava:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedÂÂÂÂKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.jÂÂÂÂava:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamiÂÂÂÂngRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

If you have encountered such an error in the past, could you please
suggest what the most probable cause of it?

Thank you,
Eugene.- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -- Hide quoted text -

Show quoted text -

jason · April 14, 2011, 1:14am

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

...

read more »- Hide quoted text -

Show quoted text -

kimchy · April 14, 2011, 9:06am

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.
On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂÂÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂÂÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂÂÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂÂÂÂÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂs.java:
274)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂs.java:
261)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.ÂÂÂÂÂjava:
349)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.processSelectedÂÂÂÂÂKeys(NioWorker.java:
280)
at
org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.jÂÂÂÂÂava:
200)
at
org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamiÂÂÂÂÂngRunnable.java:
108)
at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker
$1.run(DeadLockProofWorker.java:44)
at java.util.concurrent.ThreadPoolExecutor
$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor
$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

...

read more Â»- Hide quoted text -

Show quoted text -

jason · April 14, 2011, 5:41pm

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
274)
at

...

read more »- Hide quoted text -

Show quoted text -

kimchy · April 14, 2011, 5:57pm

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request
}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂÂÂÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂÂÂÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂÂÂÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannÂÂÂÂÂÂelsHandler.java:
51)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂDefaultChannelPipeline.java:
545)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂDefaultChannelPipeline.java:
540)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂÂs.java:
274)
at

...

read more Â»- Hide quoted text -

Show quoted text -

jason · April 14, 2011, 6:12pm

This is the part that does not work for me. Have you tried the
example from the gist? Please let me know if you see anything wrong in
my code.

On Apr 14, 10:57 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request

}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)

...

read more »- Hide quoted text -

Show quoted text -

kimchy · April 14, 2011, 6:35pm

You are not doing what I wrote in any version of your gist. Not sure where the disconnect is.... To go the extra mile, I took your code and gisted how it should work: 920160’s gists · GitHub.
On Thursday, April 14, 2011 at 9:12 PM, eugene wrote:

This is the part that does not work for me. Have you tried the
example from the gist? Please let me know if you see anything wrong in
my code.

On Apr 14, 10:57 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request

}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂÂÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂÂÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂÂÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂÂÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂÂÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂÂÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at
org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(ChannelÂÂÂÂÂÂÂs.java:
302)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFiÂÂÂÂÂÂÂreMessageReceived(FrameDecoder.java:
317)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(ÂÂÂÂÂÂÂFrameDecoder.java:
299)
at
org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceÂÂÂÂÂÂÂived(FrameDecoder.java:
216)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)

...

read more Â»- Hide quoted text -

Show quoted text -

jason · April 14, 2011, 8:18pm

Thank you very much for posting an example. I am sorry to keep
bugging you, but this code is not working for me either. I think what
I wrote in my version and what you suggested is the same. You are
recreating XContentBuilder within the loop, and I am adding only two
items so no loop is needed.
Anyway, could you try running your code and let me know if it runs
successfully?

Thanks,
Eugene.

On Apr 14, 11:35 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You are not doing what I wrote in any version of your gist. Not sure where the disconnect is.... To go the extra mile, I took your code and gisted how it should work:920160’s gists · GitHub.

On Thursday, April 14, 2011 at 9:12 PM, eugene wrote:

This is the part that does not work for me. Have you tried the
example from the gist? Please let me know if you see anything wrong in
my code.

On Apr 14, 10:57 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request

}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessageChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at

...

read more »- Hide quoted text -

Show quoted text -

kimchy · April 14, 2011, 8:51pm

Yes, it runs fine for me. Which version are you using? I ran it on master, but nothing specifically changed in that area from 0.15.2.

-shay.banon
On Thursday, April 14, 2011 at 11:18 PM, eugene wrote:

Thank you very much for posting an example. I am sorry to keep
bugging you, but this code is not working for me either. I think what
I wrote in my version and what you suggested is the same. You are
recreating XContentBuilder within the loop, and I am adding only two
items so no loop is needed.
Anyway, could you try running your code and let me know if it runs
successfully?

Thanks,
Eugene.

On Apr 14, 11:35 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You are not doing what I wrote in any version of your gist. Not sure where the disconnect is.... To go the extra mile, I took your code and gisted how it should work:920160’s gists · GitHub.

On Thursday, April 14, 2011 at 9:12 PM, eugene wrote:

This is the part that does not work for me. Have you tried the
example from the gist? Please let me know if you see anything wrong in
my code.

On Apr 14, 10:57 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request

}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType(ÂÂÂÂÂÂÂÂ))
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.ÂÂÂÂÂÂÂÂjava:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamÂÂÂÂÂÂÂÂInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequeÂÂÂÂÂÂÂÂst.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)
at
org.elasticsearch.transport.netty.MessageChannelHandler.handleRequest(MessaÂÂÂÂÂÂÂÂgeChannelHandler.java:
180)
at
org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MesÂÂÂÂÂÂÂÂsageChannelHandler.java:
85)
at
org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUÂÂÂÂÂÂÂÂpstream(SimpleChannelUpstreamHandler.java:
80)
at
org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(ÂÂÂÂÂÂÂÂDefaultChannelPipeline.java:
545)
at org.elasticsearch.common.netty.channel.DefaultChannelPipeline
$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:
754)
at

...

read more Â»- Hide quoted text -

Show quoted text -

jason · April 14, 2011, 8:55pm

I am using 0.16.0-snapshot version. I see two variables builder and
builder2 in your example. Are you adding both to BulkRequestBuilder?

On Apr 14, 1:51 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Yes, it runs fine for me. Which version are you using? I ran it on master, but nothing specifically changed in that area from 0.15.2.

-shay.banon

On Thursday, April 14, 2011 at 11:18 PM, eugene wrote:

Thank you very much for posting an example. I am sorry to keep
bugging you, but this code is not working for me either. I think what
I wrote in my version and what you suggested is the same. You are
recreating XContentBuilder within the loop, and I am adding only two
items so no loop is needed.
Anyway, could you try running your code and let me know if it runs
successfully?

Thanks,
Eugene.

On Apr 14, 11:35 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You are not doing what I wrote in any version of your gist. Not sure where the disconnect is.... To go the extra mile, I took your code and gisted how it should work:920160’s gists · GitHub.

On Thursday, April 14, 2011 at 9:12 PM, eugene wrote:

This is the part that does not work for me. Have you tried the
example from the gist? Please let me know if you see anything wrong in
my code.

On Apr 14, 10:57 am, Shay Banon shay.ba...@elasticsearch.com wrote:

You can use several index requests thats not a problem, the flow should be:

for (item in items) {
XCnotentBuilder builder = XContentFactory....
// build json using item
// add to bulk request the index request

}
On Thursday, April 14, 2011 at 8:41 PM, eugene wrote:

I just recreated the simple test, and came up with the code that
works. I updated the gist and you may take a look.

It seems that I need to aggregate XContentBuilder first by adding
objects to it, and only at the end add XContentBuilder source to the
BulkRequestBuilder. Is this an intended behavior? Why can't I add
multiple IndexRequests with each having its own XContentBuilder
source?

On Apr 14, 2:06 am, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't know, can't see something that jumps out (but I might miss something). If you can recreate it with a simple test case, for example, something that just bulk indexes dummy data (or actual data, but it should bot be bigger than a main class), then I can have a look.

On Thursday, April 14, 2011 at 4:14 AM, eugene wrote:

I updated the gist following your approach, and unfortunately, still
coudn't get rid of this exception. What do you think I am doing
wrong?

On Apr 13, 3:58 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

Also, your code is not thread safe. You are using a LinkedList and access it from different threads without proper sync. Even if you sync on the methods of ES handler it won't help.

What you want to do is have a LinkedBlockingQueue as the message passing queue (better one to use is LinkedTransferQueue, but thats in Java7, though ES comes with one, so you can use it). Have the trailer push data into the queue, and have the processor poll for data. Once it gets a new element, add it to the BulkRequestBuilder. Then, using the BulkREquestBuilder#numberOfActions you can "flush" that bulk once it passes a specified bulk size, reset to a new BulkRequestBuilder, and continue from there.

I should probably write a helper (thread safe) BulkProcessor class that does that behind the scenes, and possibly be able to allow to control the number of concurrent bulk requests allowed, bulk size / actions, and so on, so things will be simpler.

On Thursday, April 14, 2011 at 1:50 AM, eugene wrote:

Oops, I made a mistake. I corrected it and updated the gist, and the
exception occurs.

On Apr 13, 3:46 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

I don't understand your logic there in the LogProcessor run method. You create a single builder object, and then index it "bulkSize" times? Shouldn't you be index "bulkSize" different sources?

On Thursday, April 14, 2011 at 1:29 AM, eugene wrote:

Here is the gist:

BulkProcessorTest.java (version that fails) · GitHub

This is a private gist, I didn't put the output.log file and
files.properties because they contains production data. Please let me
know if you need more information.

Thank you,
Eugene.

On Apr 13, 2:48 pm, Shay Banon shay.ba...@elasticsearch.com wrote:

When you add an IndexRequest to the BulkRequestBuilder, it copies over the buffer. You should not see that in this case. Can you gist a sample of the code you use?

On Thursday, April 14, 2011 at 12:22 AM, eugene wrote:

Thanks Shay, it's a good advice. However, I am still getting this
exception. How can I avoid buffer reuse, since when I aggregate
objects using BulkRequestBuilder, I am still adding new IndexRequest
objects to the builder and populate XContentBuilder.
Also, can it be happening because I am not assigning id to each
IndexRequest? I checked and the id is set to null for all of them for
some reason.

On Apr 13, 12:53 am, Shay Banon shay.ba...@elasticsearch.com wrote:

BulkRequestBuilder is not thread safe. If you want to do that, then you can Use the XContetFactory#safeXXX methods to thread safe builders and use the original code you posted, or, do the conversion on the thread that adds elements to the bulk request.

On Wednesday, April 13, 2011 at 3:03 AM, eugene wrote:

I tried to add requests to BuildRequestBuilder but the exception is
still not going away. My program is multithreaded, where first thread
adds objects to to the queue, and another thread retrieves those
objects from the queue and indexes them as bulk. The json objects can
be thought of IndexRequest objects. What do you think will be the
best way to solve this problem? It is really not complex, but
BulkResponse behaves strangely for me.

On Apr 12, 11:55 am, Shay Banon shay.ba...@elasticsearch.com wrote:

Don't use a list of XContentBuilders, there is no need for it (and what causes the problem because of buffers reuse). Simply use prepareBulk, save the BulkRequestBuilder, and use that as your aggregator.

On Tuesday, April 12, 2011 at 9:30 PM, eugene wrote:

Hi everyone,

I am trying to send a bulk request to the Elasticsearch, and I am
getting an exception. Please see the code and the stacktrace below:

LinkedList queue = .....//assume
the queue is populated
BulkRequestBuilder request = client.prepareBulk();
int bulkSize = 2;
for (int i = 0; i < bulkSize; i++) {
XContentBuilder jsonObj = queue.remove();

request.add(Requests.indexRequest(handler.getIndex()).type(handler.getType())
.source(jsonObj));
}

BulkResponse response = request.execute().actionGet(); //throws
exception at this line
if (response.hasFailures()) {
System.err.println("--> failures...");
}

Exception in thread "Thread-1"
org.elasticsearch.transport.RemoteTransportException: [Black Cat]
[inet[/10.32.40.17:9300]][indices/bulk]
Caused by: java.io.EOFException
at
org.elasticsearch.common.io.stream.LZFStreamInput.readBytes(LZFStreamInput.java:
97)
at
org.elasticsearch.common.io.stream.StreamInput.readUTF(StreamInput.java:
123)
at
org.elasticsearch.common.io.stream.HandlesStreamInput.readUTF(HandlesStreamInput.java:
49)
at
org.elasticsearch.action.support.replication.ShardReplicationOperationRequest.readFrom(ShardReplicationOperationRequest.java:
132)
at
org.elasticsearch.action.index.IndexRequest.readFrom(IndexRequest.java:
573)
at
org.elasticsearch.action.bulk.BulkRequest.readFrom(BulkRequest.java:
261)

...

read more »- Hide quoted text -

Show quoted text -

Topic		Replies	Views
Java client problem EOFException Elasticsearch	1	282	July 6, 2017
Getting exception while doing bulk insert Elasticsearch	3	425	July 6, 2017
Java 8.1 bulk request Elasticsearch	7	3566	May 24, 2022
ES 6.4.2 - MapperParsingException when performing Bulk requests Elasticsearch	4	243	October 12, 2022
Using transport bulk from java Elasticsearch	2	717	April 19, 2018

BulkResponse throws EOFException

Related topics