honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INGEST: Allow Repeated Invocation of Pipeline (#33419) 

* Allows repeated, non-recursive invocation
of the same pipeline
This commit is contained in:
Armin Braun 2018-09-05 22:04:53 +02:00 committed by GitHub
parent 50e07dd413
commit ef1066d7f8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 23 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/PipelineProcessorTests.java
View File

@ -113,4 +113,20 @@ public class PipelineProcessorTests extends ESTestCase {
"Recursive invocation of pipeline [inner] detected.", e.getRootCause().getMessage()
);
}
public void testAllowsRepeatedPipelineInvocations() throws Exception {
String innerPipelineId = "inner";
IngestService ingestService = mock(IngestService.class);
IngestDocument testIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
Map<String, Object> outerConfig = new HashMap<>();
outerConfig.put("pipeline", innerPipelineId);
PipelineProcessor.Factory factory = new PipelineProcessor.Factory(ingestService);
Pipeline inner = new Pipeline(
innerPipelineId, null, null, new CompoundProcessor()
);
when(ingestService.getPipeline(innerPipelineId)).thenReturn(inner);
Processor outerProc = factory.create(Collections.emptyMap(), null, outerConfig);
outerProc.execute(testIngestDocument);
outerProc.execute(testIngestDocument);
}
}

10
server/src/main/java/org/elasticsearch/ingest/IngestDocument.java
View File

@ -645,10 +645,14 @@ public final class IngestDocument {
* @throws Exception On exception in pipeline execution
*/
public IngestDocument executePipeline(Pipeline pipeline) throws Exception {
if (this.executedPipelines.add(pipeline) == false) {
throw new IllegalStateException("Recursive invocation of pipeline [" + pipeline.getId() + "] detected.");
try {
if (this.executedPipelines.add(pipeline) == false) {
throw new IllegalStateException("Recursive invocation of pipeline [" + pipeline.getId() + "] detected.");
}
return pipeline.execute(this);
} finally {
executedPipelines.remove(pipeline);
}
return pipeline.execute(this);
}
@Override