Validate Search Phase Request Processors in Search Pipeline (#18363)

---------

Signed-off-by: Owais <[email protected]>

Author: owaiskazi19

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

@@ -311,25 +311,27 @@ void validatePipeline(Map<DiscoveryNode, SearchPipelineInfo> searchPipelineInfos
             new Processor.PipelineContext(Processor.PipelineSource.VALIDATE_PIPELINE)
         );
         List<Exception> exceptions = new ArrayList<>();
-        for (SearchRequestProcessor processor : pipeline.getSearchRequestProcessors()) {
-            for (Map.Entry<DiscoveryNode, SearchPipelineInfo> entry : searchPipelineInfos.entrySet()) {
-                String type = processor.getType();
-                if (entry.getValue().containsProcessor(Pipeline.REQUEST_PROCESSORS_KEY, type) == false) {
-                    String message = "Processor type [" + processor.getType() + "] is not installed on node [" + entry.getKey() + "]";
-                    exceptions.add(ConfigurationUtils.newConfigurationException(processor.getType(), processor.getTag(), null, message));
-                }
-            }
-        }
-        for (SearchResponseProcessor processor : pipeline.getSearchResponseProcessors()) {
+        validateProcessors(searchPipelineInfos, exceptions, Pipeline.REQUEST_PROCESSORS_KEY, pipeline.getSearchRequestProcessors());
+        validateProcessors(searchPipelineInfos, exceptions, Pipeline.RESPONSE_PROCESSORS_KEY, pipeline.getSearchResponseProcessors());
+        validateProcessors(searchPipelineInfos, exceptions, Pipeline.PHASE_PROCESSORS_KEY, pipeline.getSearchPhaseResultsProcessors());
+        ExceptionsHelper.rethrowAndSuppress(exceptions);
+    }
+
+    private void validateProcessors(
+        Map<DiscoveryNode, SearchPipelineInfo> searchPipelineInfos,
+        List<Exception> exceptions,
+        String processorKey,
+        List<? extends Processor> processors
+    ) {
+        for (Processor processor : processors) {
             for (Map.Entry<DiscoveryNode, SearchPipelineInfo> entry : searchPipelineInfos.entrySet()) {
                 String type = processor.getType();
-                if (entry.getValue().containsProcessor(Pipeline.RESPONSE_PROCESSORS_KEY, type) == false) {
+                if (entry.getValue().containsProcessor(processorKey, type) == false) {
                     String message = "Processor type [" + processor.getType() + "] is not installed on node [" + entry.getKey() + "]";
                     exceptions.add(ConfigurationUtils.newConfigurationException(processor.getType(), processor.getTag(), null, message));
                 }
             }
         }
-        ExceptionsHelper.rethrowAndSuppress(exceptions);
     }
 
     public void deletePipeline(DeleteSearchPipelineRequest request, ActionListener<AcknowledgedResponse> listener) throws Exception {
@@ -460,6 +462,10 @@ Map<String, Processor.Factory<SearchResponseProcessor>> getResponseProcessorFact
         return responseProcessorFactories;
     }
 
+    Map<String, Processor.Factory<SearchPhaseResultsProcessor>> getSearchPhaseResultsProcessorFactories() {
+        return phaseInjectorProcessorFactories;
+    }
+
     @Override
     public SearchPipelineInfo info() {
         List<ProcessorInfo> requestProcessorInfoList = requestProcessorFactories.keySet()

server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java

@@ -146,6 +146,10 @@ public void testSearchPipelinePlugin() {
             .getResponseProcessorFactories();
         assertEquals(1, responseProcessorFactories.size());
         assertTrue(responseProcessorFactories.containsKey("bar"));
+        Map<String, Processor.Factory<SearchPhaseResultsProcessor>> phaseInjectorProcessorFactories = searchPipelineService
+            .getSearchPhaseResultsProcessorFactories();
+        assertEquals(1, phaseInjectorProcessorFactories.size());
+        assertTrue(phaseInjectorProcessorFactories.containsKey("zoe"));
     }
 
     public void testSearchPipelinePluginDuplicate() {
@@ -879,6 +883,7 @@ public void testValidatePipeline() throws Exception {
 
         ProcessorInfo reqProcessor = new ProcessorInfo("scale_request_size");
         ProcessorInfo rspProcessor = new ProcessorInfo("fixed_score");
+        ProcessorInfo searchPhaseProcessor = new ProcessorInfo("max_score");
         DiscoveryNode n1 = new DiscoveryNode("n1", buildNewFakeTransportAddress(), Version.CURRENT);
         DiscoveryNode n2 = new DiscoveryNode("n2", buildNewFakeTransportAddress(), Version.CURRENT);
         PutSearchPipelineRequest putRequest = new PutSearchPipelineRequest(
@@ -901,7 +906,14 @@ public void testValidatePipeline() throws Exception {
         );
 
         SearchPipelineInfo completePipelineInfo = new SearchPipelineInfo(
-            Map.of(Pipeline.REQUEST_PROCESSORS_KEY, List.of(reqProcessor), Pipeline.RESPONSE_PROCESSORS_KEY, List.of(rspProcessor))
+            Map.of(
+                Pipeline.REQUEST_PROCESSORS_KEY,
+                List.of(reqProcessor),
+                Pipeline.RESPONSE_PROCESSORS_KEY,
+                List.of(rspProcessor),
+                Pipeline.PHASE_PROCESSORS_KEY,
+                List.of(searchPhaseProcessor)
+            )
         );
         SearchPipelineInfo incompletePipelineInfo = new SearchPipelineInfo(Map.of(Pipeline.REQUEST_PROCESSORS_KEY, List.of(reqProcessor)));
         // One node is missing a processor