Missing some docs when using bulkprocessor! Is it a bug?

Elastic Stack Elasticsearch
1

I was using Bulkprocessor to index some documents (1005), but I only ended up with 1003 in ES. Also, this value could change when I change the settings of bulkprocessor (setConcurrentRequests to 10). Can someone help me out? Thanks. Below is my code,

bulkProcessor = BulkProcessor.builder(
...)
.setBulkActions(1000)
.setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
.setConcurrentRequests(1)
.build();

IndexRequest ir = new IndexRequest(config.get("indexName"), outputType).source(
jsonBuilder().startObject()
.field("SessionID", entry.getKey()).field("SessionURL", sessionURL)
.endObject());

es.bulkProcessor.add(ir);

2

How did you get the numbers? Like are you counting the numbers in the
listener or are you counting after you shut down the processor? If you are
counting after, are you refreshing? That is what I'd check first.

After that I'd try to make a minimal recreation. Maybe that'd find a
mistake you made or, if not, it'd be way easier to work from for us to
reproduce the bug.

3

Thanks! The original number was counted in the loop that I used to add indexRequest to Bulkprocessor. The result number was counted in Sense after I shut down both the bulkprocessor and node.

bulkProcessor.awaitClose(20, TimeUnit.MINUTES);
node.client().admin().indices().prepareRefresh().execute().actionGet();
node.close();

4

Do you flush the bulk processor ?

Because here, bulk processor send request to Elasticsearch every 1000 docs or every 1gb.

I personally use a flush every 5 seconds with setFlushInterval.

5

Good point. I should try that. But the reason I didn't use flush before is that I have two concerns.

  1. I think bulkProcessor.awaitClose(20, TimeUnit.MINUTES) would flush all remaining docs by itself, according to the documentation.

  2. Also, does setFlushInterval sort of conflict with setBulkSize() and SetBuilActions()?

Thanks,