# Missing document in Elasticsearch when using BulkProcessor

**URL:** <https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015>\
**Category:** Elasticsearch\
**Created:** [June 7, 2016, 5:02am UTC](https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015 "2016-06-07T05:02:31Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![spireofdreams](https://avatars.discourse-cdn.com/v4/letter/s/898d66/32.png) [@spireofdreams](https://discuss.elastic.co/u/spireofdreams)\
**Post date:** [June 7, 2016, 5:02am UTC](https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015/1 "2016-06-07T05:02:31Z")

</div>

I'm using a `[java] kafka-producer` to push data to kafka-topic _x_ and a `[java] high level consumer/bulkProcessor` to read from topic _x_ and index data to elasticsearch. The producer pushes 10 docs each time. When I start my java code for bulkProcessor for the first time after running producer, I see only 9 records being pushed to ES, all with `"_version": 1`. The 10th record is not in ES.

`beforeBulk()` and `afterBulk()` methods show the following results.

```
Going to execute new bulk composed of 10 actions
java.lang.StringIndexOutOfBoundsException: String index out of range: 0

```

The exception in `afterbulk()` is only for the first record. This moment onward, if I remove the elasticsearch index and use the producer, I see 10 records consistently. I have no idea why this is happening. Any help is appreciated.

Note: ES version 2.2.0  
Kafka: 0.9.0.0

**[relevant code]**

```
public Consumer(KafkaStream a_stream, int a_threadNumber, String esHost, String esCluster, int bulkSize, String topic) {
    
/*Create transport client*/
BulkProcessor bulkProcessor;
       
this.bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
    public void beforeBulk(long executionId, BulkRequest request) {
            System.out.format("Going to execute new bulk composed of %d actions\n", request.numberOfActions());
    }

    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            System.out.format("Executed bulk composed of %d actions\n", response.getItems().length);
            int i = 0;
            while(i < response.getItems().length) {
                System.out.println(response.getItems()[i].getFailureMessage());
                i++;
            }
    }

    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            System.out.format("Error executing bulk", failure);
    }
    }).setBulkActions(bulkSize) 
            .setBulkSize(new ByteSizeValue(200, ByteSizeUnit.MB)) 
            .setFlushInterval(TimeValue.timeValueSeconds(1))
            .build();
}

public void run() {        
    ConsumerIterator<byte[], byte[]> it = m_stream.iterator();   
    while (it.hasNext()) {
        byte[] x = it.next().message();
        try {             
            bulkProcessor.add(new IndexRequest(index, type, id.toString()).source(modifyMsg(x).toString()));
        } 
        catch (Exception e) {
            logger.warn("bulkProcessor failed: " + m_threadNumber + e.getMessage());
        }                   
    }
    logger.info("Shutting down Thread: " + m_threadNumber);
}

```

Docs going to ES are of the following form:

```
{"index":"temp1","type":"temp2","id":"0","event":"we're doomed"}
{"index":"temp1","type":"temp2","id":"1","event":"we're doomed"}
{"index":"temp1","type":"temp2","id":"2","event":"we're doomed"}
...
{"index":"temp1","type":"temp2","id":"9","event":"we're doomed"}

```

Even if I use auto-generated \_id in ES, the same issue persists.

**[EDIT]**

If I add the following line in my run() method the problem is gone.

```
public void run() {
    ...
    bulkProcessor.add(new IndexRequest("")); //Added this line
    while (it.hasNext()) {
        ...         
    }
    ...
}
```

---

<div class="post-metadata">

**Author:** ![spireofdreams](https://avatars.discourse-cdn.com/v4/letter/s/898d66/32.png) [@spireofdreams](https://discuss.elastic.co/u/spireofdreams)\
**Post date:** [June 8, 2016, 12:25pm UTC](https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015/2 "2016-06-08T12:25:22Z")

</div>

My bad. In the line `bulkProcessor.add(new IndexRequest(index, type, id.toString()).source(modifyMsg(x).toString()));` the method `modifyMsg()` was initializing `index`, `type` and `id`, which was set to empty string in the constructor. That's why my first index request was failing as it had invalid index name.

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [June 9, 2016, 12:40am UTC](https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015/3 "2016-06-09T00:40:35Z")

</div>

Thanks for sharing your resolution 🙂

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 5, 2017, 10:45pm UTC](https://discuss.elastic.co/t/missing-document-in-elasticsearch-when-using-bulkprocessor/52015/4 "2017-07-05T22:45:11Z")

</div>