handle nodes that are not connected early in AbstractSearchAsyncAction
@@ -41,6 +41,7 @@ import org.elasticsearch.search.internal.InternalSearchResponse;
import org.elasticsearch.search.internal.ShardSearchTransportRequest;
import org.elasticsearch.search.query.QuerySearchResult;
import org.elasticsearch.search.query.QuerySearchResultProvider;
+import org.elasticsearch.transport.ConnectTransportException;
import org.elasticsearch.transport.Transport;
import java.util.List;
@@ -120,13 +121,12 @@ abstract class AbstractSearchAsyncAction<FirstResult extends SearchPhaseResult>
void performFirstPhase(final int shardIndex, final ShardIterator shardIt, final ShardRouting shard) {
if (shard == null) {
+ // TODO upgrade this to an assert...
// no more active shards... (we should not really get here, but just for safety)
onFirstPhaseResult(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
} else {
+ try {
final Transport.Connection connection = nodeIdToConnection.apply(shard.currentNodeId());
- if (connection == null) {
- onFirstPhaseResult(shardIndex, shard, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
- } else {
AliasFilter filter = this.aliasFilter.get(shard.index().getUUID());
assert filter != null;
@@ -144,6 +144,10 @@ abstract class AbstractSearchAsyncAction<FirstResult extends SearchPhaseResult>
onFirstPhaseResult(shardIndex, shard, connection.getNode().getId(), shardIt, t);
}
});
+ } catch (ConnectTransportException | IllegalArgumentException ex) {
+ // we are getting the connection early here so we might run into nodes that are not connected. in that case we move on to
+ // the next shard.
+ onFirstPhaseResult(shardIndex, shard, shard.currentNodeId(), shardIt, ex);
}
}
}
