# Missing records when using BulkProcessor (Java High Level Rest Client)
**URL:** https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625
**Category:** Elasticsearch
**Created:** [November 20, 2019, 6:37am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625 "2019-11-20T06:37:52Z")
**Posts on this page:** 14
**Page:** 1
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 20, 2019, 6:37am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/1 "2019-11-20T06:37:52Z")
Hi,
I use BulkProcessor to index records. I see that some of the records are missing when doing bulk indexing. I don't see any Exception or Errors, also, in afterBulk callback there are no failures. BulkProcessor is singleton spring bean so, I don't explicitly call awaitAndClose or flush. It would be of great help if somebody helps in guiding me to resolve the issue.
this.bulkProcessor = BulkProcessor.builder(
(request, bulkListener) -\> esClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
logger.debug("going to execute bulk of {} requests", request.numberOfActions());
}
```
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
logger.debug("bulk executed {} failures", response.hasFailures() ? "with" : "without");
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
logger.warn("error while executing bulk", failure);
}
})
.setBulkActions(1000)
.setFlushInterval(TimeValue.timeValueSeconds(5))
.build(); }
```
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 20, 2019, 6:50am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/2 "2019-11-20T06:50:40Z")
You probably need to close the bulk processor before exiting your application if your application is stopped at some point.
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 20, 2019, 7:00am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/3 "2019-11-20T07:00:23Z")
Our's is a continuous running application, which keeps indexing data as and when some event occurs. We will not be explicitly stopping the application, unless there is a deployment needed. Also, I call close on bulkprocessor when spring bean is about to be destroyed by the spring container.
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 20, 2019, 10:13am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/4 "2019-11-20T10:13:28Z")
One of the observation is if the application is running only one node then there are no missing records, but If I have my application running on multiple nodes, then I see that the records are missing.
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 20, 2019, 11:35am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/5 "2019-11-20T11:35:45Z")
The code you shared seems ok though. So it might be something else.
How are you checking for missing records?
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 20, 2019, 12:36pm UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/6 "2019-11-20T12:36:03Z")
For a given requestId I get the count from both elasticsearch as well as my database, both of them are not matching
POST /data\_v1/\_count
{
"query": {
"bool": {"must": [
{"term": {
"requestId": {
"value": "8f0bcc60-0b91-11ea-a375-1b01714de23a"
}
}}
]}
}
}
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 20, 2019, 3:29pm UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/7 "2019-11-20T15:29:11Z")
Are you waiting for the refresh to happen before calling this?
Or any chance you have duplicated Ids?
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 20, 2019, 10:18pm UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/8 "2019-11-20T22:18:13Z")
yes, I wait for refresh to happen before this. There is no chance of duplicate ids as this is working on single node.
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 21, 2019, 4:11am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/9 "2019-11-21T04:11:42Z")
Could you share the code part where you are adding the index request to the BulkProcessor ?
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 21, 2019, 4:47am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/10 "2019-11-21T04:47:39Z")
Below is the code, please let me know if this is good enough.
@Component
public class ElasticSearchTemplate implements InitializingBean {
```
@Autowired
private RestHighLevelClient restClient;
private BulkProcessor bulkProcessor;
@Value("${taxilla.es.initial.retry.wait.time:500}")
private int INITIAL_RETRY_WAIT_TIME;
@Value("${taxilla.es.bulk.num.of.actions:1000}")
private int BULK_NUM_OF_ACTIONS;
@Value("${taxilla.es.bulk.index.size:3}")
private int BULK_INDEX_SIZE;
@Value("${taxilla.es.bulk.flush.interval:5}")
private int BULK_FLUSH_INTERVAL;
@Value("${taxilla.es.index.failure.retry.count:5}")
private int INDEX_FAILURE_RETRY_COUNT;
private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchTemplate.class);
@Override
public void afterPropertiesSet() {
BulkProcessor.Builder bulkBuilder = BulkProcessor.builder(
(request, bulkListener) -> restClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
int numberOfActions = request.numberOfActions();
LOGGER.info("beforeBulk Executing bulk [{}] with [{}] requests and {} size", executionId, numberOfActions, request.estimatedSizeInBytes());
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
try {
if (response.hasFailures()) {
LOGGER.warn("afterBulk Bulk [{}] executed with failures :[{}]", executionId, response.buildFailureMessage());
} else {
LOGGER.info("afterBulk Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
}
} catch (Exception ex) {
LOGGER.error("after bulk error", ex);
}
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
LOGGER.error("afterBulk Error, Failed to execute bulk", failure);
}
})
.setBulkActions(BULK_NUM_OF_ACTIONS)
.setConcurrentRequests(1)
.setFlushInterval(TimeValue.timeValueSeconds(BULK_FLUSH_INTERVAL))
.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(INITIAL_RETRY_WAIT_TIME), INDEX_FAILURE_RETRY_COUNT));
this.bulkProcessor = bulkBuilder
.setBulkSize(new ByteSizeValue(BULK_INDEX_SIZE, ByteSizeUnit.MB))
.build();
}
public void refreshIndices() {
try {
restClient.indices().refresh(new RefreshRequest(SearchAppConstants.ASSET_INDEX_NAME), RequestOptions.DEFAULT);
} catch(IOException ex) {
LOGGER.error("Error while refreshing index : {}", ex.getMessage());
}
}
public void index(IndexDetails indexDetails) {
IndexRequest indexRequest = _index(indexDetails);
bulkProcessor.add(indexRequest);
}
public void delete(IndexDetails indexDetails) {
DeleteRequest deleteRequest = _delete(indexDetails);
bulkProcessor.add(deleteRequest);
}
private DeleteRequest _delete(IndexDetails indexDetails) {
return new DeleteRequest().id(indexDetails.getDocumentId())
.routing(indexDetails.getRoutingId()).index(indexDetails.getSearchAlias());
}
private IndexRequest _index(IndexDetails indexDetails) {
return new IndexRequest().routing(indexDetails.getRoutingId())
.index(indexDetails.getIndexName()).source(indexDetails.getObject())
.id(indexDetails.getDocumentId());
}
```
}
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 21, 2019, 5:34am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/11 "2019-11-21T05:34:38Z")
Are you sure that you are not sending documents with the same id? `indexDetails.getDocumentId()` can have duplicates?
---
### Author:  [@dawood\_abdullah](https://discuss.elastic.co/u/dawood_abdullah)
#### Post date: [November 21, 2019, 5:39am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/12 "2019-11-21T05:39:21Z")
yes I am sure, earlier I used highLevelRestClient.index() this used to work fine in the cluster, since it was too slow I moved to BulkProcessor. BulkProcessor is too fast, but now this problem of missing records.
Do I have to check anything with respect to write ThreadPools in Elasticsearch server. The problem is I donot see any rejections or errors from either elasticsearch server or my application.
---
### Author:  [@dadoonet](https://discuss.elastic.co/u/dadoonet)
#### Post date: [November 21, 2019, 6:36am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/13 "2019-11-21T06:36:33Z")
I don't see how this is happening.
I have been using the bulkProcessor for years now.
It would may be help if you can find a document that has not been indexed and try to index it manually to see if there are rejections?
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [December 19, 2019, 6:36am UTC](https://discuss.elastic.co/t/missing-records-when-using-bulkprocessor-java-high-level-rest-client/208625/14 "2019-12-19T06:36:43Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.