import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
/**
 * Create the XContent header for any {@link BaseNodesResponse}. This looks like:
 * <code>
 * "_nodes" : {
 *   "total" : 3,
 *   "successful" : 1,
 *   "failed" : 2,
 *   "failures" : [ { ... }, { ... } ]
 * }
 * </code>
 * Prefer the overload that properly invokes this method to calling this directly.
 *
 * @param builder XContent builder.
 * @param params XContent parameters.
 * @param total The total number of nodes touched.
 * @param successful The successful number of responses received.
 * @param failed The number of failures (effectively {@code total - successful}).
 * @param failures The failure exceptions related to {@code failed}.
 * @see #buildNodesHeader(XContentBuilder, Params, BaseNodesResponse)
 */
public static void buildNodesHeader(final XContentBuilder builder, final Params params,
                                    final int total, final int successful, final int failed,
                                    final List<FailedNodeException> failures) throws IOException {
    builder.startObject("_nodes");
    builder.field("total", total);
    builder.field("successful", successful);
    builder.field("failed", failed);

    if (failures.isEmpty() == false) {
        builder.startArray("failures");
        for (FailedNodeException failure : failures) {
            builder.startObject();
            failure.toXContent(builder, params);
            builder.endObject();
        }
        builder.endArray();
    }

    builder.endObject();
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
public ClusterStatsResponse(long timestamp,
                            ClusterName clusterName,
                            List<ClusterStatsNodeResponse> nodes,
                            List<FailedNodeException> failures) {
    super(clusterName, nodes, failures);
    this.timestamp = timestamp;
    nodesStats = new ClusterStatsNodes(nodes);
    indicesStats = new ClusterStatsIndices(nodes);
    for (ClusterStatsNodeResponse response : nodes) {
        // only the master node populates the status
        if (response.clusterStatus() != null) {
            status = response.clusterStatus();
            break;
        }
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
protected void onNodeFailure(DiscoveryNode node, int nodeIndex, Throwable t) {
    String nodeId = node.getId();
    if (logger.isDebugEnabled() && !(t instanceof NodeShouldNotConnectException)) {
        logger.debug(
            (org.apache.logging.log4j.util.Supplier<?>)
                () -> new ParameterizedMessage("failed to execute [{}] on node [{}]", actionName, nodeId), t);
    }

    // this is defensive to protect against the possibility of double invocation
    // the current implementation of TransportService#sendRequest guards against this
    // but concurrency is hard, safety is important, and the small performance loss here does not matter
    if (responses.compareAndSet(nodeIndex, null, new FailedNodeException(nodeId, "Failed node [" + nodeId + "]", t))) {
        if (counter.incrementAndGet() == responses.length()) {
            onCompletion();
        }
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    int size = in.readVInt();
    List<TaskOperationFailure> taskFailures = new ArrayList<>();
    for (int i = 0; i < size; i++) {
        taskFailures.add(new TaskOperationFailure(in));
    }
    size = in.readVInt();
    this.taskFailures = Collections.unmodifiableList(taskFailures);
    List<FailedNodeException> nodeFailures = new ArrayList<>();
    for (int i = 0; i < size; i++) {
        nodeFailures.add(new FailedNodeException(in));
    }
    this.nodeFailures = Collections.unmodifiableList(nodeFailures);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected TasksResponse newResponse(TasksRequest request, AtomicReferenceArray responses) {
    List<TaskResponse> tasks = new ArrayList<>();
    List<FailedNodeException> failedNodeExceptions = new ArrayList<>();
    List<TaskOperationFailure> taskOperationFailures = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        Object response = responses.get(i);
        if (response instanceof FailedNodeException) {
            failedNodeExceptions.add((FailedNodeException) response);
        } else {
            NodeTasksResponse tasksResponse = (NodeTasksResponse) response;
            if (tasksResponse.results != null) {
                tasks.addAll(tasksResponse.results);
            }
            if (tasksResponse.exceptions != null) {
                taskOperationFailures.addAll(tasksResponse.exceptions);
            }
        }
    }
    return newResponse(request, tasks, taskOperationFailures, failedNodeExceptions);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
/**
 * Map the responses into {@code nodeResponseClass} responses and {@link FailedNodeException}s.
 *
 * @param request The associated request.
 * @param nodesResponses All node-level responses
 * @return Never {@code null}.
 * @throws NullPointerException if {@code nodesResponses} is {@code null}
 * @see #newResponse(BaseNodesRequest, List, List)
 */
protected NodesResponse newResponse(NodesRequest request, AtomicReferenceArray nodesResponses) {
    final List<NodeResponse> responses = new ArrayList<>();
    final List<FailedNodeException> failures = new ArrayList<>();

    final boolean accumulateExceptions = accumulateExceptions();
    for (int i = 0; i < nodesResponses.length(); ++i) {
        Object response = nodesResponses.get(i);

        if (response instanceof FailedNodeException) {
            if (accumulateExceptions) {
                failures.add((FailedNodeException)response);
            } else {
                logger.warn("not accumulating exceptions, excluding exception from response", (FailedNodeException)response);
            }
        } else {
            responses.add(nodeResponseClass.cast(response));
        }
    }

    return newResponse(request, responses, failures);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected NodesStoreFilesMetaData newResponse(Request request, AtomicReferenceArray responses) {
    final List<NodeStoreFilesMetaData> nodeStoreFilesMetaDatas = new ArrayList<>();
    final List<FailedNodeException> failures = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        Object resp = responses.get(i);
        if (resp instanceof NodeStoreFilesMetaData) { // will also filter out null response for unallocated ones
            nodeStoreFilesMetaDatas.add((NodeStoreFilesMetaData) resp);
        } else if (resp instanceof FailedNodeException) {
            failures.add((FailedNodeException) resp);
        } else {
            logger.warn("unknown response type [{}], expected NodeStoreFilesMetaData or FailedNodeException", resp);
        }
    }
    return new NodesStoreFilesMetaData(clusterName, nodeStoreFilesMetaDatas.toArray(new NodeStoreFilesMetaData[nodeStoreFilesMetaDatas.size()]),
            failures.toArray(new FailedNodeException[failures.size()]));
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected NodesGatewayMetaState newResponse(Request request, AtomicReferenceArray responses) {
    final List<NodeGatewayMetaState> nodesList = new ArrayList<>();
    final List<FailedNodeException> failures = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        Object resp = responses.get(i);
        if (resp instanceof NodeGatewayMetaState) { // will also filter out null response for unallocated ones
            nodesList.add((NodeGatewayMetaState) resp);
        } else if (resp instanceof FailedNodeException) {
            failures.add((FailedNodeException) resp);
        } else {
            logger.warn("unknown response type [{}], expected NodeLocalGatewayMetaState or FailedNodeException", resp);
        }
    }
    return new NodesGatewayMetaState(clusterName, nodesList.toArray(new NodeGatewayMetaState[nodesList.size()]),
            failures.toArray(new FailedNodeException[failures.size()]));
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected NodesGatewayStartedShards newResponse(Request request, AtomicReferenceArray responses) {
    final List<NodeGatewayStartedShards> nodesList = new ArrayList<>();
    final List<FailedNodeException> failures = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        Object resp = responses.get(i);
        if (resp instanceof NodeGatewayStartedShards) { // will also filter out null response for unallocated ones
            nodesList.add((NodeGatewayStartedShards) resp);
        } else if (resp instanceof FailedNodeException) {
            failures.add((FailedNodeException) resp);
        } else {
            logger.warn("unknown response type [{}], expected NodeLocalGatewayStartedShards or FailedNodeException", resp);
        }
    }
    return new NodesGatewayStartedShards(clusterName, nodesList.toArray(new NodeGatewayStartedShards[nodesList.size()]),
            failures.toArray(new FailedNodeException[failures.size()]));
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected NodesSnapshotStatus newResponse(Request request, AtomicReferenceArray responses) {
    final List<NodeSnapshotStatus> nodesList = new ArrayList<>();
    final List<FailedNodeException> failures = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        Object resp = responses.get(i);
        if (resp instanceof NodeSnapshotStatus) { // will also filter out null response for unallocated ones
            nodesList.add((NodeSnapshotStatus) resp);
        } else if (resp instanceof FailedNodeException) {
            failures.add((FailedNodeException) resp);
        } else {
            logger.warn("unknown response type [{}], expected NodeSnapshotStatus or FailedNodeException", resp);
        }
    }
    return new NodesSnapshotStatus(clusterName, nodesList.toArray(new NodeSnapshotStatus[nodesList.size()]),
            failures.toArray(new FailedNodeException[failures.size()]));
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params)
        throws IOException {
    builder.startObject();
    builder.startArray("imports");
    for (NodeImportResponse r : this.responses) {
        r.toXContent(builder, params);
    }
    builder.endArray();
    if (nodeFailures != null && nodeFailures.size() > 0) {
        builder.startArray("failures");
        for (FailedNodeException failure : nodeFailures) {
            builder.startObject();
            builder.field("node_id", failure.nodeId());
            builder.field("reason", failure.getDetailedMessage());
            builder.endObject();
        }
        builder.endArray();
    }
    builder.endObject();
    return builder;
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    int responsesCount = in.readInt();
    this.responses = new ArrayList<NodeImportResponse>(responsesCount);
    for (int i = 0; i < responsesCount; i++) {
        responses.add(NodeImportResponse.readNew(in));
    }
    int failuresCount = in.readInt();
    this.nodeFailures = new ArrayList<FailedNodeException>(failuresCount);
    for (int i = 0; i < failuresCount; i++) {
        String nodeId = in.readString();
        String msg = in.readOptionalString();
        FailedNodeException e = new FailedNodeException(nodeId, msg, null);
        nodeFailures.add(e);
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected synchronized void processAsyncFetch(ShardId shardId, List<NodeGatewayStartedShards> responses, List<FailedNodeException> failures) {
    fetchResponses.add(new Response(shardId, responses, failures));
    if (expectedOps.countDown()) {
        finish();
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
void finish() {
    ImmutableOpenMap.Builder<String, ImmutableOpenIntMap<java.util.List<IndicesShardStoresResponse.StoreStatus>>> indicesStoreStatusesBuilder = ImmutableOpenMap.builder();
    java.util.List<IndicesShardStoresResponse.Failure> failureBuilder = new ArrayList<>();
    for (Response fetchResponse : fetchResponses) {
        ImmutableOpenIntMap<java.util.List<IndicesShardStoresResponse.StoreStatus>> indexStoreStatuses = indicesStoreStatusesBuilder.get(fetchResponse.shardId.getIndexName());
        final ImmutableOpenIntMap.Builder<java.util.List<IndicesShardStoresResponse.StoreStatus>> indexShardsBuilder;
        if (indexStoreStatuses == null) {
            indexShardsBuilder = ImmutableOpenIntMap.builder();
        } else {
            indexShardsBuilder = ImmutableOpenIntMap.builder(indexStoreStatuses);
        }
        java.util.List<IndicesShardStoresResponse.StoreStatus> storeStatuses = indexShardsBuilder.get(fetchResponse.shardId.id());
        if (storeStatuses == null) {
            storeStatuses = new ArrayList<>();
        }
        for (NodeGatewayStartedShards response : fetchResponse.responses) {
            if (shardExistsInNode(response)) {
                IndicesShardStoresResponse.StoreStatus.AllocationStatus allocationStatus = getAllocationStatus(fetchResponse.shardId.getIndexName(), fetchResponse.shardId.id(), response.getNode());
                storeStatuses.add(new IndicesShardStoresResponse.StoreStatus(response.getNode(), response.allocationId(), allocationStatus, response.storeException()));
            }
        }
        CollectionUtil.timSort(storeStatuses);
        indexShardsBuilder.put(fetchResponse.shardId.id(), storeStatuses);
        indicesStoreStatusesBuilder.put(fetchResponse.shardId.getIndexName(), indexShardsBuilder.build());
        for (FailedNodeException failure : fetchResponse.failures) {
            failureBuilder.add(new IndicesShardStoresResponse.Failure(failure.nodeId(), fetchResponse.shardId.getIndexName(), fetchResponse.shardId.id(), failure.getCause()));
        }
    }
    listener.onResponse(new IndicesShardStoresResponse(indicesStoreStatusesBuilder.build(), Collections.unmodifiableList(failureBuilder)));
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected ClusterStatsResponse newResponse(ClusterStatsRequest request,
                                           List<ClusterStatsNodeResponse> responses, List<FailedNodeException> failures) {
    return new ClusterStatsResponse(
        System.currentTimeMillis(),
        clusterService.getClusterName(),
        responses,
        failures);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
private Response newResponse(
        Request request,
        AtomicReferenceArray responses,
        List<NoShardAvailableActionException> unavailableShardExceptions,
        Map<String, List<ShardRouting>> nodes,
        ClusterState clusterState) {
    int totalShards = 0;
    int successfulShards = 0;
    List<ShardOperationResult> broadcastByNodeResponses = new ArrayList<>();
    List<ShardOperationFailedException> exceptions = new ArrayList<>();
    for (int i = 0; i < responses.length(); i++) {
        if (responses.get(i) instanceof FailedNodeException) {
            FailedNodeException exception = (FailedNodeException) responses.get(i);
            totalShards += nodes.get(exception.nodeId()).size();
            for (ShardRouting shard : nodes.get(exception.nodeId())) {
                exceptions.add(new DefaultShardOperationFailedException(shard.getIndexName(), shard.getId(), exception));
            }
        } else {
            NodeResponse response = (NodeResponse) responses.get(i);
            broadcastByNodeResponses.addAll(response.results);
            totalShards += response.getTotalShards();
            successfulShards += response.getSuccessfulShards();
            for (BroadcastShardOperationFailedException throwable : response.getExceptions()) {
                if (!TransportActions.isShardNotAvailableException(throwable)) {
                    exceptions.add(new DefaultShardOperationFailedException(throwable.getShardId().getIndexName(), throwable.getShardId().getId(), throwable));
                }
            }
        }
    }
    totalShards += unavailableShardExceptions.size();
    int failedShards = exceptions.size();
    return newResponse(request, totalShards, successfulShards, failedShards, broadcastByNodeResponses, exceptions, clusterState);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
private void onFailure(int idx, String nodeId, Throwable t) {
    if (logger.isDebugEnabled() && !(t instanceof NodeShouldNotConnectException)) {
        logger.debug(
            (org.apache.logging.log4j.util.Supplier<?>)
                () -> new ParameterizedMessage("failed to execute on node [{}]", nodeId), t);
    }
    if (accumulateExceptions()) {
        responses.set(idx, new FailedNodeException(nodeId, "Failed node [" + nodeId + "]", t));
    }
    if (counter.incrementAndGet() == responses.length()) {
        finishHim();
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    clusterName = new ClusterName(in);
    nodes = readNodesFrom(in);
    failures = in.readList(FailedNodeException::new);
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
private void onFailure(int idx, String nodeId, Throwable t) {
    if (logger.isDebugEnabled() && !(t instanceof NodeShouldNotConnectException)) {
        logger.debug(
            (org.apache.logging.log4j.util.Supplier<?>)
                () -> new ParameterizedMessage("failed to execute on node [{}]", nodeId), t);
    }
    responses.set(idx, new FailedNodeException(nodeId, "Failed node [" + nodeId + "]", t));
    if (counter.incrementAndGet() == responses.length()) {
        finishHim();
    }
}
 

import org.elasticsearch.action.FailedNodeException; //导入依赖的package包/类
@Override
protected NodesResponse newResponse(NodesRequest request, List<NodeResponse> responses, List<FailedNodeException> failures) {
    if (request.getShouldFail()) {
        throw new IllegalStateException("Simulating operation failure");
    }
    return new NodesResponse(clusterService.getClusterName(), responses, failures);
}