Bulk execution while a shard is replication might send erroneous version conflict failures for certain items

fixes elastic#2642

Author: kimchy

src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -133,6 +133,7 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
  Set<Tuple<String, String>> mappingsToUpdate = null;
 
  BulkItemResponse[] responses = new BulkItemResponse[request.items().length];
+ long[] versions = new long[request.items().length];
  for (int i = 0; i < request.items().length; i++) {
  BulkItemRequest item = request.items()[i];
  if (item.request() instanceof IndexRequest) {
@@ -163,6 +164,7 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
  version = create.version();
  op = create;
  }
+ versions[i] = indexRequest.version();
  // update the version on request so it will happen on the replicas
  indexRequest.version(version);
 
@@ -188,6 +190,10 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
  } catch (Exception e) {
  // rethrow the failure if we are going to retry on primary and let parent failure to handle it
  if (retryPrimaryException(e)) {
+ // restore updated versions...
+ for (int j = 0; j < i; j++) {
+ applyVersion(request.items()[j], versions[j]);
+ }
  throw (ElasticSearchException) e;
  }
  if (e instanceof ElasticSearchException && ((ElasticSearchException) e).status() == RestStatus.CONFLICT) {
@@ -214,6 +220,10 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
  } catch (Exception e) {
  // rethrow the failure if we are going to retry on primary and let parent failure to handle it
  if (retryPrimaryException(e)) {
+ // restore updated versions...
+ for (int j = 0; j < i; j++) {
+ applyVersion(request.items()[j], versions[j]);
+ }
  throw (ElasticSearchException) e;
  }
  if (e instanceof ElasticSearchException && ((ElasticSearchException) e).status() == RestStatus.CONFLICT) {
@@ -352,4 +362,14 @@ public void onFailure(Throwable e) {
  logger.warn("failed to update master on updated mapping for index [{}], type [{}]", e, index, type);
  }
  }
+
+ private void applyVersion(BulkItemRequest item, long version) {
+ if (item.request() instanceof IndexRequest) {
+ ((IndexRequest) item.request()).version(version);
+ } else if (item.request() instanceof DeleteRequest) {
+ ((DeleteRequest) item.request()).version(version);
+ } else {
+ // log?
+ }
+ }
 }

src/test/java/org/elasticsearch/test/integration/recovery/RelocationTests.java

@@ -121,20 +121,20 @@ public void testPrimaryRelocationWhileIndexingWith10RelocationAnd5Writers() thro
  testPrimaryRelocationWhileIndexing(10, 5, false);
  }
 
-// @Test
-// public void testPrimaryRelocationWhileBulkIndexingWith1RelocationAnd1Writer() throws Exception {
-// testPrimaryRelocationWhileIndexing(1, 1, true);
-// }
-//
-// @Test
-// public void testPrimaryRelocationWhileBulkIndexingWith10RelocationAnd1Writer() throws Exception {
-// testPrimaryRelocationWhileIndexing(10, 1, true);
-// }
-//
-// @Test
-// public void testPrimaryRelocationWhileBulkIndexingWith10RelocationAnd5Writers() throws Exception {
-// testPrimaryRelocationWhileIndexing(10, 5, true);
-// }
+ @Test
+ public void testPrimaryRelocationWhileBulkIndexingWith1RelocationAnd1Writer() throws Exception {
+ testPrimaryRelocationWhileIndexing(1, 1, true);
+ }
+
+ @Test
+ public void testPrimaryRelocationWhileBulkIndexingWith10RelocationAnd1Writer() throws Exception {
+ testPrimaryRelocationWhileIndexing(10, 1, true);
+ }
+
+ @Test
+ public void testPrimaryRelocationWhileBulkIndexingWith10RelocationAnd5Writers() throws Exception {
+ testPrimaryRelocationWhileIndexing(10, 5, true);
+ }
 
  private void testPrimaryRelocationWhileIndexing(final int numberOfRelocations, final int numberOfWriters, final boolean batch) throws Exception {
  logger.info("--> starting [node1] ...");
@@ -288,6 +288,21 @@ public void testReplicaRelocationWhileIndexingWith10RelocationAnd5Writers() thro
  testReplicaRelocationWhileIndexing(10, 5, false);
  }
 
+ @Test
+ public void testReplicaRelocationWhileBulkIndexingWith1RelocationAnd1Writer() throws Exception {
+ testReplicaRelocationWhileIndexing(1, 1, true);
+ }
+
+ @Test
+ public void testReplicaRelocationWhileBulkIndexingWith10RelocationAnd1Writer() throws Exception {
+ testReplicaRelocationWhileIndexing(10, 1, true);
+ }
+
+ @Test
+ public void testReplicaRelocationWhileBulkIndexingWith10RelocationAnd5Writers() throws Exception {
+ testReplicaRelocationWhileIndexing(10, 5, true);
+ }
+
  private void testReplicaRelocationWhileIndexing(final int numberOfRelocations, final int numberOfWriters, final boolean batch) throws Exception {
  logger.info("--> starting [node1] ...");
  startNode("node1");