Better exception handling in actions when forking to a thread pool

An execution on a thread pool might be rejected due to its settings, have better handling in those cases across the actions we have.
closes #3524

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

@@ -115,6 +115,7 @@ public class TransportSearchDfsQueryAndFetchAction extends TransportSearchTypeAc
                         final DiscoveryNode node = nodes.get(dfsResult.shardTarget().nodeId());
                         if (node.id().equals(nodes.localNodeId())) {
                             final QuerySearchRequest querySearchRequest = new QuerySearchRequest(request, dfsResult.id(), dfs);
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -125,6 +126,9 @@ public class TransportSearchDfsQueryAndFetchAction extends TransportSearchTypeAc
                                 } else {
                                     executeSecondPhase(entry.index, dfsResult, counter, node, querySearchRequest);
                                 }
+                            } catch (Throwable t) {
+                                onSecondPhaseFailure(t, querySearchRequest, entry.index, dfsResult, counter);
+                            }
                         }
                     }
                 }
@@ -144,17 +148,21 @@ public class TransportSearchDfsQueryAndFetchAction extends TransportSearchTypeAc
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onSecondPhaseFailure(t, querySearchRequest, shardIndex, dfsResult, counter);
+                }
+            });
+        }
+
+        void onSecondPhaseFailure(Throwable t, QuerySearchRequest querySearchRequest, int shardIndex, DfsSearchResult dfsResult, AtomicInteger counter) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute query phase", t, querySearchRequest.id());
             }
-                    AsyncAction.this.addShardFailure(shardIndex, dfsResult.shardTarget(), t);
+            this.addShardFailure(shardIndex, dfsResult.shardTarget(), t);
             successulOps.decrementAndGet();
             if (counter.decrementAndGet() == 0) {
                 finishHim();
             }
         }
-            });
-        }
 
         void finishHim() {
             try {

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

@@ -124,6 +124,7 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
                         final DiscoveryNode node = nodes.get(dfsResult.shardTarget().nodeId());
                         if (node.id().equals(nodes.localNodeId())) {
                             final QuerySearchRequest querySearchRequest = new QuerySearchRequest(request, dfsResult.id(), dfs);
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -134,6 +135,9 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
                                 } else {
                                     executeQuery(entry.index, dfsResult, counter, querySearchRequest, node);
                                 }
+                            } catch (Throwable t) {
+                                onQueryFailure(t, querySearchRequest, entry.index, dfsResult, counter);
+                            }
                         }
                     }
                 }
@@ -153,17 +157,21 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onQueryFailure(t, querySearchRequest, shardIndex, dfsResult, counter);
+                }
+            });
+        }
+
+        void onQueryFailure(Throwable t, QuerySearchRequest querySearchRequest, int shardIndex, DfsSearchResult dfsResult, AtomicInteger counter) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute query phase", t, querySearchRequest.id());
             }
-                    AsyncAction.this.addShardFailure(shardIndex, dfsResult.shardTarget(), t);
+            this.addShardFailure(shardIndex, dfsResult.shardTarget(), t);
             successulOps.decrementAndGet();
             if (counter.decrementAndGet() == 0) {
                 executeFetchPhase();
             }
         }
-            });
-        }
 
         void executeFetchPhase() {
             try {
@@ -217,6 +225,7 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
                         final DiscoveryNode node = nodes.get(queryResult.shardTarget().nodeId());
                         if (node.id().equals(nodes.localNodeId())) {
                             final FetchSearchRequest fetchSearchRequest = new FetchSearchRequest(request, queryResult.id(), entry.value);
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -227,6 +236,9 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
                                 } else {
                                     executeFetch(entry.index, queryResult.shardTarget(), counter, fetchSearchRequest, node);
                                 }
+                            } catch (Throwable t) {
+                                onFetchFailure(t, fetchSearchRequest, entry.index, queryResult.shardTarget(), counter);
+                            }
                         }
                     }
                 }
@@ -246,17 +258,21 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onFetchFailure(t, fetchSearchRequest, shardIndex, shardTarget, counter);
+                }
+            });
+        }
+
+        void onFetchFailure(Throwable t, FetchSearchRequest fetchSearchRequest, int shardIndex, SearchShardTarget shardTarget, AtomicInteger counter) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute fetch phase", t, fetchSearchRequest.id());
             }
-                    AsyncAction.this.addShardFailure(shardIndex, shardTarget, t);
+            this.addShardFailure(shardIndex, shardTarget, t);
             successulOps.decrementAndGet();
             if (counter.decrementAndGet() == 0) {
                 finishHim();
             }
         }
-            });
-        }
 
         void finishHim() {
             try {

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

@@ -126,6 +126,7 @@ public class TransportSearchQueryThenFetchAction extends TransportSearchTypeActi
                         final DiscoveryNode node = nodes.get(queryResult.shardTarget().nodeId());
                         if (node.id().equals(nodes.localNodeId())) {
                             final FetchSearchRequest fetchSearchRequest = new FetchSearchRequest(request, queryResult.id(), entry.value);
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -136,6 +137,9 @@ public class TransportSearchQueryThenFetchAction extends TransportSearchTypeActi
                                 } else {
                                     executeFetch(entry.index, queryResult.shardTarget(), counter, fetchSearchRequest, node);
                                 }
+                            } catch (Throwable t) {
+                                onFetchFailure(t, fetchSearchRequest, entry.index, queryResult.shardTarget(), counter);
+                            }
                         }
                     }
                 }
@@ -155,17 +159,21 @@ public class TransportSearchQueryThenFetchAction extends TransportSearchTypeActi
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onFetchFailure(t, fetchSearchRequest, shardIndex, shardTarget, counter);
+                }
+            });
+        }
+
+        void onFetchFailure(Throwable t, FetchSearchRequest fetchSearchRequest, int shardIndex, SearchShardTarget shardTarget, AtomicInteger counter) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute fetch phase", t, fetchSearchRequest.id());
             }
-                    AsyncAction.this.addShardFailure(shardIndex, shardTarget, t);
+            this.addShardFailure(shardIndex, shardTarget, t);
             successulOps.decrementAndGet();
             if (counter.decrementAndGet() == 0) {
                 finishHim();
             }
         }
-            });
-        }
 
         void finishHim() {
             try {

src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryAndFetchAction.java

@@ -170,6 +170,7 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent
                         final int shardIndex = i;
                         final DiscoveryNode node = nodes.get(target.v1());
                         if (node != null && nodes.localNodeId().equals(node.id())) {
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -180,6 +181,9 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent
                                 } else {
                                     executePhase(shardIndex, node, target.v2());
                                 }
+                            } catch (Throwable t) {
+                                onPhaseFailure(t, target.v2(), shardIndex);
+                            }
                         }
                     }
                 }
@@ -200,7 +204,7 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent
             }
         }
 
-        private void executePhase(final int shardIndex, DiscoveryNode node, final long searchId) {
+        void executePhase(final int shardIndex, DiscoveryNode node, final long searchId) {
             searchService.sendExecuteFetch(node, internalScrollSearchRequest(searchId, request), new SearchServiceListener<QueryFetchSearchResult>() {
                 @Override
                 public void onResult(QueryFetchSearchResult result) {
@@ -212,6 +216,12 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onPhaseFailure(t, searchId, shardIndex);
+                }
+            });
+        }
+
+        private void onPhaseFailure(Throwable t, long searchId, int shardIndex) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute query phase", t, searchId);
             }
@@ -221,8 +231,6 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent
                 finishHim();
             }
         }
-            });
-        }
 
         private void finishHim() {
             try {

src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryThenFetchAction.java

@@ -176,6 +176,7 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent
                         final int shardIndex = i;
                         final DiscoveryNode node = nodes.get(target.v1());
                         if (node != null && nodes.localNodeId().equals(node.id())) {
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -186,6 +187,9 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent
                                 } else {
                                     executeQueryPhase(shardIndex, counter, node, target.v2());
                                 }
+                            } catch (Throwable t) {
+                                onQueryPhaseFailure(shardIndex, counter, target.v2(), t);
+                            }
                         }
                     }
                 }
@@ -204,6 +208,12 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onQueryPhaseFailure(shardIndex, counter, searchId, t);
+                }
+            });
+        }
+
+        void onQueryPhaseFailure(final int shardIndex, final AtomicInteger counter, final long searchId, Throwable t) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute query phase", t, searchId);
             }
@@ -213,8 +223,6 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent
                 executeFetchPhase();
             }
         }
-            });
-        }
 
         private void executeFetchPhase() {
             sortedShardList = searchPhaseController.sortDocs(queryResults);

src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollScanAction.java

@@ -172,6 +172,7 @@ public class TransportSearchScrollScanAction extends AbstractComponent {
                         final int shardIndex = i;
                         final DiscoveryNode node = nodes.get(target.v1());
                         if (node != null && nodes.localNodeId().equals(node.id())) {
+                            try {
                                 if (localAsync) {
                                     threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                         @Override
@@ -182,6 +183,9 @@ public class TransportSearchScrollScanAction extends AbstractComponent {
                                 } else {
                                     executePhase(shardIndex, node, target.v2());
                                 }
+                            } catch (Throwable t) {
+                                onPhaseFailure(t, target.v2(), shardIndex);
+                            }
                         }
                     }
                 }
@@ -202,7 +206,7 @@ public class TransportSearchScrollScanAction extends AbstractComponent {
             }
         }
 
-        private void executePhase(final int shardIndex, DiscoveryNode node, final long searchId) {
+        void executePhase(final int shardIndex, DiscoveryNode node, final long searchId) {
             searchService.sendExecuteScan(node, internalScrollSearchRequest(searchId, request), new SearchServiceListener<QueryFetchSearchResult>() {
                 @Override
                 public void onResult(QueryFetchSearchResult result) {
@@ -214,6 +218,12 @@ public class TransportSearchScrollScanAction extends AbstractComponent {
 
                 @Override
                 public void onFailure(Throwable t) {
+                    onPhaseFailure(t, searchId, shardIndex);
+                }
+            });
+        }
+
+        void onPhaseFailure(Throwable t, long searchId, int shardIndex) {
             if (logger.isDebugEnabled()) {
                 logger.debug("[{}] Failed to execute query phase", t, searchId);
             }
@@ -223,8 +233,6 @@ public class TransportSearchScrollScanAction extends AbstractComponent {
                 finishHim();
             }
         }
-            });
-        }
 
         private void finishHim() {
             try {

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

@@ -179,12 +179,16 @@ public abstract class TransportSearchTypeAction extends TransportAction<SearchRe
                         if (shard != null) {
                             if (shard.currentNodeId().equals(nodes.localNodeId())) {
                                 if (localAsync) {
+                                    try {
                                         threadPool.executor(ThreadPool.Names.SEARCH).execute(new Runnable() {
                                             @Override
                                             public void run() {
                                                 performFirstPhase(fShardIndex, shardIt);
                                             }
                                         });
+                                    } catch (Throwable t) {
+                                        onFirstPhaseResult(shardIndex, shard, shard.currentNodeId(), shardIt, t);
+                                    }
                                 } else {
                                     performFirstPhase(fShardIndex, shardIt);
                                 }

src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastOperationAction.java

@@ -214,6 +214,7 @@ public abstract class TransportBroadcastOperationAction<Request extends Broadcas
                 // no more active shards... (we should not really get here, just safety)
                 onOperation(null, shardIt, shardIndex, new NoShardAvailableActionException(shardIt.shardId()));
             } else {
+                try {
                     final ShardRequest shardRequest = newShardRequest(shard, request);
                     if (shard.currentNodeId().equals(nodes.localNodeId())) {
                         if (localAsync) {
@@ -222,17 +223,13 @@ public abstract class TransportBroadcastOperationAction<Request extends Broadcas
                                 public void run() {
                                     try {
                                         onOperation(shard, shardIndex, shardOperation(shardRequest));
-                                } catch (Exception e) {
+                                    } catch (Throwable e) {
                                         onOperation(shard, shardIt, shardIndex, e);
                                     }
                                 }
                             });
                         } else {
-                        try {
                             onOperation(shard, shardIndex, shardOperation(shardRequest));
-                        } catch (Throwable e) {
-                            onOperation(shard, shardIt, shardIndex, e);
-                        }
                         }
                     } else {
                         DiscoveryNode node = nodes.get(shard.currentNodeId());
@@ -263,6 +260,9 @@ public abstract class TransportBroadcastOperationAction<Request extends Broadcas
                             });
                         }
                     }
+                } catch (Throwable e) {
+                    onOperation(shard, shardIt, shardIndex, e);
+                }
             }
         }
 

src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeOperationAction.java

@@ -137,6 +137,7 @@ public abstract class TransportMasterNodeOperationAction<Request extends MasterN
                     }
                 });
             } else {
+                try {
                     threadPool.executor(executor).execute(new Runnable() {
                         @Override
                         public void run() {
@@ -147,6 +148,9 @@ public abstract class TransportMasterNodeOperationAction<Request extends MasterN
                             }
                         }
                     });
+                } catch (Throwable t) {
+                    listener.onFailure(t);
+                }
             }
         } else {
             if (nodes.masterNode() == null) {

src/main/java/org/elasticsearch/action/support/nodes/TransportNodesOperationAction.java

@@ -142,6 +142,7 @@ public abstract class TransportNodesOperationAction<Request extends NodesOperati
             transportRequestOptions.withCompress(transportCompress());
             for (final String nodeId : nodesIds) {
                 final DiscoveryNode node = clusterState.nodes().nodes().get(nodeId);
+                try {
                     if (nodeId.equals("_local") || nodeId.equals(clusterState.nodes().localNodeId())) {
                         threadPool.executor(executor()).execute(new Runnable() {
                             @Override
@@ -192,6 +193,9 @@ public abstract class TransportNodesOperationAction<Request extends NodesOperati
                             });
                         }
                     }
+                } catch (Throwable t) {
+                    onFailure(nodeId, t);
+                }
             }
         }
 

src/main/java/org/elasticsearch/action/support/replication/TransportShardReplicationOperationAction.java

@@ -411,6 +411,7 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
 
                 foundPrimary = true;
                 if (shard.currentNodeId().equals(clusterState.nodes().localNodeId())) {
+                    try {
                         if (request.operationThreaded()) {
                             request.beforeLocalFork();
                             threadPool.executor(executor).execute(new Runnable() {
@@ -422,6 +423,9 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
                         } else {
                             performOnPrimary(shard.id(), fromClusterEvent, shard, clusterState);
                         }
+                    } catch (Throwable t) {
+                        listener.onFailure(t);
+                    }
                 } else {
                     DiscoveryNode node = clusterState.nodes().get(shard.currentNodeId());
                     transportService.sendRequest(node, transportAction, request, transportOptions, new BaseTransportResponseHandler<Response>() {
@@ -686,6 +690,7 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
             } else {
                 if (request.operationThreaded()) {
                     request.beforeLocalFork();
+                    try {
                         threadPool.executor(executor).execute(new Runnable() {
                             @Override
                             public void run() {
@@ -702,6 +707,17 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
                                 }
                             }
                         });
+                    } catch (Throwable e) {
+                        if (!ignoreReplicaException(e)) {
+                            logger.warn("Failed to perform " + transportAction + " on replica " + shardIt.shardId(), e);
+                            shardStateAction.shardFailed(shard, "Failed to perform [" + transportAction + "] on replica, message [" + detailedMessage(e) + "]");
+                        }
+                        // we want to decrement the counter here, in teh failure handling, cause we got rejected
+                        // from executing on the thread pool
+                        if (counter.decrementAndGet() == 0) {
+                            listener.onResponse(response.response());
+                        }
+                    }
                 } else {
                     try {
                         shardOperationOnReplica(shardRequest);

src/main/java/org/elasticsearch/action/support/single/custom/TransportSingleCustomOperationAction.java

@@ -219,6 +219,7 @@ public abstract class TransportSingleCustomOperationAction<Request extends Singl
                 if (shard.currentNodeId().equals(nodes.localNodeId())) {
                     // we don't prefer local shard, so try and do it here
                     if (!request.preferLocalShard()) {
+                        try {
                             if (request.operationThreaded()) {
                                 request.beforeLocalFork();
                                 threadPool.executor(executor).execute(new Runnable() {
@@ -233,13 +234,12 @@ public abstract class TransportSingleCustomOperationAction<Request extends Singl
                                     }
                                 });
                             } else {
-                            try {
                                 final Response response = shardOperation(request, shard.id());
                                 listener.onResponse(response);
+                            }
                         } catch (Throwable e) {
                             onFailure(shard, e);
                         }
-                        }
                     } else {
                         perform(lastException);
                     }

src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java

@@ -184,12 +184,13 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
             request.shardId = shardIt.shardId().id();
             if (shard.currentNodeId().equals(nodes.localNodeId())) {
                 request.beforeLocalFork();
+                try {
                     threadPool.executor(executor).execute(new Runnable() {
                         @Override
                         public void run() {
                             try {
                                 shardOperation(request, listener);
-                        } catch (Exception e) {
+                            } catch (Throwable e) {
                                 if (retryOnFailure(e)) {
                                     retry(fromClusterEvent, null);
                                 } else {
@@ -198,6 +199,13 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins
                             }
                         }
                     });
+                } catch (Throwable e) {
+                    if (retryOnFailure(e)) {
+                        retry(fromClusterEvent, null);
+                    } else {
+                        listener.onFailure(e);
+                    }
+                }
             } else {
                 DiscoveryNode node = nodes.get(shard.currentNodeId());
                 transportService.sendRequest(node, transportAction, request, transportOptions(), new BaseTransportResponseHandler<Response>() {

src/main/java/org/elasticsearch/action/support/single/shard/TransportShardSingleOperationAction.java

@@ -151,6 +151,7 @@ public abstract class TransportShardSingleOperationAction<Request extends Single
             }
 
             if (shardRouting.currentNodeId().equals(nodes.localNodeId())) {
+                try {
                     if (request.operationThreaded()) {
                         request.beforeLocalFork();
                         threadPool.executor(executor).execute(new Runnable() {
@@ -165,13 +166,12 @@ public abstract class TransportShardSingleOperationAction<Request extends Single
                             }
                         });
                     } else {
-                    try {
                         final Response response = shardOperation(request, shardRouting.id());
                         listener.onResponse(response);
+                    }
                 } catch (Throwable e) {
                     onFailure(shardRouting, e);
                 }
-                }
             } else {
                 DiscoveryNode node = nodes.get(shardRouting.currentNodeId());
                 transportService.sendRequest(node, transportShardAction, new ShardSingleOperationRequest(request, shardRouting.id()), new BaseTransportResponseHandler<Response>() {