From 67645aba4c296167c53d06da9e853acaeafa451a Mon Sep 17 00:00:00 2001 From: Michael Froh Date: Thu, 29 Jun 2023 01:04:53 +0000 Subject: [PATCH] Move PipelineSource into PipelineContext and explicitly pass to create Signed-off-by: Michael Froh --- .../common/FilterQueryRequestProcessor.java | 3 +- .../common/RenameFieldResponseProcessor.java | 3 +- .../common/ScriptRequestProcessor.java | 13 +-- .../SearchPipelineCommonModulePlugin.java | 4 +- .../FilterQueryRequestProcessorTests.java | 7 +- .../RenameFieldResponseProcessorTests.java | 7 +- .../plugins/SearchPipelinePlugin.java | 88 ++++++++++++++++- .../search/pipeline/PipelineWithMetrics.java | 35 +++++-- .../opensearch/search/pipeline/Processor.java | 97 ++++--------------- .../pipeline/SearchPipelineService.java | 11 ++- .../pipeline/SearchPipelineServiceTests.java | 39 ++++---- 11 files changed, 175 insertions(+), 132 deletions(-) diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessor.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessor.java index a43335b73a72e..d8862aa59cede 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessor.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessor.java @@ -101,7 +101,8 @@ public FilterQueryRequestProcessor create( Map> processorFactories, String tag, String description, - Map config + Map config, + PipelineContext pipelineContext ) throws Exception { Map query = ConfigurationUtils.readOptionalMap(TYPE, tag, config, QUERY_KEY); if (query == null) { diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessor.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessor.java index 4c40dda5928f0..c8b3c06a71562 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessor.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessor.java @@ -140,7 +140,8 @@ public RenameFieldResponseProcessor create( Map> processorFactories, String tag, String description, - Map config + Map config, + PipelineContext pipelineContext ) throws Exception { String oldField = ConfigurationUtils.readStringProperty(TYPE, tag, config, "field"); String newField = ConfigurationUtils.readStringProperty(TYPE, tag, config, "target_field"); diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/ScriptRequestProcessor.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/ScriptRequestProcessor.java index 23163d6ce2f97..43ab3d4622d6b 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/ScriptRequestProcessor.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/ScriptRequestProcessor.java @@ -141,22 +141,13 @@ public Factory(ScriptService scriptService) { this.scriptService = scriptService; } - /** - * Creates a new instance of {@link ScriptRequestProcessor}. - * - * @param registry The registry of processor factories. - * @param processorTag The processor's tag. - * @param description The processor's description. - * @param config The configuration options for the processor. - * @return The created {@link ScriptRequestProcessor} instance. - * @throws Exception if an error occurs during the creation process. - */ @Override public ScriptRequestProcessor create( Map> registry, String processorTag, String description, - Map config + Map config, + PipelineContext pipelineContext ) throws Exception { Map scriptConfig = new HashMap<>(); for (String key : SCRIPT_CONFIG_KEYS) { diff --git a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java index dc25de460fdba..49681b80fdead 100644 --- a/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java +++ b/modules/search-pipeline-common/src/main/java/org/opensearch/search/pipeline/common/SearchPipelineCommonModulePlugin.java @@ -33,7 +33,7 @@ public SearchPipelineCommonModulePlugin() {} * @return A map of processor factories, where the keys are the processor types and the values are the corresponding factory instances. */ @Override - public Map> getRequestProcessors(Processor.Parameters parameters) { + public Map> getRequestProcessors(Parameters parameters) { return Map.of( FilterQueryRequestProcessor.TYPE, new FilterQueryRequestProcessor.Factory(parameters.namedXContentRegistry), @@ -43,7 +43,7 @@ public Map> getRequestProcesso } @Override - public Map> getResponseProcessors(Processor.Parameters parameters) { + public Map> getResponseProcessors(Parameters parameters) { return Map.of(RenameFieldResponseProcessor.TYPE, new RenameFieldResponseProcessor.Factory()); } } diff --git a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessorTests.java b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessorTests.java index ea3008e005508..ecf746af556a2 100644 --- a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessorTests.java +++ b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/FilterQueryRequestProcessorTests.java @@ -39,10 +39,13 @@ public void testFilterQuery() throws Exception { public void testFactory() throws Exception { FilterQueryRequestProcessor.Factory factory = new FilterQueryRequestProcessor.Factory(this.xContentRegistry()); Map configMap = new HashMap<>(Map.of("query", Map.of("term", Map.of("field", "value")))); - FilterQueryRequestProcessor processor = factory.create(Collections.emptyMap(), null, null, configMap); + FilterQueryRequestProcessor processor = factory.create(Collections.emptyMap(), null, null, configMap, null); assertEquals(new TermQueryBuilder("field", "value"), processor.filterQuery); // Missing "query" parameter: - expectThrows(IllegalArgumentException.class, () -> factory.create(Collections.emptyMap(), null, null, Collections.emptyMap())); + expectThrows( + IllegalArgumentException.class, + () -> factory.create(Collections.emptyMap(), null, null, Collections.emptyMap(), null) + ); } } diff --git a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessorTests.java b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessorTests.java index a2fc7f6acfa7c..7f3a2acfbdc08 100644 --- a/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessorTests.java +++ b/modules/search-pipeline-common/src/test/java/org/opensearch/search/pipeline/common/RenameFieldResponseProcessorTests.java @@ -115,12 +115,15 @@ public void testFactory() throws Exception { config.put("target_field", newField); RenameFieldResponseProcessor.Factory factory = new RenameFieldResponseProcessor.Factory(); - RenameFieldResponseProcessor processor = factory.create(Collections.emptyMap(), null, null, config); + RenameFieldResponseProcessor processor = factory.create(Collections.emptyMap(), null, null, config, null); assertEquals(processor.getType(), "rename_field"); assertEquals(processor.getOldField(), oldField); assertEquals(processor.getNewField(), newField); assertFalse(processor.isIgnoreMissing()); - expectThrows(OpenSearchParseException.class, () -> factory.create(Collections.emptyMap(), null, null, Collections.emptyMap())); + expectThrows( + OpenSearchParseException.class, + () -> factory.create(Collections.emptyMap(), null, null, Collections.emptyMap(), null) + ); } } diff --git a/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java b/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java index 3d76bab93a60c..d2ef2b65c5944 100644 --- a/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java +++ b/server/src/main/java/org/opensearch/plugins/SearchPipelinePlugin.java @@ -8,13 +8,24 @@ package org.opensearch.plugins; +import org.opensearch.client.Client; +import org.opensearch.common.util.concurrent.ThreadContext; +import org.opensearch.core.xcontent.NamedXContentRegistry; +import org.opensearch.env.Environment; +import org.opensearch.index.analysis.AnalysisRegistry; +import org.opensearch.script.ScriptService; import org.opensearch.search.pipeline.Processor; import org.opensearch.search.pipeline.SearchPhaseResultsProcessor; +import org.opensearch.search.pipeline.SearchPipelineService; import org.opensearch.search.pipeline.SearchRequestProcessor; import org.opensearch.search.pipeline.SearchResponseProcessor; +import org.opensearch.threadpool.Scheduler; import java.util.Collections; import java.util.Map; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.LongSupplier; /** * An extension point for {@link Plugin} implementation to add custom search pipeline processors. @@ -29,7 +40,7 @@ public interface SearchPipelinePlugin { * in pipeline configurations, and the value is a {@link org.opensearch.search.pipeline.Processor.Factory} * to create the processor from a given pipeline configuration. */ - default Map> getRequestProcessors(Processor.Parameters parameters) { + default Map> getRequestProcessors(Parameters parameters) { return Collections.emptyMap(); } @@ -40,7 +51,7 @@ default Map> getRequestProcess * in pipeline configurations, and the value is a {@link org.opensearch.search.pipeline.Processor.Factory} * to create the processor from a given pipeline configuration. */ - default Map> getResponseProcessors(Processor.Parameters parameters) { + default Map> getResponseProcessors(Parameters parameters) { return Collections.emptyMap(); } @@ -51,7 +62,78 @@ default Map> getResponseProce * in pipeline configurations, and the value is a {@link org.opensearch.search.pipeline.Processor.Factory} * to create the processor from a given pipeline configuration. */ - default Map> getSearchPhaseResultsProcessors(Processor.Parameters parameters) { + default Map> getSearchPhaseResultsProcessors(Parameters parameters) { return Collections.emptyMap(); } + + /** + * Infrastructure class that holds services that can be used by processor factories to create processor instances + * and that gets passed around to all {@link SearchPipelinePlugin}s. + */ + class Parameters { + + /** + * Useful to provide access to the node's environment like config directory to processor factories. + */ + public final Environment env; + + /** + * Provides processors script support. + */ + public final ScriptService scriptService; + + /** + * Provide analyzer support + */ + public final AnalysisRegistry analysisRegistry; + + /** + * Allows processors to read headers set by {@link org.opensearch.action.support.ActionFilter} + * instances that have run while handling the current search. + */ + public final ThreadContext threadContext; + + public final LongSupplier relativeTimeSupplier; + + public final SearchPipelineService searchPipelineService; + + public final Consumer genericExecutor; + + public final NamedXContentRegistry namedXContentRegistry; + + /** + * Provides scheduler support + */ + public final BiFunction scheduler; + + /** + * Provides access to the node's cluster client + */ + public final Client client; + + public Parameters( + Environment env, + ScriptService scriptService, + AnalysisRegistry analysisRegistry, + ThreadContext threadContext, + LongSupplier relativeTimeSupplier, + BiFunction scheduler, + SearchPipelineService searchPipelineService, + Client client, + Consumer genericExecutor, + NamedXContentRegistry namedXContentRegistry + ) { + this.env = env; + this.scriptService = scriptService; + this.threadContext = threadContext; + this.analysisRegistry = analysisRegistry; + this.relativeTimeSupplier = relativeTimeSupplier; + this.scheduler = scheduler; + this.searchPipelineService = searchPipelineService; + this.client = client; + this.genericExecutor = genericExecutor; + this.namedXContentRegistry = namedXContentRegistry; + } + + } } diff --git a/server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java b/server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java index 612e979e56070..060894a37e5ed 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java +++ b/server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java @@ -77,19 +77,28 @@ static PipelineWithMetrics create( Map> phaseResultsProcessorFactories, NamedWriteableRegistry namedWriteableRegistry, OperationMetrics totalRequestProcessingMetrics, - OperationMetrics totalResponseProcessingMetrics + OperationMetrics totalResponseProcessingMetrics, + Processor.PipelineContext pipelineContext ) throws Exception { String description = ConfigurationUtils.readOptionalStringProperty(null, null, config, DESCRIPTION_KEY); Integer version = ConfigurationUtils.readIntProperty(null, null, config, VERSION_KEY, null); List> requestProcessorConfigs = ConfigurationUtils.readOptionalList(null, null, config, REQUEST_PROCESSORS_KEY); - List requestProcessors = readProcessors(requestProcessorFactories, requestProcessorConfigs); + List requestProcessors = readProcessors( + requestProcessorFactories, + requestProcessorConfigs, + pipelineContext + ); List> responseProcessorConfigs = ConfigurationUtils.readOptionalList( null, null, config, RESPONSE_PROCESSORS_KEY ); - List responseProcessors = readProcessors(responseProcessorFactories, responseProcessorConfigs); + List responseProcessors = readProcessors( + responseProcessorFactories, + responseProcessorConfigs, + pipelineContext + ); List> phaseResultsProcessorConfigs = ConfigurationUtils.readOptionalList( null, null, @@ -98,7 +107,8 @@ static PipelineWithMetrics create( ); List phaseResultsProcessors = readProcessors( phaseResultsProcessorFactories, - phaseResultsProcessorConfigs + phaseResultsProcessorConfigs, + pipelineContext ); if (config.isEmpty() == false) { throw new OpenSearchParseException( @@ -125,7 +135,8 @@ static PipelineWithMetrics create( private static List readProcessors( Map> processorFactories, - List> requestProcessorConfigs + List> requestProcessorConfigs, + Processor.PipelineContext pipelineContext ) throws Exception { List processors = new ArrayList<>(); if (requestProcessorConfigs == null) { @@ -140,7 +151,19 @@ private static List readProcessors( Map config = (Map) entry.getValue(); String tag = ConfigurationUtils.readOptionalStringProperty(null, null, config, TAG_KEY); String description = ConfigurationUtils.readOptionalStringProperty(null, tag, config, DESCRIPTION_KEY); - processors.add(processorFactories.get(type).create(processorFactories, tag, description, config)); + processors.add(processorFactories.get(type).create(processorFactories, tag, description, config, pipelineContext)); + if (config.isEmpty() == false) { + String processorName = type; + if (tag != null) { + processorName = processorName + ":" + tag; + } + throw new OpenSearchParseException( + "processor [" + + processorName + + "] doesn't support one or more provided configuration parameters: " + + Arrays.toString(config.keySet().toArray()) + ); + } } } return Collections.unmodifiableList(processors); diff --git a/server/src/main/java/org/opensearch/search/pipeline/Processor.java b/server/src/main/java/org/opensearch/search/pipeline/Processor.java index 6fa28c7708e51..cc96132479c74 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/Processor.java +++ b/server/src/main/java/org/opensearch/search/pipeline/Processor.java @@ -8,19 +8,7 @@ package org.opensearch.search.pipeline; -import org.opensearch.client.Client; -import org.opensearch.common.util.concurrent.ThreadContext; -import org.opensearch.core.xcontent.NamedXContentRegistry; -import org.opensearch.env.Environment; -import org.opensearch.index.analysis.AnalysisRegistry; -import org.opensearch.plugins.SearchPipelinePlugin; -import org.opensearch.script.ScriptService; -import org.opensearch.threadpool.Scheduler; - import java.util.Map; -import java.util.function.BiFunction; -import java.util.function.Consumer; -import java.util.function.LongSupplier; /** * A processor implementation may modify the request or response from a search call. @@ -67,87 +55,36 @@ interface Factory { * @param tag The tag for the processor * @param description A short description of what this processor does * @param config The configuration for the processor - * * Note: Implementations are responsible for removing the used configuration * keys, so that after creation the config map should be empty. + * @param pipelineContext Contextual information about the enclosing pipeline. */ - T create(Map> processorFactories, String tag, String description, Map config) throws Exception; + T create( + Map> processorFactories, + String tag, + String description, + Map config, + PipelineContext pipelineContext + ) throws Exception; } /** - * Infrastructure class that holds services that can be used by processor factories to create processor instances - * and that gets passed around to all {@link SearchPipelinePlugin}s. + * Contextual information about the enclosing pipeline. A processor factory may change processor initialization behavior or + * pass this information to the created processor instance. */ - class Parameters { - - /** - * Useful to provide access to the node's environment like config directory to processor factories. - */ - public final Environment env; - - /** - * Provides processors script support. - */ - public final ScriptService scriptService; - - /** - * Provide analyzer support - */ - public final AnalysisRegistry analysisRegistry; - - /** - * Allows processors to read headers set by {@link org.opensearch.action.support.ActionFilter} - * instances that have run while handling the current search. - */ - public final ThreadContext threadContext; - - public final LongSupplier relativeTimeSupplier; - - public final SearchPipelineService searchPipelineService; - - public final Consumer genericExecutor; + class PipelineContext { + private final PipelineSource pipelineSource; - public final NamedXContentRegistry namedXContentRegistry; - - /** - * Provides scheduler support - */ - public final BiFunction scheduler; - - /** - * Provides access to the node's cluster client - */ - public final Client client; - - public Parameters( - Environment env, - ScriptService scriptService, - AnalysisRegistry analysisRegistry, - ThreadContext threadContext, - LongSupplier relativeTimeSupplier, - BiFunction scheduler, - SearchPipelineService searchPipelineService, - Client client, - Consumer genericExecutor, - NamedXContentRegistry namedXContentRegistry - ) { - this.env = env; - this.scriptService = scriptService; - this.threadContext = threadContext; - this.analysisRegistry = analysisRegistry; - this.relativeTimeSupplier = relativeTimeSupplier; - this.scheduler = scheduler; - this.searchPipelineService = searchPipelineService; - this.client = client; - this.genericExecutor = genericExecutor; - this.namedXContentRegistry = namedXContentRegistry; + public PipelineContext(PipelineSource pipelineSource) { + this.pipelineSource = pipelineSource; } + public PipelineSource getPipelineSource() { + return pipelineSource; + } } /** - * Passed via the "pipeline_source" configuration to a processor factory to convey the context for pipeline creation. - *

* A processor factory may change the processor initialization behavior based on the creation context (e.g. avoiding * creating expensive resources during validation or in a request-scoped pipeline.) */ diff --git a/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java b/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java index 70dc8546a077f..83a7a0564467e 100644 --- a/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java +++ b/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java @@ -103,7 +103,7 @@ public SearchPipelineService( this.scriptService = scriptService; this.threadPool = threadPool; this.namedWriteableRegistry = namedWriteableRegistry; - Processor.Parameters parameters = new Processor.Parameters( + SearchPipelinePlugin.Parameters parameters = new SearchPipelinePlugin.Parameters( env, scriptService, analysisRegistry, @@ -189,7 +189,8 @@ void innerUpdatePipelines(SearchPipelineMetadata newSearchPipelineMetadata) { phaseInjectorProcessorFactories, namedWriteableRegistry, totalRequestProcessingMetrics, - totalResponseProcessingMetrics + totalResponseProcessingMetrics, + new Processor.PipelineContext(Processor.PipelineSource.UPDATE_PIPELINE) ); newPipelines.put(newConfiguration.getId(), new PipelineHolder(newConfiguration, newPipeline)); @@ -289,7 +290,8 @@ void validatePipeline(Map searchPipelineInfos phaseInjectorProcessorFactories, namedWriteableRegistry, new OperationMetrics(), // Use ephemeral metrics for validation - new OperationMetrics() + new OperationMetrics(), + new Processor.PipelineContext(Processor.PipelineSource.VALIDATE_PIPELINE) ); List exceptions = new ArrayList<>(); for (SearchRequestProcessor processor : pipeline.getSearchRequestProcessors()) { @@ -388,7 +390,8 @@ public PipelinedRequest resolvePipeline(SearchRequest searchRequest) { phaseInjectorProcessorFactories, namedWriteableRegistry, totalRequestProcessingMetrics, - totalResponseProcessingMetrics + totalResponseProcessingMetrics, + new Processor.PipelineContext(Processor.PipelineSource.SEARCH_REQUEST) ); } catch (Exception e) { throw new SearchPipelineProcessingException(e); diff --git a/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java b/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java index e5bc9df698889..d340d9b8360e8 100644 --- a/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java +++ b/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java @@ -77,12 +77,12 @@ public class SearchPipelineServiceTests extends OpenSearchTestCase { private static final SearchPipelinePlugin DUMMY_PLUGIN = new SearchPipelinePlugin() { @Override - public Map> getRequestProcessors(Processor.Parameters parameters) { - return Map.of("foo", (factories, tag, description, config) -> null); + public Map> getRequestProcessors(Parameters parameters) { + return Map.of("foo", (factories, tag, description, config, ctx) -> null); } - public Map> getResponseProcessors(Processor.Parameters parameters) { - return Map.of("bar", (factories, tag, description, config) -> null); + public Map> getResponseProcessors(Parameters parameters) { + return Map.of("bar", (factories, tag, description, config, ctx) -> null); } @Override @@ -303,7 +303,7 @@ public SearchPhaseName getAfterPhase() { private SearchPipelineService createWithProcessors() { Map> requestProcessors = new HashMap<>(); - requestProcessors.put("scale_request_size", (processorFactories, tag, description, config) -> { + requestProcessors.put("scale_request_size", (processorFactories, tag, description, config, ctx) -> { float scale = ((Number) config.remove("scale")).floatValue(); return new FakeRequestProcessor( "scale_request_size", @@ -313,7 +313,7 @@ private SearchPipelineService createWithProcessors() { ); }); Map> responseProcessors = new HashMap<>(); - responseProcessors.put("fixed_score", (processorFactories, tag, description, config) -> { + responseProcessors.put("fixed_score", (processorFactories, tag, description, config, ctx) -> { float score = ((Number) config.remove("score")).floatValue(); return new FakeResponseProcessor("fixed_score", tag, description, rsp -> rsp.getHits().forEach(h -> h.score(score))); }); @@ -354,12 +354,12 @@ private SearchPipelineService createWithProcessors( this.writableRegistry(), Collections.singletonList(new SearchPipelinePlugin() { @Override - public Map> getRequestProcessors(Processor.Parameters parameters) { + public Map> getRequestProcessors(Parameters parameters) { return requestProcessors; } @Override - public Map> getResponseProcessors(Processor.Parameters parameters) { + public Map> getResponseProcessors(Parameters parameters) { return responseProcessors; } @@ -897,10 +897,9 @@ public void testInfo() { } public void testExceptionOnPipelineCreation() { - Map> badFactory = Map.of( - "bad_factory", - (pf, t, f, c) -> { throw new RuntimeException(); } - ); + Map> badFactory = Map.of("bad_factory", (pf, t, f, c, ctx) -> { + throw new RuntimeException(); + }); SearchPipelineService searchPipelineService = createWithProcessors(badFactory, Collections.emptyMap(), Collections.emptyMap()); Map pipelineSourceMap = new HashMap<>(); @@ -920,7 +919,7 @@ public void testExceptionOnRequestProcessing() { }); Map> throwingRequestProcessorFactory = Map.of( "throwing_request", - (pf, t, f, c) -> throwingRequestProcessor + (pf, t, f, c, ctx) -> throwingRequestProcessor ); SearchPipelineService searchPipelineService = createWithProcessors( @@ -945,7 +944,7 @@ public void testExceptionOnResponseProcessing() throws Exception { }); Map> throwingResponseProcessorFactory = Map.of( "throwing_response", - (pf, t, f, c) -> throwingResponseProcessor + (pf, t, f, c, ctx) -> throwingResponseProcessor ); SearchPipelineService searchPipelineService = createWithProcessors( @@ -973,18 +972,18 @@ public void testStats() throws Exception { }); Map> requestProcessors = Map.of( "successful_request", - (pf, t, f, c) -> new FakeRequestProcessor("successful_request", "2", null, r -> {}), + (pf, t, f, c, ctx) -> new FakeRequestProcessor("successful_request", "2", null, r -> {}), "throwing_request", - (pf, t, f, c) -> throwingRequestProcessor + (pf, t, f, c, ctx) -> throwingRequestProcessor ); SearchResponseProcessor throwingResponseProcessor = new FakeResponseProcessor("throwing_response", "3", null, r -> { throw new RuntimeException(); }); Map> responseProcessors = Map.of( "successful_response", - (pf, t, f, c) -> new FakeResponseProcessor("successful_response", "4", null, r -> {}), + (pf, t, f, c, ctx) -> new FakeResponseProcessor("successful_response", "4", null, r -> {}), "throwing_response", - (pf, t, f, c) -> throwingResponseProcessor + (pf, t, f, c, ctx) -> throwingResponseProcessor ); SearchPipelineService searchPipelineService = createWithProcessors(requestProcessors, responseProcessors, Collections.emptyMap()); @@ -1091,8 +1090,8 @@ private static void assertPipelineStats(OperationStats stats, long count, long f public void testAdHocRejectingProcessor() { String processorType = "ad_hoc_rejecting"; - Map> requestProcessorFactories = Map.of(processorType, (pf, t, d, c) -> { - if (c.get(Processor.PIPELINE_SOURCE) == Processor.PipelineSource.SEARCH_REQUEST) { + Map> requestProcessorFactories = Map.of(processorType, (pf, t, d, c, ctx) -> { + if (ctx.getPipelineSource() == Processor.PipelineSource.SEARCH_REQUEST) { throw new IllegalArgumentException(processorType + " cannot be created as part of a pipeline defined in a search request"); } return new FakeRequestProcessor(processorType, t, d, r -> {});