import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
private void onShardResult(int shardIndex, String nodeId, FirstResult result, ShardIterator shardIt) {
    result.shardTarget(new SearchShardTarget(nodeId, shardIt.shardId()));
    onShardSuccess(shardIndex, result);
    // we need to increment successful ops first before we compare the exit condition otherwise if we
    // are fast we could concurrently update totalOps but then preempt one of the threads which can
    // cause the successor to read a wrong value from successfulOps if second phase is very fast ie. count etc.
    // increment all the "future" shards to update the total ops since we some may work and some may not...
    // and when that happens, we break on total ops, so we must maintain them
    final int xTotalOps = totalOps.addAndGet(shardIt.remaining() + 1);
    if (xTotalOps == expectedTotalOps) {
        onPhaseDone();
    } else if (xTotalOps > expectedTotalOps) {
        throw new AssertionError("unexpected higher total ops [" + xTotalOps + "] compared to expected ["
            + expectedTotalOps + "]");
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public void testToXContent() throws IOException {
    ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(0, 0, "some message", null),
            new SearchShardTarget("nodeId", new ShardId(new Index("indexName", "indexUuid"), 123)));
    BytesReference xContent = toXContent(failure, XContentType.JSON, randomBoolean());
    assertEquals(
            "{\"shard\":123,"
                    + "\"index\":\"indexName\","
                    + "\"node\":\"nodeId\","
                    + "\"reason\":{"
                        + "\"type\":\"parsing_exception\","
                        + "\"reason\":\"some message\","
                        + "\"line\":0,"
                        + "\"col\":0"
                    + "}"
            + "}",
            xContent.utf8ToString());
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public CrateSearchContext(long id,
                          final long nowInMillis,
                          SearchShardTarget shardTarget,
                          Engine.Searcher engineSearcher,
                          IndexService indexService,
                          final IndexShard indexShard,
                          ScriptService scriptService,
                          PageCacheRecycler pageCacheRecycler,
                          BigArrays bigArrays,
                          Counter timeEstimateCounter,
                          Optional<Scroll> scroll) {
    super(id, new CrateSearchShardRequest(nowInMillis, scroll, indexShard),
            shardTarget, engineSearcher, indexService,
            indexShard, scriptService, pageCacheRecycler,
            bigArrays, timeEstimateCounter, ParseFieldMatcher.STRICT, SearchService.NO_TIMEOUT);
    this.engineSearcher = engineSearcher;
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public PercolateContext(PercolateShardRequest request, SearchShardTarget searchShardTarget, IndexShard indexShard,
                        IndexService indexService, PageCacheRecycler pageCacheRecycler,
                        BigArrays bigArrays, ScriptService scriptService, Query aliasFilter, ParseFieldMatcher parseFieldMatcher) {
    super(parseFieldMatcher, request);
    this.indexShard = indexShard;
    this.indexService = indexService;
    this.fieldDataService = indexService.fieldData();
    this.searchShardTarget = searchShardTarget;
    this.percolateQueries = indexShard.percolateRegistry().percolateQueries();
    this.types = new String[]{request.documentType()};
    this.pageCacheRecycler = pageCacheRecycler;
    this.bigArrays = bigArrays.withCircuitBreaking();
    this.querySearchResult = new QuerySearchResult(0, searchShardTarget);
    this.engineSearcher = indexShard.acquireSearcher("percolate");
    this.searcher = new ContextIndexSearcher(engineSearcher, indexService.cache().query(), indexShard.getQueryCachingPolicy());
    this.scriptService = scriptService;
    this.numberOfShards = request.getNumberOfShards();
    this.aliasFilter = aliasFilter;
    this.startTime = request.getStartTime();
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public DefaultSearchContext(long id, ShardSearchRequest request, SearchShardTarget shardTarget,
                            Engine.Searcher engineSearcher, IndexService indexService, IndexShard indexShard,
                            ScriptService scriptService, PageCacheRecycler pageCacheRecycler,
                            BigArrays bigArrays, Counter timeEstimateCounter, ParseFieldMatcher parseFieldMatcher,
                            TimeValue timeout
) {
    super(parseFieldMatcher, request);
    this.id = id;
    this.request = request;
    this.searchType = request.searchType();
    this.shardTarget = shardTarget;
    this.engineSearcher = engineSearcher;
    this.scriptService = scriptService;
    this.pageCacheRecycler = pageCacheRecycler;
    // SearchContexts use a BigArrays that can circuit break
    this.bigArrays = bigArrays.withCircuitBreaking();
    this.dfsResult = new DfsSearchResult(id, shardTarget);
    this.queryResult = new QuerySearchResult(id, shardTarget);
    this.fetchResult = new FetchSearchResult(id, shardTarget);
    this.indexShard = indexShard;
    this.indexService = indexService;
    this.searcher = new ContextIndexSearcher(engineSearcher, indexService.cache().query(), indexShard.getQueryCachingPolicy());
    this.timeEstimateCounter = timeEstimateCounter;
    this.timeoutInMillis = timeout.millis();
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
void onFirstPhaseResult(int shardIndex, ShardRouting shard, FirstResult result, ShardIterator shardIt) {
    result.shardTarget(new SearchShardTarget(shard.currentNodeId(), shard.index(), shard.id()));
    processFirstPhaseResult(shardIndex, result);
    // we need to increment successful ops first before we compare the exit condition otherwise if we
    // are fast we could concurrently update totalOps but then preempt one of the threads which can
    // cause the successor to read a wrong value from successfulOps if second phase is very fast ie. count etc.
    successfulOps.incrementAndGet();
    // increment all the "future" shards to update the total ops since we some may work and some may not...
    // and when that happens, we break on total ops, so we must maintain them
    final int xTotalOps = totalOps.addAndGet(shardIt.remaining() + 1);
    if (xTotalOps == expectedTotalOps) {
        try {
            innerMoveToSecondPhase();
        } catch (Throwable e) {
            if (logger.isDebugEnabled()) {
                logger.debug(shardIt.shardId() + ": Failed to execute [" + request + "] while moving to second phase", e);
            }
            raiseEarlyFailure(new ReduceSearchPhaseException(firstPhaseName(), "", e, buildShardFailures()));
        }
    } else if (xTotalOps > expectedTotalOps) {
        raiseEarlyFailure(new IllegalStateException("unexpected higher total ops [" + xTotalOps + "] compared to expected [" + expectedTotalOps + "]"));
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
protected final void addShardFailure(final int shardIndex, @Nullable SearchShardTarget shardTarget, Throwable t) {
    // we don't aggregate shard failures on non active shards (but do keep the header counts right)
    if (TransportActions.isShardNotAvailableException(t)) {
        return;
    }

    // lazily create shard failures, so we can early build the empty shard failure list in most cases (no failures)
    if (shardFailures == null) {
        synchronized (shardFailuresMutex) {
            if (shardFailures == null) {
                shardFailures = new AtomicArray<>(shardsIts.size());
            }
        }
    }
    ShardSearchFailure failure = shardFailures.get(shardIndex);
    if (failure == null) {
        shardFailures.set(shardIndex, new ShardSearchFailure(t, shardTarget));
    } else {
        // the failure is already present, try and not override it with an exception that is less meaningless
        // for example, getting illegal shard state
        if (TransportActions.isReadOverrideException(t)) {
            shardFailures.set(shardIndex, new ShardSearchFailure(t, shardTarget));
        }
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
void executeFetch(final int shardIndex, final SearchShardTarget shardTarget, final AtomicInteger counter,
                  final ShardFetchSearchRequest fetchSearchRequest, DiscoveryNode node) {
    searchService.sendExecuteFetch(node, fetchSearchRequest, new ActionListener<FetchSearchResult>() {
        @Override
        public void onResponse(FetchSearchResult result) {
            result.shardTarget(shardTarget);
            fetchResults.set(shardIndex, result);
            if (counter.decrementAndGet() == 0) {
                finishHim();
            }
        }

        @Override
        public void onFailure(Throwable t) {
            // the search context might not be cleared on the node where the fetch was executed for example
            // because the action was rejected by the thread pool. in this case we need to send a dedicated
            // request to clear the search context. by setting docIdsToLoad to null, the context will be cleared
            // in TransportSearchTypeAction.releaseIrrelevantSearchContexts() after the search request is done.
            docIdsToLoad.set(shardIndex, null);
            onFetchFailure(t, fetchSearchRequest, shardIndex, shardTarget, counter);
        }
    });
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    shardTarget = new SearchShardTarget(in);
    queryResult = readQuerySearchResult(in);
    queryResult.shardTarget(shardTarget);
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    shardTarget = new SearchShardTarget(in);
    result = readQueryFetchSearchResult(in);
    result.shardTarget(shardTarget);
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
/**
 * Finalizes the shard in repository and then removes it from cluster state
 * <p>
 * This is non-blocking method that runs on a thread from SNAPSHOT thread pool
 *
 * @param entry   snapshot
 * @param failure failure reason or null if snapshot was successful
 */
private void endSnapshot(final SnapshotsInProgress.Entry entry, final String failure) {
    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(new Runnable() {
        @Override
        public void run() {
            final Snapshot snapshot = entry.snapshot();
            try {
                final Repository repository = repositoriesService.repository(snapshot.getRepository());
                logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshot, entry.state(), failure);
                ArrayList<ShardSearchFailure> failures = new ArrayList<>();
                ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>();
                for (ObjectObjectCursor<ShardId, ShardSnapshotStatus> shardStatus : entry.shards()) {
                    ShardId shardId = shardStatus.key;
                    ShardSnapshotStatus status = shardStatus.value;
                    if (status.state().failed()) {
                        failures.add(new ShardSearchFailure(status.reason(), new SearchShardTarget(status.nodeId(), shardId)));
                        shardFailures.add(new SnapshotShardFailure(status.nodeId(), shardId, status.reason()));
                    }
                }
                SnapshotInfo snapshotInfo = repository.finalizeSnapshot(
                    snapshot.getSnapshotId(),
                    entry.indices(),
                    entry.startTime(),
                    failure,
                    entry.shards().size(),
                    Collections.unmodifiableList(shardFailures),
                    entry.getRepositoryStateId());
                removeSnapshotFromClusterState(snapshot, snapshotInfo, null);
            } catch (Exception e) {
                logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] failed to finalize snapshot", snapshot), e);
                removeSnapshotFromClusterState(snapshot, null, e);
            }
        }
    });
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public final void onShardFailure(final int shardIndex, @Nullable SearchShardTarget shardTarget, Exception e) {
    // we don't aggregate shard failures on non active shards (but do keep the header counts right)
    if (TransportActions.isShardNotAvailableException(e)) {
        return;
    }
    AtomicArray<ShardSearchFailure> shardFailures = this.shardFailures.get();
    // lazily create shard failures, so we can early build the empty shard failure list in most cases (no failures)
    if (shardFailures == null) { // this is double checked locking but it's fine since SetOnce uses a volatile read internally
        synchronized (shardFailuresMutex) {
            shardFailures = this.shardFailures.get(); // read again otherwise somebody else has created it?
            if (shardFailures == null) { // still null so we are the first and create a new instance
                shardFailures = new AtomicArray<>(getNumShards());
                this.shardFailures.set(shardFailures);
            }
        }
    }
    ShardSearchFailure failure = shardFailures.get(shardIndex);
    if (failure == null) {
        shardFailures.set(shardIndex, new ShardSearchFailure(e, shardTarget));
    } else {
        // the failure is already present, try and not override it with an exception that is less meaningless
        // for example, getting illegal shard state
        if (TransportActions.isReadOverrideException(e)) {
            shardFailures.set(shardIndex, new ShardSearchFailure(e, shardTarget));
        }
    }

    if (results.hasResult(shardIndex)) {
        assert failure == null : "shard failed before but shouldn't: " + failure;
        successfulOps.decrementAndGet(); // if this shard was successful before (initial phase) we have to adjust the counter
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
private void executeFetch(final int shardIndex, final SearchShardTarget shardTarget,
                          final CountedCollector<FetchSearchResult> counter,
                          final ShardFetchSearchRequest fetchSearchRequest, final QuerySearchResult querySearchResult,
                          final Transport.Connection connection) {
    context.getSearchTransport().sendExecuteFetch(connection, fetchSearchRequest, context.getTask(),
        new ActionListener<FetchSearchResult>() {
            @Override
            public void onResponse(FetchSearchResult result) {
                counter.onResult(shardIndex, result, shardTarget);
            }

            @Override
            public void onFailure(Exception e) {
                try {
                    if (logger.isDebugEnabled()) {
                        logger.debug((Supplier<?>) () -> new ParameterizedMessage("[{}] Failed to execute fetch phase",
                            fetchSearchRequest.id()), e);
                    }
                    counter.onFailure(shardIndex, shardTarget, e);
                } finally {
                    // the search context might not be cleared on the node where the fetch was executed for example
                    // because the action was rejected by the thread pool. in this case we need to send a dedicated
                    // request to clear the search context.
                    releaseIrrelevantSearchContext(querySearchResult);
                }
            }
        });
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public ShardSearchFailure(Exception e, @Nullable SearchShardTarget shardTarget) {
    final Throwable actual = ExceptionsHelper.unwrapCause(e);
    if (actual != null && actual instanceof SearchException) {
        this.shardTarget = ((SearchException) actual).shard();
    } else if (shardTarget != null) {
        this.shardTarget = shardTarget;
    }
    status = ExceptionsHelper.status(actual);
    this.reason = ExceptionsHelper.detailedMessage(e);
    this.cause = actual;
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    if (in.readBoolean()) {
        shardTarget = new SearchShardTarget(in);
    }
    reason = in.readString();
    status = RestStatus.readFrom(in);
    cause = in.readException();
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public static ShardSearchFailure fromXContent(XContentParser parser) throws IOException {
    XContentParser.Token token;
    ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation);
    String currentFieldName = null;
    int shardId = -1;
    String indexName = null;
    String nodeId = null;
    ElasticsearchException exception = null;
    while((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
        if (token == XContentParser.Token.FIELD_NAME) {
            currentFieldName = parser.currentName();
        } else if (token.isValue()) {
            if (SHARD_FIELD.equals(currentFieldName)) {
                shardId  = parser.intValue();
            } else if (INDEX_FIELD.equals(currentFieldName)) {
                indexName  = parser.text();
            } else if (NODE_FIELD.equals(currentFieldName)) {
                nodeId  = parser.text();
            } else {
                throwUnknownField(currentFieldName, parser.getTokenLocation());
            }
        } else if (token == XContentParser.Token.START_OBJECT) {
            if (REASON_FIELD.equals(currentFieldName)) {
                exception = ElasticsearchException.fromXContent(parser);
            } else {
                throwUnknownField(currentFieldName, parser.getTokenLocation());
            }
        } else {
            throwUnknownToken(token, parser.getTokenLocation());
        }
    }
    return new ShardSearchFailure(exception,
            new SearchShardTarget(nodeId, new ShardId(new Index(indexName, IndexMetaData.INDEX_UUID_NA_VALUE), shardId)));
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
/**
 * Sets the result to the given array index and then runs {@link #countDown()}
 */
void onResult(int index, R result, SearchShardTarget target) {
    try {
        result.shardTarget(target);
        resultConsumer.consume(index, result);
    } finally {
        countDown();
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
/**
 * Escalates the failure via {@link SearchPhaseContext#onShardFailure(int, SearchShardTarget, Exception)}
 * and then runs {@link #countDown()}
 */
void onFailure(final int shardIndex, @Nullable SearchShardTarget shardTarget, Exception e) {
    try {
        context.onShardFailure(shardIndex, shardTarget, e);
    } finally {
        countDown();
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
@Override
public void run() throws IOException {
    // TODO we can potentially also consume the actual per shard results from the initial phase here in the aggregateDfs
    // to free up memory early
    final AggregatedDfs dfs = searchPhaseController.aggregateDfs(dfsSearchResults);
    final CountedCollector<QuerySearchResultProvider> counter = new CountedCollector<>(queryResult::consumeResult,
        dfsSearchResults.asList().size(),
        () -> {
            context.executeNextPhase(this, nextPhaseFactory.apply(queryResult));
        }, context);
    for (final AtomicArray.Entry<DfsSearchResult> entry : dfsSearchResults.asList()) {
        DfsSearchResult dfsResult = entry.value;
        final int shardIndex = entry.index;
        final SearchShardTarget searchShardTarget = dfsResult.shardTarget();
        Transport.Connection connection = context.getConnection(searchShardTarget.getNodeId());
        QuerySearchRequest querySearchRequest = new QuerySearchRequest(context.getRequest(), dfsResult.id(), dfs);
        searchTransportService.sendExecuteQuery(connection, querySearchRequest, context.getTask(),
            ActionListener.wrap(
                result -> counter.onResult(shardIndex, result, searchShardTarget),
                exception ->  {
                    try {
                        if (context.getLogger().isDebugEnabled()) {
                            context.getLogger().debug((Supplier<?>) () -> new ParameterizedMessage("[{}] Failed to execute query phase",
                                querySearchRequest.id()), exception);
                        }
                        counter.onFailure(shardIndex, searchShardTarget, exception);
                    } finally {
                        // the query might not have been executed at all (for example because thread pool rejected
                        // execution) and the search context that was created in dfs phase might not be released.
                        // release it again to be in the safe side
                        context.sendReleaseSearchContext(querySearchRequest.id(), connection);
                    }
                }));
    }
}
 

import org.elasticsearch.search.SearchShardTarget; //导入依赖的package包/类
public void testConvert() throws IOException {
    RestRequest request = new FakeRestRequest();
    RestChannel channel = new DetailedExceptionRestChannel(request);
    ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
            new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
    ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
            new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
    SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",  new ShardSearchFailure[] {failure, failure1});
    BytesRestResponse response = new BytesRestResponse(channel, new RemoteTransportException("foo", ex));
    String text = response.content().utf8ToString();
    String expected = "{\"error\":{\"root_cause\":[{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}],\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}]},\"status\":400}";
    assertEquals(expected.trim(), text.trim());
    String stackTrace = ExceptionsHelper.stackTrace(ex);
    assertTrue(stackTrace.contains("Caused by: ParsingException[foobar]"));
}