import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
/**
 * Adds a list of requests to be executed. Either index or delete requests.
 */
public BulkRequest add(Iterable<DocWriteRequest> requests) {
    for (DocWriteRequest request : requests) {
        add(request);
    }
    return this;
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
BulkRequest getBulkRequest() {
    if (itemResponses.isEmpty()) {
        return bulkRequest;
    } else {
        BulkRequest modifiedBulkRequest = new BulkRequest();
        modifiedBulkRequest.setRefreshPolicy(bulkRequest.getRefreshPolicy());
        modifiedBulkRequest.waitForActiveShards(bulkRequest.waitForActiveShards());
        modifiedBulkRequest.timeout(bulkRequest.timeout());

        int slot = 0;
        List<DocWriteRequest> requests = bulkRequest.requests();
        originalSlots = new int[requests.size()]; // oversize, but that's ok
        for (int i = 0; i < requests.size(); i++) {
            DocWriteRequest request = requests.get(i);
            if (failedSlots.get(i) == false) {
                modifiedBulkRequest.add(request);
                originalSlots[slot++] = i;
            }
        }
        return modifiedBulkRequest;
    }
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的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.DocWriteRequest; //导入依赖的package包/类
public void testIndexRequestOpTypeFromString() throws Exception {
    String create = "create";
    String index = "index";
    String createUpper = "CREATE";
    String indexUpper = "INDEX";

    IndexRequest indexRequest = new IndexRequest("");
    indexRequest.opType(create);
    assertThat(indexRequest.opType() , equalTo(DocWriteRequest.OpType.CREATE));
    indexRequest.opType(createUpper);
    assertThat(indexRequest.opType() , equalTo(DocWriteRequest.OpType.CREATE));
    indexRequest.opType(index);
    assertThat(indexRequest.opType() , equalTo(DocWriteRequest.OpType.INDEX));
    indexRequest.opType(indexUpper);
    assertThat(indexRequest.opType() , equalTo(DocWriteRequest.OpType.INDEX));
}
 

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

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

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

    // Pretend the mappings haven't made it to the node yet, and throw  a rejection
    Exception err = new ReplicationOperation.RetryOnPrimaryException(shardId, "rejection");

    try {
        TransportShardBulkAction.executeBulkItemRequest(metaData, shard, bulkShardRequest, location,
                0, updateHelper, threadPool::absoluteTimeInMillis, new ThrowingMappingUpdatePerformer(err));
        fail("should have thrown a retry exception");
    } catch (ReplicationOperation.RetryOnPrimaryException e) {
        assertThat(e, equalTo(err));
    }

    closeShards(shard);
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的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.DocWriteRequest; //导入依赖的package包/类
public void processDocumentActions(Stream<DocumentAction> documentActions)
{
	LOG.trace("Processing document actions ...");
	BulkProcessor bulkProcessor = bulkProcessorFactory.create(client);
	try
	{
		documentActions.forEachOrdered(documentAction ->
		{
			DocWriteRequest docWriteRequest = toDocWriteRequest(documentAction);
			bulkProcessor.add(docWriteRequest);
		});
	}
	finally
	{
		waitForCompletion(bulkProcessor);
		LOG.debug("Processed document actions.");
	}
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
private static void setRandomVersion(DocWriteRequest<?> request, Map<String, String> expectedParams) {
    if (randomBoolean()) {
        long version = randomFrom(Versions.MATCH_ANY, Versions.MATCH_DELETED, Versions.NOT_FOUND, randomNonNegativeLong());
        request.version(version);
        if (version != Versions.MATCH_ANY) {
            expectedParams.put("version", Long.toString(version));
        }
    }
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
private static void setRandomVersionType(DocWriteRequest<?> request, Map<String, String> expectedParams) {
    if (randomBoolean()) {
        VersionType versionType = randomFrom(VersionType.values());
        request.versionType(versionType);
        if (versionType != VersionType.INTERNAL) {
            expectedParams.put("version_type", versionType.name().toLowerCase(Locale.ROOT));
        }
    }
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
/**
 * @return Whether this bulk request contains index request with an ingest pipeline enabled.
 */
public boolean hasIndexRequestsWithPipelines() {
    for (DocWriteRequest actionRequest : requests) {
        if (actionRequest instanceof IndexRequest) {
            IndexRequest indexRequest = (IndexRequest) actionRequest;
            if (Strings.hasText(indexRequest.getPipeline())) {
                return true;
            }
        }
    }

    return false;
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    waitForActiveShards = ActiveShardCount.readFrom(in);
    int size = in.readVInt();
    for (int i = 0; i < size; i++) {
        requests.add(DocWriteRequest.readDocumentRequest(in));
    }
    refreshPolicy = RefreshPolicy.readFrom(in);
    timeout = new TimeValue(in);
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
@Override
public void writeTo(StreamOutput out) throws IOException {
    super.writeTo(out);
    waitForActiveShards.writeTo(out);
    out.writeVInt(requests.size());
    for (DocWriteRequest request : requests) {
        DocWriteRequest.writeDocumentRequest(out, request);
    }
    refreshPolicy.writeTo(out);
    timeout.writeTo(out);
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
private boolean setResponseFailureIfIndexMatches(AtomicArray<BulkItemResponse> responses, int idx, DocWriteRequest request, String index, Exception e) {
    if (index.equals(request.index())) {
        responses.set(idx, new BulkItemResponse(idx, request.opType(), new BulkItemResponse.Failure(request.index(), request.type(), request.id(), e)));
        return true;
    }
    return false;
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
Index resolveIfAbsent(DocWriteRequest request) {
    Index concreteIndex = indices.get(request.index());
    if (concreteIndex == null) {
        concreteIndex = indexNameExpressionResolver.concreteSingleIndex(state, request);
        indices.put(request.index(), concreteIndex);
    }
    return concreteIndex;
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
@Override
protected WriteReplicaResult<Request> shardOperationOnReplica(
    Request replicaRequest, IndexShard replica) throws Exception {
    BulkItemRequest[] itemRequests = new BulkItemRequest[1];
    WriteRequest.RefreshPolicy refreshPolicy = replicaRequest.getRefreshPolicy();
    itemRequests[0] = new BulkItemRequest(0, ((DocWriteRequest) replicaRequest));
    BulkShardRequest bulkShardRequest = new BulkShardRequest(replicaRequest.shardId(), refreshPolicy, itemRequests);
    WriteReplicaResult<BulkShardRequest> result = shardBulkAction.shardOperationOnReplica(bulkShardRequest, replica);
    // a replica operation can never throw a document-level failure,
    // as the same document has been already indexed successfully in the primary
    return new WriteReplicaResult<>(replicaRequest, result.location, null, replica, logger);
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
public static BulkRequest toSingleItemBulkRequest(ReplicatedWriteRequest request) {
    BulkRequest bulkRequest = new BulkRequest();
    bulkRequest.add(((DocWriteRequest) request));
    bulkRequest.setRefreshPolicy(request.getRefreshPolicy());
    bulkRequest.timeout(request.timeout());
    bulkRequest.waitForActiveShards(request.waitForActiveShards());
    request.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE);
    return bulkRequest;
}
 

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

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

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

    // Return a mapping conflict (IAE) when trying to update the mapping
    Exception err = new IllegalArgumentException("mapping conflict");

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

    // Translog shouldn't change, as there were conflicting mappings
    assertThat(newLocation, equalTo(location));

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

    // Since this was not a conflict failure, the primary response
    // should be filled out with the failure information
    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.INDEX));
    assertTrue(primaryResponse.isFailed());
    assertThat(primaryResponse.getFailureMessage(), containsString("mapping conflict"));
    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.BAD_REQUEST));

    closeShards(shard);
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的package包/类
public void testUpdateReplicaRequestWithFailure() 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 ElasticsearchException("I'm dead <(x.x)>");
    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 dead <(x.x)>"));
    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.INTERNAL_SERVER_ERROR));
}
 

import org.elasticsearch.action.DocWriteRequest; //导入依赖的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.action.DocWriteRequest; //导入依赖的package包/类
public void testUpdateReplicaRequestWithSuccess() throws Exception {
    DocWriteRequest writeRequest = new IndexRequest("index", "type", "id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
    BulkItemRequest replicaRequest = new BulkItemRequest(0, writeRequest);

    boolean created = randomBoolean();
    Translog.Location resultLocation = new Translog.Location(42, 42, 42);
    Engine.IndexResult indexResult = new FakeResult(1, 1, created, resultLocation);
    DocWriteResponse indexResponse = new IndexResponse(shardId, "index", "id", 1, 1, created);
    BulkItemResultHolder goodResults = new BulkItemResultHolder(indexResponse, indexResult, replicaRequest);

    Translog.Location originalLocation = new Translog.Location(21, 21, 21);
    BulkItemRequest[] items = new BulkItemRequest[0];
    BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
    Translog.Location newLocation = TransportShardBulkAction.updateReplicaRequest(goodResults,
            DocWriteRequest.OpType.INDEX, originalLocation, bulkShardRequest);

    BulkItemResponse primaryResponse = replicaRequest.getPrimaryResponse();

    // Check that the translog is successfully advanced
    assertThat(newLocation, equalTo(resultLocation));
    // Since this was not a conflict failure, the primary response
    // should be filled out with the failure information
    assertThat(primaryResponse.getItemId(), equalTo(0));
    assertThat(primaryResponse.getId(), equalTo("id"));
    assertThat(primaryResponse.getOpType(), equalTo(DocWriteRequest.OpType.INDEX));
    DocWriteResponse response = primaryResponse.getResponse();
    assertThat(response.status(), equalTo(created ? RestStatus.CREATED : RestStatus.OK));
}