[Bugfix] Better exception handling in search pipelines

Thanks to @noCharger for reporting a failing negative test case.

Since we were rethrowing exceptions when resolving search pipelines and
processing search requests, that could end up killing the listener
thread.

Also, we want to make sure that any exception thrown from search
pipelines are wrapped in SearchPipelineProcessingException.

Signed-off-by: Michael Froh <froh@amazon.com>

modules/search-pipeline-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/30_filter_query.yml

@@ -122,3 +122,23 @@ teardown:
         index: test
         body: { }
   - match: { hits.total.value: 2 }
+---
+"Test invalid inline query":
+  - do:
+      catch: bad_request
+      search:
+        index: test
+        body: {
+          search_pipeline: {
+            "request_processors": [
+              "filter_query": {
+                "query": {
+                  "woozlewuzzle": {
+                    "field": "foo"
+                  }
+                }
+              }
+            ]
+          }
+        }
+  - match: { status: 400 }

server/src/main/java/org/opensearch/action/search/TransportSearchAction.java

@@ -401,7 +401,7 @@ private void executeRequest(
             );
         } catch (Exception e) {
             originalListener.onFailure(e);
-            throw new RuntimeException(e);
+            return;
         }
 
         ActionListener<SearchSourceBuilder> rewriteListener = ActionListener.wrap(source -> {

server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java

@@ -400,8 +400,12 @@ public PipelinedRequest resolvePipeline(SearchRequest searchRequest) throws Exce
                 pipeline = pipelineHolder.pipeline;
             }
         }
-        SearchRequest transformedRequest = pipeline.transformRequest(searchRequest);
-        return new PipelinedRequest(pipeline, transformedRequest);
+        try {
+            SearchRequest transformedRequest = pipeline.transformRequest(searchRequest);
+            return new PipelinedRequest(pipeline, transformedRequest);
+        } catch (Exception e) {
+            throw new SearchPipelineProcessingException(e);
+        }
     }
 
     Map<String, Processor.Factory<SearchRequestProcessor>> getRequestProcessorFactories() {