handle reduce phase failures when searching (raise exception to listener)

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/ReduceSearchPhaseException.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.search;
+
+import org.elasticsearch.ElasticSearchException;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class ReduceSearchPhaseException extends ElasticSearchException {
+
+    public ReduceSearchPhaseException(String phaseName, String msg) {
+        super("Failed to reduce [" + phaseName + "] " + msg);
+    }
+
+    public ReduceSearchPhaseException(String phaseName, String msg, Throwable cause) {
+        super("Failed to reduce [" + phaseName + "]" + msg, cause);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryAndFetchAction.java

@@ -21,10 +21,7 @@
 
 import com.google.inject.Inject;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.search.SearchOperationThreading;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.*;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.node.Node;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -73,6 +70,10 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
             super(request, listener);
         }
 
+        @Override protected String firstPhaseName() {
+            return "dfs";
+        }
+
         @Override protected void sendExecuteFirstPhase(Node node, InternalSearchRequest request, SearchServiceListener<DfsSearchResult> listener) {
             searchService.sendExecuteDfs(node, request, listener);
         }
@@ -154,6 +155,14 @@ private void executeSecondPhase(final AtomicInteger counter, Node node, QuerySea
         }
 
         private void finishHim() {
+            try {
+                innerFinishHim();
+            } catch (Exception e) {
+                listener.onFailure(new ReduceSearchPhaseException("query_fetch", "", e));
+            }
+        }
+
+        private void innerFinishHim() {
             sortedShardList = searchPhaseController.sortDocs(queryFetchResults.values());
             final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults, queryFetchResults);
             String scrollIdX = null;

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java

@@ -21,10 +21,7 @@
 
 import com.google.inject.Inject;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.search.SearchOperationThreading;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.*;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.node.Node;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -77,6 +74,10 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
             super(request, listener);
         }
 
+        @Override protected String firstPhaseName() {
+            return "dfs";
+        }
+
         @Override protected void sendExecuteFirstPhase(Node node, InternalSearchRequest request, SearchServiceListener<DfsSearchResult> listener) {
             searchService.sendExecuteDfs(node, request, listener);
         }
@@ -160,6 +161,14 @@ private void executeQuery(final AtomicInteger counter, QuerySearchRequest queryS
         }
 
         private void executeFetchPhase() {
+            try {
+                innerExecuteFetchPhase();
+            } catch (Exception e) {
+                listener.onFailure(new ReduceSearchPhaseException("query", "", e));
+            }
+        }
+
+        private void innerExecuteFetchPhase() {
             sortedShardList = searchPhaseController.sortDocs(queryResults.values());
             final Map<SearchShardTarget, ExtTIntArrayList> docIdsToLoad = searchPhaseController.docIdsToLoad(sortedShardList);
 
@@ -236,6 +245,14 @@ private void executeFetch(final AtomicInteger counter, FetchSearchRequest fetchS
         }
 
         private void finishHim() {
+            try {
+                innerFinishHim();
+            } catch (Exception e) {
+                listener.onFailure(new ReduceSearchPhaseException("fetch", "", e));
+            }
+        }
+
+        private void innerFinishHim() {
             final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, fetchResults);
             String scrollIdX = null;
             if (request.scroll() != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java

@@ -64,6 +64,10 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
             super(request, listener);
         }
 
+        @Override protected String firstPhaseName() {
+            return "query_fetch";
+        }
+
         @Override protected void sendExecuteFirstPhase(Node node, InternalSearchRequest request, SearchServiceListener<QueryFetchSearchResult> listener) {
             searchService.sendExecuteFetch(node, request, listener);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java

@@ -21,10 +21,7 @@
 
 import com.google.inject.Inject;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.search.SearchOperationThreading;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.*;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.node.Node;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -71,6 +68,10 @@ private AsyncAction(SearchRequest request, ActionListener<SearchResponse> listen
             super(request, listener);
         }
 
+        @Override protected String firstPhaseName() {
+            return "query";
+        }
+
         @Override protected void sendExecuteFirstPhase(Node node, InternalSearchRequest request, SearchServiceListener<QuerySearchResult> listener) {
             searchService.sendExecuteQuery(node, request, listener);
         }
@@ -157,6 +158,14 @@ private void executeFetch(final AtomicInteger counter, FetchSearchRequest fetchS
         }
 
         private void finishHim() {
+            try {
+                innerFinishHim();
+            } catch (Exception e) {
+                listener.onFailure(new ReduceSearchPhaseException("fetch", "", e));
+            }
+        }
+
+        private void innerFinishHim() {
             InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, fetchResults);
             String scrollId = null;
             if (request.scroll() != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/type/TransportSearchTypeAction.java

@@ -20,10 +20,7 @@
 package org.elasticsearch.action.search.type;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.search.SearchOperationThreading;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.*;
 import org.elasticsearch.action.support.BaseAction;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
@@ -194,7 +191,11 @@ private void onFirstPhaseResult(ShardRouting shard, FirstResult result, Iterator
             }
             if (successulOps.incrementAndGet() == expectedSuccessfulOps ||
                     totalOps.incrementAndGet() == expectedTotalOps) {
-                moveToSecondPhase();
+                try {
+                    moveToSecondPhase();
+                } catch (Exception e) {
+                    listener.onFailure(new ReduceSearchPhaseException(firstPhaseName(), "", e));
+                }
             }
         }
 
@@ -207,7 +208,11 @@ private void onFirstPhaseResult(ShardRouting shard, final Iterator<ShardRouting>
             if (totalOps.incrementAndGet() == expectedTotalOps) {
                 // no more shards, add a failure
                 shardFailures.add(new ShardSearchFailure(t));
-                moveToSecondPhase();
+                try {
+                    moveToSecondPhase();
+                } catch (Exception e) {
+                    listener.onFailure(new ReduceSearchPhaseException(firstPhaseName(), "", e));
+                }
             } else {
                 if (shardIt.hasNext()) {
                     performFirstPhase(shardIt);
@@ -230,5 +235,7 @@ protected ShardSearchFailure[] buildShardFailures() {
         protected abstract void processFirstPhaseResult(ShardRouting shard, FirstResult result);
 
         protected abstract void moveToSecondPhase();
+
+        protected abstract String firstPhaseName();
     }
 }