import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
private void handleUpdateFailureWithRetry(final ActionListener<UpdateResponse> listener, final UpdateRequest request,
                                          final Exception failure, int retryCount) {
    final Throwable cause = unwrapCause(failure);
    if (cause instanceof VersionConflictEngineException) {
        if (retryCount < request.retryOnConflict()) {
            logger.trace("Retry attempt [{}] of [{}] on version conflict on [{}][{}][{}]",
                    retryCount + 1, request.retryOnConflict(), request.index(), request.getShardId(), request.id());
            threadPool.executor(executor()).execute(new ActionRunnable<UpdateResponse>(listener) {
                @Override
                protected void doRun() {
                    shardOperation(request, listener, retryCount + 1);
                }
            });
            return;
        }
    }
    listener.onFailure(cause instanceof Exception ? (Exception) cause : new NotSerializableExceptionWrapper(cause));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testExternalVersioningInitialDelete() throws Exception {
    createIndex("test");
    ensureGreen();

    // Note - external version doesn't throw version conflicts on deletes of non existent records. This is different from internal versioning

    DeleteResponse deleteResponse = client().prepareDelete("test", "type", "1").setVersion(17).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertEquals(DocWriteResponse.Result.NOT_FOUND, deleteResponse.getResult());

    // this should conflict with the delete command transaction which told us that the object was deleted at version 17.
    assertThrows(
            client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(13).setVersionType(VersionType.EXTERNAL).execute(),
            VersionConflictEngineException.class
    );

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(18).
            setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(18L));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
@Override
public void putIfAbsent(Value value, Handler<AsyncResult<Void>> handler) {
    client.prepareIndex(context.database(), context.collection(), value.getId())
            .setSource(context.toJson(value).encode(), XContentType.JSON)
            .setOpType(IndexRequest.OpType.CREATE)
            .execute(new ElasticHandler<>(response -> {

                if (response.getResult().equals(DocWriteResponse.Result.CREATED)) {
                    handler.handle(result());
                } else {
                    handler.handle(error(new ValueAlreadyPresentException(value.getId())));
                }
            }, exception -> {
                if (nested(exception) instanceof VersionConflictEngineException) {
                    handler.handle(error(new ValueAlreadyPresentException(value.getId())));
                } else {
                    handler.handle(error(exception));
                }
            }));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
@Test(expected = VersionConflictEngineException.class)
public void testVersionBadUpdates() {
	TestVersionedObject testVersionedObject = new TestVersionedObject();
	testVersionedObject.name = "one";
	Key key1 = getGDS().save(testVersionedObject).now();
	
	Assert.assertNotNull(testVersionedObject.id);
	Assert.assertEquals(1, testVersionedObject.ver);
	
	testVersionedObject.name = "two";
	Key key2 = getGDS().save(testVersionedObject).now();
	
	Assert.assertEquals(2, testVersionedObject.ver);
	Assert.assertEquals(key1.id, key2.id);
	
	testVersionedObject.ver = 1;
	getGDS().save(testVersionedObject).now();
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testEngineGCDeletesSetting() throws InterruptedException {
    createIndex("test");
    client().prepareIndex("test", "type", "1").setSource("f", 1).get(); // set version to 1
    client().prepareDelete("test", "type", "1").get(); // sets version to 2
    // delete is still in cache this should work & set version to 3
    client().prepareIndex("test", "type", "1").setSource("f", 2).setVersion(2).get();
    client().admin().indices().prepareUpdateSettings("test").setSettings(Settings.builder().put("index.gc_deletes", 0)).get();

    client().prepareDelete("test", "type", "1").get(); // sets version to 4
    Thread.sleep(300); // wait for cache time to change TODO: this needs to be solved better. To be discussed.
    // delete is should not be in cache
    assertThrows(client().prepareIndex("test", "type", "1").setSource("f", 3).setVersion(4), VersionConflictEngineException.class);

}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testInternalVersioningInitialDelete() throws Exception {
    createIndex("test");
    ensureGreen();

    assertThrows(client().prepareDelete("test", "type", "1").setVersion(17).execute(),
            VersionConflictEngineException.class);

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1")
            .setCreate(true).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(1L));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testSimpleVersioningWithFlush() throws Exception {
    createIndex("test");
    ensureGreen();

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(1L));

    client().admin().indices().prepareFlush().execute().actionGet();

    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_2").setVersion(1).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(2L));

    client().admin().indices().prepareFlush().execute().actionGet();

    assertThrows(client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(1).execute(),
            VersionConflictEngineException.class);

    assertThrows(client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(1).execute(),
            VersionConflictEngineException.class);

    assertThrows(client().prepareIndex("test", "type", "1").setCreate(true).setSource("field1", "value1_1").execute(),
            VersionConflictEngineException.class);

    assertThrows(client().prepareDelete("test", "type", "1").setVersion(1).execute(), VersionConflictEngineException.class);
    assertThrows(client().prepareDelete("test", "type", "1").setVersion(1).execute(), VersionConflictEngineException.class);

    client().admin().indices().prepareRefresh().execute().actionGet();
    for (int i = 0; i < 10; i++) {
        assertThat(client().prepareGet("test", "type", "1").execute().actionGet().getVersion(), equalTo(2L));
    }

    for (int i = 0; i < 10; i++) {
        SearchResponse searchResponse = client().prepareSearch().setQuery(matchAllQuery()).setVersion(true).execute().actionGet();
        assertThat(searchResponse.getHits().getAt(0).getVersion(), equalTo(2L));
    }
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testUpdateReplicaRequestWithConflictFailure() throws Exception {
    DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
    BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);

    Exception err = new VersionConflictEngineException(shardId, "type", "id", "I'm conflicted <(;_;)>");
    Engine.IndexResult indexResult = new Engine.IndexResult(err, 0, 0);
    BulkItemResultHolder failedResults = new BulkItemResultHolder(null, indexResult, replicaRequest);

    Translog.Location location = new Translog.Location(0, 0, 0);
    BulkItemRequest[] items = new BulkItemRequest[0];
    BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
    Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(failedResults,
            DocWriteRequest.OpType.UPDATE, location, bulkShardRequest);

    BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();

    // Since this was not a conflict failure, the primary response
    // should be filled out with the failure information
    assertThat(newLocation, equalTo(location));
    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.INDEX));
    assertTrue(primaryResponse.isFailed());
    assertThat(primaryResponse.getFailureMessage(), containsString("I'm conflicted <(;_;)>"));
    BulkItemResponse.Failure failure = primaryResponse.getFailure();
    assertThat(failure.getIndex(), equalTo("index"));
    assertThat(failure.getType(), equalTo("type"));
    assertThat(failure.getId(), equalTo("id"));
    assertThat(failure.getCause(), equalTo(err));
    assertThat(failure.getStatus(), equalTo(RestStatus.CONFLICT));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
@Override
public void onFailure(Throwable e) {
    e = Exceptions.unwrap(e); // unwrap to get rid of RemoteTransportException
    if (e instanceof VersionConflictEngineException) {
        // treat version conflict as rows affected = 0
        result.set(TaskResult.ZERO);
    } else {
        result.setException(e);
    }
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
protected boolean isConflictException(Throwable e) {
    Throwable cause = ExceptionsHelper.unwrapCause(e);
    // on version conflict or document missing, it means
    // that a new change has crept into the replica, and it's fine
    if (cause instanceof VersionConflictEngineException) {
        return true;
    }
    if (cause instanceof DocumentAlreadyExistsException) {
        return true;
    }
    return false;
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
protected <T> Transform<Throwable, T> handleVersionConflict(
    Provider<T> emptyProvider, Runnable reportWriteDroppedByDuplicate
) {
    return throwable -> {
        if (throwable instanceof VersionConflictEngineException) {
            // Index request rejected, document already exists. That's ok, return success.
            reportWriteDroppedByDuplicate.run();
            return emptyProvider.get();
        }
        throw new RuntimeException(throwable);
    };
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
final protected void saveDocuments(String index, String documentType, List<? extends IDocument> documents) {
    if (documents.size() == 0) {
        return;
    }

    BulkRequestBuilder bulkRequestBuilder = this.getClient().prepareBulk();

    for (IDocument document : documents) {
        this.updateDocumentDatesTracking(document);
        ObjectWriter writer = getObjectWriter(document.getClass());
        IndexRequestBuilder request = this.getClient().prepareIndex()
                .setIndex(index)
                .setId(document.getId())
                .setType(documentType)
                .setOpType(IndexRequest.OpType.INDEX)
                .setSource(this.getSourceFromDocument(writer, document));

        bulkRequestBuilder.add(request);
    }

    try {
        bulkRequestBuilder.setRefresh(true);
        BulkResponse response = bulkRequestBuilder.execute().actionGet();
        if (response.hasFailures()) {
            throw new SearchServiceException(String.format("Unexpected failure bulk saving documents %s", response.buildFailureMessage()));
        }
    } catch (VersionConflictEngineException vc) {
        throw new ConcurrentModificationException(vc);
    } catch (ElasticsearchException ex) {
        throw new SearchServiceException("Unexpected failure while indexing a document.", ex);
    }
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
/**
 * Bulk update a set of documents using partial documents.
 *
 * @param index            document index
 * @param documentType     Type of the document
 * @param filterProvider   Serialization filterProvider to use in document serialization to only output fields we wish to update.
 * @param documentIds      documents to update.
 * @param documentTemplate the document template to apply changes from
 */
final protected void updateDocuments(String index, String documentType, FilterProvider filterProvider, Set<String> documentIds, IDocument documentTemplate) {
    if (documentIds.size() == 0) {
        return;
    }

    BulkRequestBuilder bulkRequestBuilder = this.getClient().prepareBulk();
    ObjectWriter writer = getObjectWriter(documentTemplate.getClass()).with(filterProvider);
    this.updateDocumentDatesTracking(documentTemplate);

    for (String documentId : documentIds) {
        UpdateRequestBuilder request = this.getClient().prepareUpdate()
                .setIndex(index)
                .setId(documentId)
                .setType(documentType)
                .setDoc(this.getSourceFromDocument(writer, documentTemplate));

        bulkRequestBuilder.add(request);
    }

    try {
        bulkRequestBuilder.setRefresh(true);
        BulkResponse response = bulkRequestBuilder.execute().actionGet();
        if (response.hasFailures()) {
            throw new SearchServiceException(String.format("Unexpected failure bulk updating documents %s", response.buildFailureMessage()));
        }
    } catch (VersionConflictEngineException vc) {
        throw new ConcurrentModificationException(vc);
    } catch (ElasticsearchException ex) {
        throw new SearchServiceException("Unexpected failure while indexing a document.", ex);
    }
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
/**
 * Save a single document to Elasticsearch, performing an "upsert".
 */
final protected <T extends IDocument> T createDocument(String index, String documentType, T document) {
    if (StringUtils.isEmpty(document.getId())) {
        document.setId(this.generateId());
    }
    this.updateDocumentDatesTracking(document);

    IndexRequestBuilder request = this.getClient().prepareIndex()
            .setIndex(index)
            .setType(documentType)
            .setId(document.getId())
            .setOpType(IndexRequest.OpType.CREATE)
            .setSource(this.getSourceFromDocument(document))
            .setRefresh(true);

    if (document.getVersion().isPresent()) {
        request.setVersion(document.getVersion().get());
    }

    try {
        IndexResponse response = request.execute().actionGet();
        document.setVersion(response.getVersion());
    } catch (VersionConflictEngineException vc) {
        throw new ConcurrentModificationException(vc);
    } catch (ElasticsearchException ex) {
        throw new SearchServiceException("Unexpected failure while indexing a document.", ex);
    }

    return document;
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
/**
 * Save a single document to Elasticsearch, performing an "upsert".
 */
final protected <T extends IDocument> T saveDocument(String index, String documentType, T document) {
    if (StringUtils.isEmpty(document.getId())) {
        document.setId(this.generateId());
    }
    this.updateDocumentDatesTracking(document);

    IndexRequestBuilder request = this.getClient().prepareIndex()
            .setIndex(index)
            .setType(documentType)
            .setId(document.getId())
            .setOpType(IndexRequest.OpType.INDEX)
            .setSource(this.getSourceFromDocument(document))
            .setRefresh(true);

    if (document.getVersion().isPresent()) {
        request.setVersion(document.getVersion().get());
    }

    try {
        IndexResponse response = request.execute().actionGet();
        document.setVersion(response.getVersion());
    } catch (VersionConflictEngineException vc) {
        throw new ConcurrentModificationException(vc);
    } catch (ElasticsearchException ex) {
        throw new SearchServiceException("Unexpected failure while indexing a document.", ex);
    }

    return document;
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
boolean ignoreReplicaException(Throwable e) {
    if (TransportActions.isShardNotAvailableException(e)) {
        return true;
    }
    Throwable cause = ExceptionsHelper.unwrapCause(e);
    return cause instanceof VersionConflictEngineException;
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
private static boolean isConflictException(final Exception e) {
    return ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException;
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testExternalGTE() throws Exception {
    createIndex("test");

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(12).setVersionType(VersionType.EXTERNAL_GTE).get();
    assertThat(indexResponse.getVersion(), equalTo(12L));

    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_2").setVersion(12).setVersionType(VersionType.EXTERNAL_GTE).get();
    assertThat(indexResponse.getVersion(), equalTo(12L));

    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_2").setVersion(14).setVersionType(VersionType.EXTERNAL_GTE).get();
    assertThat(indexResponse.getVersion(), equalTo(14L));

    assertThrows(client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(13).setVersionType(VersionType.EXTERNAL_GTE),
            VersionConflictEngineException.class);

    client().admin().indices().prepareRefresh().execute().actionGet();
    if (randomBoolean()) {
        refresh();
    }
    for (int i = 0; i < 10; i++) {
        assertThat(client().prepareGet("test", "type", "1").get().getVersion(), equalTo(14L));
    }

    // deleting with a lower version fails.
    assertThrows(
            client().prepareDelete("test", "type", "1").setVersion(2).setVersionType(VersionType.EXTERNAL_GTE),
            VersionConflictEngineException.class);

    // Delete with a higher or equal version deletes all versions up to the given one.
    long v = randomIntBetween(14, 17);
    DeleteResponse deleteResponse = client().prepareDelete("test", "type", "1").setVersion(v).setVersionType(VersionType.EXTERNAL_GTE).execute().actionGet();
    assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
    assertThat(deleteResponse.getVersion(), equalTo(v));

    // Deleting with a lower version keeps on failing after a delete.
    assertThrows(
            client().prepareDelete("test", "type", "1").setVersion(2).setVersionType(VersionType.EXTERNAL_GTE).execute(),
            VersionConflictEngineException.class);


    // But delete with a higher version is OK.
    deleteResponse = client().prepareDelete("test", "type", "1").setVersion(18).setVersionType(VersionType.EXTERNAL_GTE).execute().actionGet();
    assertEquals(DocWriteResponse.Result.NOT_FOUND, deleteResponse.getResult());
    assertThat(deleteResponse.getVersion(), equalTo(18L));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testExternalVersioning() throws Exception {
    createIndex("test");
    ensureGreen();

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(12).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(12L));

    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(14).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(14L));

    assertThrows(client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(13).setVersionType(VersionType.EXTERNAL).execute(),
            VersionConflictEngineException.class);

    if (randomBoolean()) {
        refresh();
    }
    for (int i = 0; i < 10; i++) {
        assertThat(client().prepareGet("test", "type", "1").execute().actionGet().getVersion(), equalTo(14L));
    }

    // deleting with a lower version fails.
    assertThrows(
            client().prepareDelete("test", "type", "1").setVersion(2).setVersionType(VersionType.EXTERNAL).execute(),
            VersionConflictEngineException.class);

    // Delete with a higher version deletes all versions up to the given one.
    DeleteResponse deleteResponse = client().prepareDelete("test", "type", "1").setVersion(17).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
    assertThat(deleteResponse.getVersion(), equalTo(17L));

    // Deleting with a lower version keeps on failing after a delete.
    assertThrows(
            client().prepareDelete("test", "type", "1").setVersion(2).setVersionType(VersionType.EXTERNAL).execute(),
            VersionConflictEngineException.class);


    // But delete with a higher version is OK.
    deleteResponse = client().prepareDelete("test", "type", "1").setVersion(18).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertEquals(DocWriteResponse.Result.NOT_FOUND, deleteResponse.getResult());
    assertThat(deleteResponse.getVersion(), equalTo(18L));


    // TODO: This behavior breaks rest api returning http status 201, good news is that it this is only the case until deletes GC kicks in.
    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(19).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(19L));


    deleteResponse = client().prepareDelete("test", "type", "1").setVersion(20).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
    assertThat(deleteResponse.getVersion(), equalTo(20L));

    // Make sure that the next delete will be GC. Note we do it on the index settings so it will be cleaned up
    HashMap<String, Object> newSettings = new HashMap<>();
    newSettings.put("index.gc_deletes", -1);
    client().admin().indices().prepareUpdateSettings("test").setSettings(newSettings).execute().actionGet();

    Thread.sleep(300); // gc works based on estimated sampled time. Give it a chance...

    // And now we have previous version return -1
    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(20).setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertThat(indexResponse.getVersion(), equalTo(20L));
}
 

import org.elasticsearch.index.engine.VersionConflictEngineException; //导入依赖的package包/类
public void testExecuteBulkIndexRequest() throws Exception {
    IndexMetaData metaData = indexMetaData();
    IndexShard shard = newStartedShard(true);

    BulkItemRequest[] items = new BulkItemRequest[1];
    boolean create = randomBoolean();
    DocWriteRequest writeRequest = new IndexRequest("index", "type", "id")
            .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar")
            .create(create);
    BulkItemRequest primaryRequest = new BulkItemRequest(0, writeRequest);
    items[0] = primaryRequest;
    BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);

    Translog.Location location = new Translog.Location(0, 0, 0);
    UpdateHelper updateHelper = null;

    Translog.Location newLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
            location, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());

    // Translog should change, since there were no problems
    assertThat(newLocation, not(location));

    BulkItemResponse primaryResponse = bulkShardRequest.items()[0].getPrimaryResponse();

    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(create ? DocWriteRequest.OpType.CREATE : DocWriteRequest.OpType.INDEX));
    assertFalse(primaryResponse.isFailed());

    // Assert that the document actually made it there
    assertDocCount(shard, 1);

    writeRequest = new IndexRequest("index", "type", "id")
            .source(Requests.INDEX_CONTENT_TYPE, "foo", "bar")
            .create(true);
    primaryRequest = new BulkItemRequest(0, writeRequest);
    items[0] = primaryRequest;
    bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);

    Translog.Location secondLocation = TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest,
            newLocation, 0, updateHelper, threadPool::absoluteTimeInMillis, new NoopMappingUpdatePerformer());

    // Translog should not change, since the document was not indexed due to a version conflict
    assertThat(secondLocation, equalTo(newLocation));

    BulkItemRequest replicaRequest = bulkShardRequest.items()[0];

    primaryResponse = bulkShardRequest.items()[0].getPrimaryResponse();

    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.CREATE));
    // Should be failed since the document already exists
    assertTrue(primaryResponse.isFailed());

    BulkItemResponse.Failure failure = primaryResponse.getFailure();
    assertThat(failure.getIndex(), equalTo("index"));
    assertThat(failure.getType(), equalTo("type"));
    assertThat(failure.getId(), equalTo("id"));
    assertThat(failure.getCause().getClass(), equalTo(VersionConflictEngineException.class));
    assertThat(failure.getCause().getMessage(),
            containsString("version conflict, document already exists (current version [1])"));
    assertThat(failure.getStatus(), equalTo(RestStatus.CONFLICT));

    assertThat(replicaRequest, equalTo(primaryRequest));

    // Assert that the document count is still 1
    assertDocCount(shard, 1);
    closeShards(shard);
}