import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
/**
 * Execute the given {@link IndexRequest} on a replica shard, throwing a
 * {@link RetryOnReplicaException} if the operation needs to be re-tried.
 */
public static Engine.IndexResult executeIndexRequestOnReplica(DocWriteResponse primaryResponse, IndexRequest request, IndexShard replica) throws IOException {
    final ShardId shardId = replica.shardId();
    SourceToParse sourceToParse =
        SourceToParse.source(SourceToParse.Origin.REPLICA, shardId.getIndexName(), request.type(), request.id(), request.source(),
            request.getContentType()).routing(request.routing()).parent(request.parent());

    final Engine.Index operation;
    final long version = primaryResponse.getVersion();
    final VersionType versionType = request.versionType().versionTypeForReplicationAndRecovery();
    assert versionType.validateVersionForWrites(version);
    final long seqNo = primaryResponse.getSeqNo();
    try {
        operation = replica.prepareIndexOnReplica(sourceToParse, seqNo, version, versionType, request.getAutoGeneratedTimestamp(), request.isRetry());
    } catch (MapperParsingException e) {
        return new Engine.IndexResult(e, version, seqNo);
    }
    Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
    if (update != null) {
        throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
    }
    return replica.index(operation);
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
@Override
public void writeTo(StreamOutput out) throws IOException {
    out.writeVInt(id);
    if (out.getVersion().before(Version.V_6_0_0_alpha1_UNRELEASED)) { // TODO remove once backported
        // old nodes expect updated version and version type on the request
        if (primaryResponse != null) {
            request.version(primaryResponse.getVersion());
            request.versionType(request.versionType().versionTypeForReplicationAndRecovery());
            DocWriteRequest.writeDocumentRequest(out, request);
        } else {
            DocWriteRequest.writeDocumentRequest(out, request);
        }
    } else {
        DocWriteRequest.writeDocumentRequest(out, request);
    }
    out.writeOptionalStreamable(primaryResponse);
    if (out.getVersion().before(Version.V_6_0_0_alpha1_UNRELEASED)) { // TODO remove once backported
        if (primaryResponse != null) {
            out.writeBoolean(primaryResponse.isFailed()
                    || primaryResponse.getResponse().getResult() == DocWriteResponse.Result.NOOP);
        } else {
            out.writeBoolean(false);
        }
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的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.action.DocWriteResponse; //导入依赖的package包/类
public void testInheritMaxValidAutoIDTimestampOnRecovery() throws Exception {
    try (ReplicationGroup shards = createGroup(0)) {
        shards.startAll();
        final IndexRequest indexRequest = new IndexRequest(index.getName(), "type").source("{}", XContentType.JSON);
        indexRequest.onRetry(); // force an update of the timestamp
        final IndexResponse response = shards.index(indexRequest);
        assertEquals(DocWriteResponse.Result.CREATED, response.getResult());
        if (randomBoolean()) { // lets check if that also happens if no translog record is replicated
            shards.flush();
        }
        IndexShard replica = shards.addReplica();
        shards.recoverReplica(replica);

        SegmentsStats segmentsStats = replica.segmentStats(false);
        SegmentsStats primarySegmentStats = shards.getPrimary().segmentStats(false);
        assertNotEquals(IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP, primarySegmentStats.getMaxUnsafeAutoIdTimestamp());
        assertEquals(primarySegmentStats.getMaxUnsafeAutoIdTimestamp(), segmentsStats.getMaxUnsafeAutoIdTimestamp());
        assertNotEquals(Long.MAX_VALUE, segmentsStats.getMaxUnsafeAutoIdTimestamp());
    }
}
 

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

    DocWriteResponse noopUpdateResponse = new UpdateResponse(shardId, "index", "id", 0, DocWriteResponse.Result.NOOP);
    BulkItemResultHolder noopResults = new BulkItemResultHolder(noopUpdateResponse, null, 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(noopResults,
            DocWriteRequest.OpType.UPDATE, location, bulkShardRequest);

    BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();

    // Basically nothing changes in the request since it's a noop
    assertThat(newLocation, equalTo(location));
    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.UPDATE));
    assertThat(primaryResponse.getResponse(), equalTo(noopUpdateResponse));
    assertThat(primaryResponse.getResponse().getResult(), equalTo(DocWriteResponse.Result.NOOP));
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
/**
 * Index a Fact into ElasticSearch.
 *
 * @param fact Fact to index
 * @return Indexed Fact
 */
public FactDocument indexFact(FactDocument fact) {
  if (fact == null || fact.getId() == null) return null;
  IndexResponse response;

  try {
    IndexRequest request = new IndexRequest(INDEX_NAME, TYPE_NAME, fact.getId().toString())
            .source(FACT_DOCUMENT_WRITER.writeValueAsBytes(encodeValues(fact)), XContentType.JSON);
    response = clientFactory.getHighLevelClient().index(request);
  } catch (IOException ex) {
    throw logAndExit(ex, String.format("Could not perform request to index Fact with id = %s.", fact.getId()));
  }

  if (response.status() != RestStatus.OK && response.status() != RestStatus.CREATED) {
    LOGGER.warning("Could not index Fact with id = %s.", fact.getId());
  } else if (response.getResult() == DocWriteResponse.Result.CREATED) {
    LOGGER.info("Successfully indexed Fact with id = %s.", fact.getId());
  } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
    LOGGER.info("Successfully re-indexed existing Fact with id = %s.", fact.getId());
  }

  return fact;
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的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.action.DocWriteResponse; //导入依赖的package包/类
public FeatureStoreResponse addElement(StorableElement element,
                                       @Nullable FeatureValidation validation,
                                       String store) throws ExecutionException, InterruptedException {
    FeatureStoreRequestBuilder builder = FeatureStoreAction.INSTANCE.newRequestBuilder(client());
    builder.request().setStorableElement(element);
    builder.request().setAction(FeatureStoreAction.FeatureStoreRequest.Action.CREATE);
    builder.request().setStore(store);
    builder.request().setValidation(validation);
    FeatureStoreResponse response = builder.execute().get();
    assertEquals(1, response.getResponse().getVersion());
    assertEquals(IndexFeatureStore.ES_TYPE, response.getResponse().getType());
    assertEquals(DocWriteResponse.Result.CREATED, response.getResponse().getResult());
    assertEquals(element.id(), response.getResponse().getId());
    assertEquals(store, response.getResponse().getIndex());
    return response;
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
public void testFailuresOnDuplicates() throws Exception {
    addElement(randomFeature("duplicated"));

    AddFeaturesToSetRequestBuilder builder = AddFeaturesToSetAction.INSTANCE.newRequestBuilder(client());
    builder.request().setFeatureSet("duplicated_set");
    builder.request().setFeatureNameQuery("duplicated*");
    builder.request().setStore(IndexFeatureStore.DEFAULT_STORE);
    AddFeaturesToSetResponse resp = builder.execute().get();

    assertEquals(DocWriteResponse.Result.CREATED, resp.getResponse().getResult());
    assertEquals(1, resp.getResponse().getVersion());


    AddFeaturesToSetRequestBuilder builder2 = AddFeaturesToSetAction.INSTANCE.newRequestBuilder(client());
    builder2.request().setFeatureSet("duplicated_set");
    builder2.request().setFeatureNameQuery("duplicated");
    builder2.request().setStore(IndexFeatureStore.DEFAULT_STORE);

    Throwable iae = unwrap(expectThrows(ExecutionException.class, () -> builder2.execute().get()),
            IllegalArgumentException.class);
    assertNotNull(iae);
    assertThat(iae.getMessage(), containsString("defined twice in this set"));
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
@Override
public boolean delete(DeleteRequest request) {
    StopWatch watch = new StopWatch();
    String index = request.index == null ? this.index : request.index;
    boolean deleted = false;
    try {
        DeleteResponse response = client().prepareDelete(index, type, request.id).get();
        deleted = response.getResult() == DocWriteResponse.Result.DELETED;
        return deleted;
    } catch (ElasticsearchException e) {
        throw new SearchException(e);   // due to elastic search uses async executor to run, we have to wrap the exception to retain the original place caused the exception
    } finally {
        long elapsedTime = watch.elapsedTime();
        ActionLogContext.track("elasticsearch", elapsedTime, 0, deleted ? 1 : 0);
        logger.debug("delete, index={}, type={}, id={}, elapsedTime={}", index, type, request.id, elapsedTime);
        checkSlowOperation(elapsedTime);
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
/**
 * Parse the current token and update the parsing context appropriately.
 */
public static void parseXContentFields(XContentParser parser, Builder context) throws IOException {
    XContentParser.Token token = parser.currentToken();
    String currentFieldName = parser.currentName();

    if (FOUND.equals(currentFieldName)) {
        if (token.isValue()) {
            context.setFound(parser.booleanValue());
        }
    } else {
        DocWriteResponse.parseInnerToXContent(parser, context);
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
/**
 * Parse the current token and update the parsing context appropriately.
 */
public static void parseXContentFields(XContentParser parser, Builder context) throws IOException {
    XContentParser.Token token = parser.currentToken();
    String currentFieldName = parser.currentName();

    if (GET.equals(currentFieldName)) {
        if (token == XContentParser.Token.START_OBJECT) {
            context.setGetResult(GetResult.fromXContentEmbedded(parser));
        }
    } else {
        DocWriteResponse.parseInnerToXContent(parser, context);
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
/**
 * Parse the current token and update the parsing context appropriately.
 */
public static void parseXContentFields(XContentParser parser, Builder context) throws IOException {
    XContentParser.Token token = parser.currentToken();
    String currentFieldName = parser.currentName();

    if (CREATED.equals(currentFieldName)) {
        if (token.isValue()) {
            context.setCreated(parser.booleanValue());
        }
    } else {
        DocWriteResponse.parseInnerToXContent(parser, context);
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
BulkItemResultHolder(@Nullable DocWriteResponse response,
                     @Nullable Engine.Result operationResult,
                     BulkItemRequest replicaRequest) {
    this.response = response;
    this.operationResult = operationResult;
    this.replicaRequest = replicaRequest;
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
@Override
public void setForcedRefresh(boolean forcedRefresh) {
    /*
     * Each DocWriteResponse already has a location for whether or not it forced a refresh so we just set that information on the
     * response.
     */
    for (BulkItemResponse response : responses) {
        DocWriteResponse r = response.getResponse();
        if (r != null) {
            r.setForcedRefresh(forcedRefresh);
        }
    }
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
private static Engine.DeleteResult executeDeleteRequestOnReplica(DocWriteResponse primaryResponse, DeleteRequest request, IndexShard replica) throws IOException {
    final VersionType versionType = request.versionType().versionTypeForReplicationAndRecovery();
    final long version = primaryResponse.getVersion();
    assert versionType.validateVersionForWrites(version);
    final Engine.Delete delete = replica.prepareDeleteOnReplica(request.type(), request.id(),
            primaryResponse.getSeqNo(), request.primaryTerm(), version, versionType);
    return replica.delete(delete);
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
public void testCreatedFlagWithFlush() throws Exception {
    createIndex("test");
    ensureGreen();

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").execute().actionGet();
    assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());

    client().prepareDelete("test", "type", "1").execute().actionGet();

    flush();

    indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_2").execute().actionGet();
    assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
public void testCreatedFlagParallelExecution() throws Exception {
    createIndex("test");
    ensureGreen();

    int threadCount = 20;
    final int docCount = 300;
    int taskCount = docCount * threadCount;

    final AtomicIntegerArray createdCounts = new AtomicIntegerArray(docCount);
    ExecutorService threadPool = Executors.newFixedThreadPool(threadCount);
    List<Callable<Void>> tasks = new ArrayList<>(taskCount);
    final Random random = random();
    for (int i=0;i< taskCount; i++ ) {
        tasks.add(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                int docId = random.nextInt(docCount);
                IndexResponse indexResponse = index("test", "type", Integer.toString(docId), "field1", "value");
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    createdCounts.incrementAndGet(docId);
                }
                return null;
            }
        });
    }

    threadPool.invokeAll(tasks);

    for (int i=0;i<docCount;i++) {
        assertThat(createdCounts.get(i), lessThanOrEqualTo(1));
    }
    terminate(threadPool);
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
public void testCreatedFlagWithExternalVersioning() throws Exception {
    createIndex("test");
    ensureGreen();

    IndexResponse indexResponse = client().prepareIndex("test", "type", "1").setSource("field1", "value1_1").setVersion(123)
                                          .setVersionType(VersionType.EXTERNAL).execute().actionGet();
    assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
}
 

import org.elasticsearch.action.DocWriteResponse; //导入依赖的package包/类
public void testCreateFlagWithBulk() {
    createIndex("test");
    ensureGreen();

    BulkResponse bulkResponse = client().prepareBulk().add(client().prepareIndex("test", "type", "1").setSource("field1", "value1_1")).execute().actionGet();
    assertThat(bulkResponse.hasFailures(), equalTo(false));
    assertThat(bulkResponse.getItems().length, equalTo(1));
    IndexResponse indexResponse = bulkResponse.getItems()[0].getResponse();
    assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
}