[Bugfix] Better exception handling in search pipelines

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 @@
             );
         } 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 @@
                 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() {