Ingest: Remove generics from Processor.Factory

The factory for ingest processor is generic, but that is only for the
return type of the create mehtod. However, the actual consumer of the
factories only cares about Processor, so generics are not needed.

This change removes the generic type from the factory. It also removes
AbstractProcessorFactory which only existed in order pull the optional
tag from config. This functionality is moved to the caller of the
factories in ConfigurationUtil, and the create method now takes the tag.
This allows the covariant return of the implementation to work with
tests not needing casts.

core/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java

@@ -24,7 +24,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
+import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
 
 import java.io.IOException;
@@ -91,7 +91,7 @@ public class SimulateProcessorResult implements Writeable, ToXContent {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         if (processorTag != null) {
-            builder.field(AbstractProcessorFactory.TAG_KEY, processorTag);
+            builder.field(ConfigurationUtils.TAG_KEY, processorTag);
         }
         if (failure == null) {
             ingestDocument.toXContent(builder, params);

core/src/main/java/org/elasticsearch/ingest/AbstractProcessorFactory.java

@@ -1,38 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.elasticsearch.ingest;
-
-import java.util.Map;
-
-/**
- * A processor implementation may modify the data belonging to a document.
- * Whether changes are made and what exactly is modified is up to the implementation.
- */
-public abstract class AbstractProcessorFactory<P extends Processor> implements Processor.Factory<P> {
-    public static final String TAG_KEY = "tag";
-
-    @Override
-    public P create(Map<String, Object> config) throws Exception {
-        String tag = ConfigurationUtils.readOptionalStringProperty(null, null, config, TAG_KEY);
-        return doCreate(tag, config);
-    }
-
-    protected abstract P doCreate(String tag, Map<String, Object> config) throws Exception;
-}

core/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java

@@ -29,6 +29,8 @@ import java.util.Map;
 
 public final class ConfigurationUtils {
 
+    public static final String TAG_KEY = "tag";
+
     private ConfigurationUtils() {
     }
 
@@ -255,8 +257,8 @@ public final class ConfigurationUtils {
                 ConfigurationUtils.readOptionalList(null, null, config, Pipeline.ON_FAILURE_KEY);
 
             List<Processor> onFailureProcessors = readProcessorConfigs(onFailureProcessorConfigs, processorRegistry);
-            Processor processor;
+            String tag = ConfigurationUtils.readOptionalStringProperty(null, null, config, TAG_KEY);
-            processor = factory.create(config);
+            Processor processor = factory.create(tag, config);
 
             if (onFailureProcessorConfigs != null && onFailureProcessors.isEmpty()) {
                 throw newConfigurationException(processor.getType(), processor.getTag(), Pipeline.ON_FAILURE_KEY,

core/src/main/java/org/elasticsearch/ingest/Processor.java

@@ -45,14 +45,17 @@ public interface Processor {
     /**
      * A factory that knows how to construct a processor based on a map of maps.
      */
-    interface Factory<P extends Processor> {
+    interface Factory {
 
         /**
          * Creates a processor based on the specified map of maps config.
          *
+         * @param tag The tag for the processor
+         * @param config Configuration for the processor to create
+         *
          * Implementations are responsible for removing the used keys, so that after creating a pipeline ingest can
          * verify if all configurations settings have been used.
          */
-        P create(Map<String, Object> config) throws Exception;
+        Processor create(String tag, Map<String, Object> config) throws Exception;
     }
 }

core/src/main/java/org/elasticsearch/ingest/ProcessorsRegistry.java

@@ -37,12 +37,12 @@ public final class ProcessorsRegistry {
     private final ClusterService clusterService;
 
     private ProcessorsRegistry(ScriptService scriptService, ClusterService clusterService,
-                               Map<String, Function<ProcessorsRegistry, Processor.Factory<?>>> providers) {
+                               Map<String, Function<ProcessorsRegistry, Processor.Factory>> providers) {
         this.templateService = new InternalTemplateService(scriptService);
         this.scriptService = scriptService;
         this.clusterService = clusterService;
         Map<String, Processor.Factory> processorFactories = new HashMap<>();
-        for (Map.Entry<String, Function<ProcessorsRegistry, Processor.Factory<?>>> entry : providers.entrySet()) {
+        for (Map.Entry<String, Function<ProcessorsRegistry, Processor.Factory>> entry : providers.entrySet()) {
             processorFactories.put(entry.getKey(), entry.getValue().apply(this));
         }
         this.processorFactories = Collections.unmodifiableMap(processorFactories);
@@ -71,13 +71,13 @@ public final class ProcessorsRegistry {
 
     public static final class Builder {
 
-        private final Map<String, Function<ProcessorsRegistry, Processor.Factory<?>>> providers = new HashMap<>();
+        private final Map<String, Function<ProcessorsRegistry, Processor.Factory>> providers = new HashMap<>();
 
         /**
          * Adds a processor factory under a specific name.
          */
-        public void registerProcessor(String name, Function<ProcessorsRegistry, Processor.Factory<?>> provider) {
+        public void registerProcessor(String name, Function<ProcessorsRegistry, Processor.Factory> provider) {
-            Function<ProcessorsRegistry, Processor.Factory<?>> previous = this.providers.putIfAbsent(name, provider);
+            Function<ProcessorsRegistry, Processor.Factory> previous = this.providers.putIfAbsent(name, provider);
             if (previous != null) {
                 throw new IllegalArgumentException("Processor factory already registered for name [" + name + "]");
             }

core/src/main/java/org/elasticsearch/node/NodeModule.java

@@ -62,7 +62,7 @@ public class NodeModule extends AbstractModule {
     /**
      * Adds a processor factory under a specific type name.
      */
-    public void registerProcessor(String type, Function<ProcessorsRegistry, Processor.Factory<?>> provider) {
+    public void registerProcessor(String type, Function<ProcessorsRegistry, Processor.Factory> provider) {
         processorsRegistryBuilder.registerProcessor(type, provider);
     }
 }

core/src/test/java/org/elasticsearch/ingest/ConfigurationUtilsTests.java

@@ -98,7 +98,7 @@ public class ConfigurationUtilsTests extends ESTestCase {
     public void testReadProcessors() throws Exception {
         Processor processor = mock(Processor.class);
         ProcessorsRegistry.Builder builder = new ProcessorsRegistry.Builder();
-        builder.registerProcessor("test_processor", (registry) -> config -> processor);
+        builder.registerProcessor("test_processor", (registry) -> (tag, config) -> processor);
         ProcessorsRegistry registry = builder.build(mock(ScriptService.class), mock(ClusterService.class));
 
 

core/src/test/java/org/elasticsearch/ingest/PipelineFactoryTests.java

@@ -40,7 +40,7 @@ public class PipelineFactoryTests extends ESTestCase {
     public void testCreate() throws Exception {
         Map<String, Object> processorConfig0 = new HashMap<>();
         Map<String, Object> processorConfig1 = new HashMap<>();
-        processorConfig0.put(AbstractProcessorFactory.TAG_KEY, "first-processor");
+        processorConfig0.put(ConfigurationUtils.TAG_KEY, "first-processor");
         Map<String, Object> pipelineConfig = new HashMap<>();
         pipelineConfig.put(Pipeline.DESCRIPTION_KEY, "_description");
         pipelineConfig.put(Pipeline.PROCESSORS_KEY,

core/src/test/java/org/elasticsearch/ingest/PipelineStoreTests.java

@@ -58,7 +58,7 @@ public class PipelineStoreTests extends ESTestCase {
     public void init() throws Exception {
         store = new PipelineStore(Settings.EMPTY);
         ProcessorsRegistry.Builder registryBuilder = new ProcessorsRegistry.Builder();
-        registryBuilder.registerProcessor("set", (registry) -> config -> {
+        registryBuilder.registerProcessor("set", (registry) -> (tag, config) -> {
             String field = (String) config.remove("field");
             String value = (String) config.remove("value");
             return new Processor() {
@@ -78,7 +78,7 @@ public class PipelineStoreTests extends ESTestCase {
                 }
             };
         });
-        registryBuilder.registerProcessor("remove", (registry) -> config -> {
+        registryBuilder.registerProcessor("remove", (registry) -> (tag, config) -> {
             String field = (String) config.remove("field");
             return new Processor() {
                 @Override

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AbstractStringProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.Map;
 
@@ -53,7 +53,7 @@ abstract class AbstractStringProcessor extends AbstractProcessor {
 
     protected abstract String process(String value);
 
-    static abstract class Factory<T extends AbstractStringProcessor> extends AbstractProcessorFactory<T> {
+    static abstract class Factory implements Processor.Factory {
         protected final String processorType;
 
         protected Factory(String processorType) {
@@ -61,11 +61,11 @@ abstract class AbstractStringProcessor extends AbstractProcessor {
         }
 
         @Override
-        public T doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public AbstractStringProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(processorType, processorTag, config, "field");
             return newProcessor(processorTag, field);
         }
 
-        protected abstract T newProcessor(String processorTag, String field);
+        protected abstract AbstractStringProcessor newProcessor(String processorTag, String field);
     }
 }

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AppendProcessor.java

@@ -20,9 +20,10 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
+import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.TemplateService;
 import org.elasticsearch.ingest.ValueSource;
 
@@ -64,7 +65,7 @@ public final class AppendProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<AppendProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final TemplateService templateService;
 
@@ -73,7 +74,7 @@ public final class AppendProcessor extends AbstractProcessor {
         }
 
         @Override
-        public AppendProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public AppendProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
             return new AppendProcessor(processorTag, templateService.compile(field), ValueSource.wrap(value, templateService));

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ConvertProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -160,9 +160,9 @@ public final class ConvertProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<ConvertProcessor> {
+    public static final class Factory implements Processor.Factory {
         @Override
-        public ConvertProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public ConvertProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String typeProperty = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "type");
             String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field", field);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateIndexNameProcessor.java

@@ -21,9 +21,9 @@ package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormat;
@@ -120,10 +120,10 @@ public final class DateIndexNameProcessor extends AbstractProcessor {
         return dateFormats;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<DateIndexNameProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         @Override
-        protected DateIndexNameProcessor doCreate(String tag, Map<String, Object> config) throws Exception {
+        public DateIndexNameProcessor create(String tag, Map<String, Object> config) throws Exception {
             String localeString = ConfigurationUtils.readOptionalStringProperty(TYPE, tag, config, "locale");
             String timezoneString = ConfigurationUtils.readOptionalStringProperty(TYPE, tag, config, "timezone");
             DateTimeZone timezone = timezoneString == null ? DateTimeZone.UTC : DateTimeZone.forID(timezoneString);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateProcessor.java

@@ -21,9 +21,9 @@ package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.ISODateTimeFormat;
@@ -108,10 +108,10 @@ public final class DateProcessor extends AbstractProcessor {
         return formats;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<DateProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         @SuppressWarnings("unchecked")
-        public DateProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public DateProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field", DEFAULT_TARGET_FIELD);
             String timezoneString = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "timezone");

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/FailProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.TemplateService;
 
 import java.util.Map;
@@ -56,7 +56,7 @@ public final class FailProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<FailProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final TemplateService templateService;
 
@@ -65,7 +65,7 @@ public final class FailProcessor extends AbstractProcessor {
         }
 
         @Override
-        public FailProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public FailProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String message = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "message");
             return new FailProcessor(processorTag, templateService.compile(message));
         }

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ForEachProcessor.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
@@ -83,7 +82,7 @@ public final class ForEachProcessor extends AbstractProcessor {
         return processors;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<ForEachProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final ProcessorsRegistry processorRegistry;
 
@@ -92,7 +91,7 @@ public final class ForEachProcessor extends AbstractProcessor {
         }
 
         @Override
-        protected ForEachProcessor doCreate(String tag, Map<String, Object> config) throws Exception {
+        public ForEachProcessor create(String tag, Map<String, Object> config) throws Exception {
             String field = readStringProperty(TYPE, tag, config, "field");
             List<Map<String, Map<String, Object>>> processorConfigs = readList(TYPE, tag, config, "processors");
             List<Processor> processors = ConfigurationUtils.readProcessorConfigs(processorConfigs, processorRegistry);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.HashMap;
 import java.util.List;
@@ -114,7 +114,7 @@ public final class GrokProcessor extends AbstractProcessor {
         return combinedPattern;
     }
 
-    public final static class Factory extends AbstractProcessorFactory<GrokProcessor> {
+    public final static class Factory implements Processor.Factory {
 
         private final Map<String, String> builtinPatterns;
 
@@ -123,7 +123,7 @@ public final class GrokProcessor extends AbstractProcessor {
         }
 
         @Override
-        public GrokProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public GrokProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String matchField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             List<String> matchPatterns = ConfigurationUtils.readList(TYPE, processorTag, config, "patterns");
             boolean traceMatch = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "trace_match", false);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GsubProcessor.java

@@ -20,8 +20,8 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -78,9 +78,9 @@ public final class GsubProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<GsubProcessor> {
+    public static final class Factory implements Processor.Factory {
         @Override
-        public GsubProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public GsubProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = readStringProperty(TYPE, processorTag, config, "field");
             String pattern = readStringProperty(TYPE, processorTag, config, "pattern");
             String replacement = readStringProperty(TYPE, processorTag, config, "replacement");

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JoinProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.List;
 import java.util.Map;
@@ -70,9 +70,9 @@ public final class JoinProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public final static class Factory extends AbstractProcessorFactory<JoinProcessor> {
+    public final static class Factory implements Processor.Factory {
         @Override
-        public JoinProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public JoinProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String separator = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "separator");
             return new JoinProcessor(processorTag, field, separator);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/LowercaseProcessor.java

@@ -44,7 +44,7 @@ public final class LowercaseProcessor extends AbstractStringProcessor {
         return TYPE;
     }
 
-    public final static class Factory extends AbstractStringProcessor.Factory<LowercaseProcessor> {
+    public final static class Factory extends AbstractStringProcessor.Factory {
 
         public Factory() {
             super(TYPE);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RemoveProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.TemplateService;
 
 import java.util.Map;
@@ -55,7 +55,7 @@ public final class RemoveProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<RemoveProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final TemplateService templateService;
 
@@ -64,7 +64,7 @@ public final class RemoveProcessor extends AbstractProcessor {
         }
 
         @Override
-        public RemoveProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public RemoveProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             return new RemoveProcessor(processorTag, templateService.compile(field));
         }

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RenameProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.Map;
 
@@ -75,9 +75,9 @@ public final class RenameProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<RenameProcessor> {
+    public static final class Factory implements Processor.Factory {
         @Override
-        public RenameProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public RenameProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field");
             return new RenameProcessor(processorTag, field, targetField);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java

@@ -24,8 +24,8 @@ import java.util.Map;
 
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.script.CompiledScript;
 import org.elasticsearch.script.ExecutableScript;
 import org.elasticsearch.script.Script;
@@ -77,7 +77,7 @@ public final class ScriptProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<ScriptProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final ScriptService scriptService;
 
@@ -86,7 +86,7 @@ public final class ScriptProcessor extends AbstractProcessor {
         }
 
         @Override
-        public ScriptProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public ScriptProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = readOptionalStringProperty(TYPE, processorTag, config, "field");
             String lang = readStringProperty(TYPE, processorTag, config, "lang");
             String inline = readOptionalStringProperty(TYPE, processorTag, config, "inline");

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SetProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.TemplateService;
 import org.elasticsearch.ingest.ValueSource;
 
@@ -75,7 +75,7 @@ public final class SetProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<SetProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         private final TemplateService templateService;
 
@@ -84,7 +84,7 @@ public final class SetProcessor extends AbstractProcessor {
         }
 
         @Override
-        public SetProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public SetProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
             boolean overrideEnabled = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "override", true);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SortProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.Collections;
 import java.util.List;
@@ -111,10 +111,10 @@ public final class SortProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public final static class Factory extends AbstractProcessorFactory<SortProcessor> {
+    public final static class Factory implements Processor.Factory {
 
         @Override
-        public SortProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public SortProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, FIELD);
             try {
                 SortOrder direction = SortOrder.fromString(

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SplitProcessor.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -72,9 +72,9 @@ public final class SplitProcessor extends AbstractProcessor {
         return TYPE;
     }
 
-    public static class Factory extends AbstractProcessorFactory<SplitProcessor> {
+    public static class Factory implements Processor.Factory {
         @Override
-        public SplitProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public SplitProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             return new SplitProcessor(processorTag, field, ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "separator"));
         }

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/TrimProcessor.java

@@ -41,7 +41,7 @@ public final class TrimProcessor extends AbstractStringProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractStringProcessor.Factory<TrimProcessor> {
+    public static final class Factory extends AbstractStringProcessor.Factory {
 
         public Factory() {
             super(TYPE);

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/UppercaseProcessor.java

@@ -43,7 +43,7 @@ public final class UppercaseProcessor extends AbstractStringProcessor {
         return TYPE;
     }
 
-    public static final class Factory extends AbstractStringProcessor.Factory<UppercaseProcessor> {
+    public static final class Factory extends AbstractStringProcessor.Factory {
 
         public Factory() {
             super(TYPE);

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/AppendProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -52,8 +51,7 @@ public class AppendProcessorFactoryTests extends ESTestCase {
         }
         config.put("value", value);
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        AppendProcessor appendProcessor = factory.create(processorTag, config);
-        AppendProcessor appendProcessor = factory.create(config);
         assertThat(appendProcessor.getTag(), equalTo(processorTag));
         assertThat(appendProcessor.getField().execute(Collections.emptyMap()), equalTo("field1"));
         assertThat(appendProcessor.getValue().copyAndResolve(Collections.emptyMap()), equalTo(value));
@@ -63,7 +61,7 @@ public class AppendProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("value", "value1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -74,7 +72,7 @@ public class AppendProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[value] required property is missing"));
@@ -86,7 +84,7 @@ public class AppendProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("value", null);
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[value] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ConvertProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 import org.hamcrest.Matchers;
 
@@ -39,8 +38,7 @@ public class ConvertProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("type", type.toString());
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        ConvertProcessor convertProcessor = factory.create(processorTag, config);
-        ConvertProcessor convertProcessor = factory.create(config);
         assertThat(convertProcessor.getTag(), equalTo(processorTag));
         assertThat(convertProcessor.getField(), equalTo("field1"));
         assertThat(convertProcessor.getTargetField(), equalTo("field1"));
@@ -54,7 +52,7 @@ public class ConvertProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("type", type);
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), Matchers.equalTo("[type] type [" + type + "] not supported, cannot convert field."));
@@ -70,7 +68,7 @@ public class ConvertProcessorFactoryTests extends ESTestCase {
         String type = "type-" + randomAsciiOfLengthBetween(1, 10);
         config.put("type", type);
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), Matchers.equalTo("[field] required property is missing"));
@@ -82,7 +80,7 @@ public class ConvertProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), Matchers.equalTo("[type] required property is missing"));
@@ -97,8 +95,7 @@ public class ConvertProcessorFactoryTests extends ESTestCase {
         config.put("target_field", "field2");
         config.put("type", type.toString());
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        ConvertProcessor convertProcessor = factory.create(processorTag, config);
-        ConvertProcessor convertProcessor = factory.create(config);
         assertThat(convertProcessor.getTag(), equalTo(processorTag));
         assertThat(convertProcessor.getField(), equalTo("field1"));
         assertThat(convertProcessor.getTargetField(), equalTo("field2"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateIndexNameFactoryTests.java

@@ -36,7 +36,7 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("date_rounding", "y");
 
-        DateIndexNameProcessor processor = factory.create(config);
+        DateIndexNameProcessor processor = factory.create(null, config);
         assertThat(processor.getDateFormats().size(), Matchers.equalTo(1));
         assertThat(processor.getField(), Matchers.equalTo("_field"));
         assertThat(processor.getIndexNamePrefix(), Matchers.equalTo(""));
@@ -53,7 +53,7 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         config.put("date_rounding", "y");
         config.put("date_formats", Arrays.asList("UNIX", "UNIX_MS"));
 
-        DateIndexNameProcessor processor = factory.create(config);
+        DateIndexNameProcessor processor = factory.create(null, config);
         assertThat(processor.getDateFormats().size(), Matchers.equalTo(2));
 
         config = new HashMap<>();
@@ -62,7 +62,7 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         config.put("date_rounding", "y");
         config.put("index_name_format", "yyyyMMdd");
 
-        processor = factory.create(config);
+        processor = factory.create(null, config);
         assertThat(processor.getIndexNameFormat(), Matchers.equalTo("yyyyMMdd"));
 
         config = new HashMap<>();
@@ -71,7 +71,7 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         config.put("date_rounding", "y");
         config.put("timezone", "+02:00");
 
-        processor = factory.create(config);
+        processor = factory.create(null, config);
         assertThat(processor.getTimezone(), Matchers.equalTo(DateTimeZone.forOffsetHours(2)));
 
         config = new HashMap<>();
@@ -79,7 +79,7 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         config.put("index_name_prefix", "_prefix");
         config.put("date_rounding", "y");
 
-        processor = factory.create(config);
+        processor = factory.create(null, config);
         assertThat(processor.getIndexNamePrefix(), Matchers.equalTo("_prefix"));
     }
 
@@ -87,12 +87,12 @@ public class DateIndexNameFactoryTests extends ESTestCase {
         DateIndexNameProcessor.Factory factory = new DateIndexNameProcessor.Factory();
         Map<String, Object> config = new HashMap<>();
         config.put("date_rounding", "y");
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(null, config));
         assertThat(e.getMessage(), Matchers.equalTo("[field] required property is missing"));
 
         config.clear();
         config.put("field", "_field");
-        e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        e = expectThrows(ElasticsearchParseException.class, () -> factory.create(null, config));
         assertThat(e.getMessage(), Matchers.equalTo("[date_rounding] required property is missing"));
     }
 

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 import org.joda.time.DateTimeZone;
 
@@ -42,8 +41,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("field", sourceField);
         config.put("formats", Collections.singletonList("dd/MM/yyyyy"));
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        DateProcessor processor = factory.create(processorTag, config);
-        DateProcessor processor = factory.create(config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo(sourceField));
         assertThat(processor.getTargetField(), equalTo(DateProcessor.DEFAULT_TARGET_FIELD));
@@ -60,7 +58,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("formats", Collections.singletonList("dd/MM/yyyyy"));
 
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("processor creation should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[field] required property is missing"));
@@ -76,7 +74,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("target_field", targetField);
 
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("processor creation should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[formats] required property is missing"));
@@ -92,7 +90,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         Locale locale = randomLocale(random());
         config.put("locale", locale.toLanguageTag());
 
-        DateProcessor processor = factory.create(config);
+        DateProcessor processor = factory.create(null, config);
         assertThat(processor.getLocale().toLanguageTag(), equalTo(locale.toLanguageTag()));
     }
 
@@ -104,7 +102,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("formats", Collections.singletonList("dd/MM/yyyyy"));
         config.put("locale", "invalid_locale");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("should fail with invalid locale");
         } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("Invalid language tag specified: invalid_locale"));
@@ -120,7 +118,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
 
         DateTimeZone timezone = randomDateTimeZone();
         config.put("timezone", timezone.getID());
-        DateProcessor processor = factory.create(config);
+        DateProcessor processor = factory.create(null, config);
         assertThat(processor.getTimezone(), equalTo(timezone));
     }
 
@@ -132,7 +130,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("match_formats", Collections.singletonList("dd/MM/yyyyy"));
         config.put("timezone", "invalid_timezone");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("invalid timezone should fail");
         } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("The datetime zone id 'invalid_timezone' is not recognised"));
@@ -146,7 +144,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("field", sourceField);
         config.put("formats", Arrays.asList("dd/MM/yyyy", "dd-MM-yyyy"));
 
-        DateProcessor processor = factory.create(config);
+        DateProcessor processor = factory.create(null, config);
         assertThat(processor.getFormats(), equalTo(Arrays.asList("dd/MM/yyyy", "dd-MM-yyyy")));
     }
 
@@ -158,7 +156,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("formats", "dd/MM/yyyy");
 
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("processor creation should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[formats] property isn't a list, but of type [java.lang.String]"));
@@ -174,7 +172,7 @@ public class DateProcessorFactoryTests extends ESTestCase {
         config.put("target_field", targetField);
         config.put("formats", Arrays.asList("dd/MM/yyyy", "dd-MM-yyyy"));
 
-        DateProcessor processor = factory.create(config);
+        DateProcessor processor = factory.create(null, config);
         assertThat(processor.getTargetField(), equalTo(targetField));
     }
 }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/FailProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -44,8 +43,7 @@ public class FailProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("message", "error");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        FailProcessor failProcessor = factory.create(processorTag, config);
-        FailProcessor failProcessor = factory.create(config);
         assertThat(failProcessor.getTag(), equalTo(processorTag));
         assertThat(failProcessor.getMessage().execute(Collections.emptyMap()), equalTo("error"));
     }
@@ -53,7 +51,7 @@ public class FailProcessorFactoryTests extends ESTestCase {
     public void testCreateMissingMessageField() throws Exception {
         Map<String, Object> config = new HashMap<>();
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[message] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorFactoryTests.java

@@ -38,14 +38,14 @@ public class ForEachProcessorFactoryTests extends ESTestCase {
     public void testCreate() throws Exception {
         ProcessorsRegistry.Builder builder = new ProcessorsRegistry.Builder();
         Processor processor = new TestProcessor(ingestDocument -> {});
-        builder.registerProcessor("_name", (registry) -> config -> processor);
+        builder.registerProcessor("_name", (registry) -> (tag, config) -> processor);
         ProcessorsRegistry registry = builder.build(mock(ScriptService.class), mock(ClusterService.class));
         ForEachProcessor.Factory forEachFactory = new ForEachProcessor.Factory(registry);
 
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("processors", Collections.singletonList(Collections.singletonMap("_name", Collections.emptyMap())));
-        ForEachProcessor forEachProcessor = forEachFactory.create(config);
+        ForEachProcessor forEachProcessor = forEachFactory.create(null, config);
         assertThat(forEachProcessor, Matchers.notNullValue());
         assertThat(forEachProcessor.getField(), Matchers.equalTo("_field"));
         assertThat(forEachProcessor.getProcessors().size(), Matchers.equalTo(1));
@@ -54,7 +54,7 @@ public class ForEachProcessorFactoryTests extends ESTestCase {
         config = new HashMap<>();
         config.put("processors", Collections.singletonList(Collections.singletonMap("_name", Collections.emptyMap())));
         try {
-            forEachFactory.create(config);
+            forEachFactory.create(null, config);
             fail("exception expected");
         } catch (Exception e) {
             assertThat(e.getMessage(), Matchers.equalTo("[field] required property is missing"));
@@ -63,7 +63,7 @@ public class ForEachProcessorFactoryTests extends ESTestCase {
         config = new HashMap<>();
         config.put("field", "_field");
         try {
-            forEachFactory.create(config);
+            forEachFactory.create(null, config);
             fail("exception expected");
         } catch (Exception e) {
             assertThat(e.getMessage(), Matchers.equalTo("[processors] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Collections;
@@ -39,8 +38,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("patterns", Collections.singletonList("(?<foo>\\w+)"));
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        GrokProcessor processor = factory.create(processorTag, config);
-        GrokProcessor processor = factory.create(config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getMatchField(), equalTo("_field"));
         assertThat(processor.getGrok(), notNullValue());
@@ -50,7 +48,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap());
         Map<String, Object> config = new HashMap<>();
         config.put("patterns", Collections.singletonList("(?<foo>\\w+)"));
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create("tag", config));
         assertThat(e.getMessage(), equalTo("[field] required property is missing"));
     }
 
@@ -58,7 +56,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         GrokProcessor.Factory factory = new GrokProcessor.Factory(Collections.emptyMap());
         Map<String, Object> config = new HashMap<>();
         config.put("field", "foo");
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create("tag", config));
         assertThat(e.getMessage(), equalTo("[patterns] required property is missing"));
     }
 
@@ -67,7 +65,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "foo");
         config.put("patterns", Collections.emptyList());
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create("tag", config));
         assertThat(e.getMessage(), equalTo("[patterns] List of patterns must not be empty"));
     }
 
@@ -78,7 +76,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("patterns", Collections.singletonList("%{MY_PATTERN:name}!"));
         config.put("pattern_definitions", Collections.singletonMap("MY_PATTERN", "foo"));
-        GrokProcessor processor = factory.create(config);
+        GrokProcessor processor = factory.create(null, config);
         assertThat(processor.getMatchField(), equalTo("_field"));
         assertThat(processor.getGrok(), notNullValue());
         assertThat(processor.getGrok().match("foo!"), equalTo(true));
@@ -89,7 +87,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("patterns", Collections.singletonList("["));
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create("tag", config));
         assertThat(e.getMessage(), equalTo("[patterns] Invalid regex pattern found in: [[]. premature end of char-class"));
     }
 
@@ -99,7 +97,7 @@ public class GrokProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("patterns", Collections.singletonList("%{MY_PATTERN:name}!"));
         config.put("pattern_definitions", Collections.singletonMap("MY_PATTERN", "["));
-        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create(config));
+        ElasticsearchParseException e = expectThrows(ElasticsearchParseException.class, () -> factory.create("tag", config));
         assertThat(e.getMessage(),
             equalTo("[patterns] Invalid regex pattern found in: [%{MY_PATTERN:name}!]. premature end of char-class"));
     }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GsubProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -38,8 +37,7 @@ public class GsubProcessorFactoryTests extends ESTestCase {
         config.put("pattern", "\\.");
         config.put("replacement", "-");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        GsubProcessor gsubProcessor = factory.create(processorTag, config);
-        GsubProcessor gsubProcessor = factory.create(config);
         assertThat(gsubProcessor.getTag(), equalTo(processorTag));
         assertThat(gsubProcessor.getField(), equalTo("field1"));
         assertThat(gsubProcessor.getPattern().toString(), equalTo("\\."));
@@ -52,7 +50,7 @@ public class GsubProcessorFactoryTests extends ESTestCase {
         config.put("pattern", "\\.");
         config.put("replacement", "-");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -65,7 +63,7 @@ public class GsubProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("replacement", "-");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[pattern] required property is missing"));
@@ -78,7 +76,7 @@ public class GsubProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("pattern", "\\.");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[replacement] required property is missing"));
@@ -92,7 +90,7 @@ public class GsubProcessorFactoryTests extends ESTestCase {
         config.put("pattern", "[");
         config.put("replacement", "-");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[pattern] Invalid regex pattern. Unclosed character class"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/JoinProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -36,8 +35,7 @@ public class JoinProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("separator", "-");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        JoinProcessor joinProcessor = factory.create(processorTag, config);
-        JoinProcessor joinProcessor = factory.create(config);
         assertThat(joinProcessor.getTag(), equalTo(processorTag));
         assertThat(joinProcessor.getField(), equalTo("field1"));
         assertThat(joinProcessor.getSeparator(), equalTo("-"));
@@ -48,7 +46,7 @@ public class JoinProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("separator", "-");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -60,7 +58,7 @@ public class JoinProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[separator] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/LowercaseProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -35,8 +34,7 @@ public class LowercaseProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        LowercaseProcessor uppercaseProcessor = (LowercaseProcessor)factory.create(processorTag, config);
-        LowercaseProcessor uppercaseProcessor = factory.create(config);
         assertThat(uppercaseProcessor.getTag(), equalTo(processorTag));
         assertThat(uppercaseProcessor.getField(), equalTo("field1"));
     }
@@ -45,7 +43,7 @@ public class LowercaseProcessorFactoryTests extends ESTestCase {
         LowercaseProcessor.Factory factory = new LowercaseProcessor.Factory();
         Map<String, Object> config = new HashMap<>();
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RemoveProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -44,8 +43,7 @@ public class RemoveProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        RemoveProcessor removeProcessor = factory.create(processorTag, config);
-        RemoveProcessor removeProcessor = factory.create(config);
         assertThat(removeProcessor.getTag(), equalTo(processorTag));
         assertThat(removeProcessor.getField().execute(Collections.emptyMap()), equalTo("field1"));
     }
@@ -53,7 +51,7 @@ public class RemoveProcessorFactoryTests extends ESTestCase {
     public void testCreateMissingField() throws Exception {
         Map<String, Object> config = new HashMap<>();
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RenameProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -36,8 +35,7 @@ public class RenameProcessorFactoryTests extends ESTestCase {
         config.put("field", "old_field");
         config.put("target_field", "new_field");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        RenameProcessor renameProcessor = factory.create(processorTag, config);
-        RenameProcessor renameProcessor = factory.create(config);
         assertThat(renameProcessor.getTag(), equalTo(processorTag));
         assertThat(renameProcessor.getField(), equalTo("old_field"));
         assertThat(renameProcessor.getTargetField(), equalTo("new_field"));
@@ -48,7 +46,7 @@ public class RenameProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("target_field", "new_field");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -60,7 +58,7 @@ public class RenameProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "old_field");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[target_field] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ScriptProcessorFactoryTests.java

@@ -55,7 +55,7 @@ public class ScriptProcessorFactoryTests extends ESTestCase {
         configMap.put("lang", "mockscript");
 
         ElasticsearchException exception = expectThrows(ElasticsearchException.class,
-            () -> factory.doCreate(randomAsciiOfLength(10), configMap));
+            () -> factory.create(randomAsciiOfLength(10), configMap));
 
         assertThat(exception.getMessage(), is("[null] Only one of [file], [id], or [inline] may be configured"));
     }
@@ -66,7 +66,7 @@ public class ScriptProcessorFactoryTests extends ESTestCase {
         configMap.put("lang", "mockscript");
 
         ElasticsearchException exception = expectThrows(ElasticsearchException.class,
-            () -> factory.doCreate(randomAsciiOfLength(10), configMap));
+            () -> factory.create(randomAsciiOfLength(10), configMap));
 
         assertThat(exception.getMessage(), is("[null] Need [file], [id], or [inline] parameter to refer to scripts"));
     }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SetProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -45,8 +44,7 @@ public class SetProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("value", "value1");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        SetProcessor setProcessor = factory.create(processorTag, config);
-        SetProcessor setProcessor = factory.create(config);
         assertThat(setProcessor.getTag(), equalTo(processorTag));
         assertThat(setProcessor.getField().execute(Collections.emptyMap()), equalTo("field1"));
         assertThat(setProcessor.getValue().copyAndResolve(Collections.emptyMap()), equalTo("value1"));
@@ -60,8 +58,7 @@ public class SetProcessorFactoryTests extends ESTestCase {
         config.put("value", "value1");
         config.put("override", overrideEnabled);
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        SetProcessor setProcessor = factory.create(processorTag, config);
-        SetProcessor setProcessor = factory.create(config);
         assertThat(setProcessor.getTag(), equalTo(processorTag));
         assertThat(setProcessor.getField().execute(Collections.emptyMap()), equalTo("field1"));
         assertThat(setProcessor.getValue().copyAndResolve(Collections.emptyMap()), equalTo("value1"));
@@ -72,7 +69,7 @@ public class SetProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("value", "value1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -83,7 +80,7 @@ public class SetProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[value] required property is missing"));
@@ -95,7 +92,7 @@ public class SetProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("value", null);
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[value] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -36,8 +35,7 @@ public class SplitProcessorFactoryTests extends ESTestCase {
         config.put("field", "field1");
         config.put("separator", "\\.");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        SplitProcessor splitProcessor = factory.create(processorTag, config);
-        SplitProcessor splitProcessor = factory.create(config);
         assertThat(splitProcessor.getTag(), equalTo(processorTag));
         assertThat(splitProcessor.getField(), equalTo("field1"));
         assertThat(splitProcessor.getSeparator(), equalTo("\\."));
@@ -48,7 +46,7 @@ public class SplitProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("separator", "\\.");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));
@@ -60,7 +58,7 @@ public class SplitProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[separator] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorTests.java

@@ -84,7 +84,7 @@ public class SplitProcessorTests extends ESTestCase {
         Map<String, Object> splitConfig = new HashMap<>();
         splitConfig.put("field", "flags");
         splitConfig.put("separator", "\\|");
-        Processor splitProcessor = (new SplitProcessor.Factory()).create(splitConfig);
+        Processor splitProcessor = (new SplitProcessor.Factory()).create("tag", splitConfig);
         Map<String, Object> source = new HashMap<>();
         source.put("flags", "new|hot|super|fun|interesting");
         IngestDocument ingestDocument = new IngestDocument(source, new HashMap<>());

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/TrimProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -35,8 +34,7 @@ public class TrimProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        TrimProcessor uppercaseProcessor = (TrimProcessor)factory.create(processorTag, config);
-        TrimProcessor uppercaseProcessor = factory.create(config);
         assertThat(uppercaseProcessor.getTag(), equalTo(processorTag));
         assertThat(uppercaseProcessor.getField(), equalTo("field1"));
     }
@@ -45,7 +43,7 @@ public class TrimProcessorFactoryTests extends ESTestCase {
         TrimProcessor.Factory factory = new TrimProcessor.Factory();
         Map<String, Object> config = new HashMap<>();
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/UppercaseProcessorFactoryTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -35,8 +34,7 @@ public class UppercaseProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "field1");
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        UppercaseProcessor uppercaseProcessor = (UppercaseProcessor)factory.create(processorTag, config);
-        UppercaseProcessor uppercaseProcessor = factory.create(config);
         assertThat(uppercaseProcessor.getTag(), equalTo(processorTag));
         assertThat(uppercaseProcessor.getField(), equalTo("field1"));
     }
@@ -45,7 +43,7 @@ public class UppercaseProcessorFactoryTests extends ESTestCase {
         UppercaseProcessor.Factory factory = new UppercaseProcessor.Factory();
         Map<String, Object> config = new HashMap<>();
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("factory create should have failed");
         } catch(ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[field] required property is missing"));

plugins/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java

@@ -25,8 +25,8 @@ import org.apache.tika.metadata.TikaCoreProperties;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -150,12 +150,12 @@ public final class AttachmentProcessor extends AbstractProcessor {
         return indexedChars;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<AttachmentProcessor> {
+    public static final class Factory implements Processor.Factory {
 
         static final Set<Property> DEFAULT_PROPERTIES = EnumSet.allOf(Property.class);
 
         @Override
-        public AttachmentProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public AttachmentProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String field = readStringProperty(TYPE, processorTag, config, "field");
             String targetField = readStringProperty(TYPE, processorTag, config, "target_field", "attachment");
             List<String> properyNames = readOptionalList(TYPE, processorTag, config, "properties");

plugins/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorFactoryTests.java

@@ -20,7 +20,7 @@
 package org.elasticsearch.ingest.attachment;
 
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
+import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.ArrayList;
@@ -46,9 +46,8 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
 
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
 
-        AttachmentProcessor processor = factory.create(config);
+        AttachmentProcessor processor = factory.create(processorTag, config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("attachment"));
@@ -62,8 +61,7 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         config.put("indexed_chars", indexedChars);
 
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
+        AttachmentProcessor processor = factory.create(processorTag, config);
-        AttachmentProcessor processor = factory.create(config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getIndexedChars(), is(indexedChars));
     }
@@ -72,7 +70,7 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("target_field", "_field");
-        AttachmentProcessor processor = factory.create(config);
+        AttachmentProcessor processor = factory.create(null, config);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("_field"));
     }
@@ -89,7 +87,7 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("properties", fieldNames);
-        AttachmentProcessor processor = factory.create(config);
+        AttachmentProcessor processor = factory.create(null, config);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getProperties(), equalTo(properties));
     }
@@ -99,7 +97,7 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("properties", Collections.singletonList("invalid"));
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[properties] illegal field option [invalid]"));
@@ -113,7 +111,7 @@ public class AttachmentProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("properties", "invalid");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[properties] property isn't a list, but of type [java.lang.String]"));

plugins/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpProcessor.java

@@ -34,8 +34,8 @@ import org.elasticsearch.SpecialPermission;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.ingest.AbstractProcessor;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -217,7 +217,7 @@ public final class GeoIpProcessor extends AbstractProcessor {
         return geoData;
     }
 
-    public static final class Factory extends AbstractProcessorFactory<GeoIpProcessor> {
+    public static final class Factory implements Processor.Factory {
         static final Set<Property> DEFAULT_CITY_PROPERTIES = EnumSet.of(
             Property.CONTINENT_NAME, Property.COUNTRY_ISO_CODE, Property.REGION_NAME,
             Property.CITY_NAME, Property.LOCATION
@@ -231,7 +231,7 @@ public final class GeoIpProcessor extends AbstractProcessor {
         }
 
         @Override
-        public GeoIpProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public GeoIpProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             String ipField = readStringProperty(TYPE, processorTag, config, "field");
             String targetField = readStringProperty(TYPE, processorTag, config, "target_field", "geoip");
             String databaseFile = readStringProperty(TYPE, processorTag, config, "database_file", "GeoLite2-City.mmdb.gz");

plugins/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpProcessorFactoryTests.java

@@ -23,7 +23,6 @@ import com.carrotsearch.randomizedtesting.generators.RandomPicks;
 import com.maxmind.geoip2.DatabaseReader;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Randomness;
-import org.elasticsearch.ingest.AbstractProcessorFactory;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.StreamsUtils;
 import org.junit.AfterClass;
@@ -74,11 +73,9 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
 
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
-
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
 
-        GeoIpProcessor processor = factory.create(config);
+        GeoIpProcessor processor = factory.create(processorTag, config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("geoip"));
@@ -92,11 +89,9 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("database_file", "GeoLite2-Country.mmdb.gz");
-
         String processorTag = randomAsciiOfLength(10);
-        config.put(AbstractProcessorFactory.TAG_KEY, processorTag);
 
-        GeoIpProcessor processor = factory.create(config);
+        GeoIpProcessor processor = factory.create(processorTag, config);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("geoip"));
@@ -109,7 +104,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("target_field", "_field");
-        GeoIpProcessor processor = factory.create(config);
+        GeoIpProcessor processor = factory.create(null, config);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("_field"));
     }
@@ -119,7 +114,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("database_file", "GeoLite2-Country.mmdb.gz");
-        GeoIpProcessor processor = factory.create(config);
+        GeoIpProcessor processor = factory.create(null, config);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("geoip"));
         assertThat(processor.getDbReader().getMetadata().getDatabaseType(), equalTo("GeoLite2-Country"));
@@ -135,7 +130,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         String cityProperty = RandomPicks.randomFrom(Randomness.get(), cityOnlyProperties).toString();
         config.put("properties", Collections.singletonList(cityProperty));
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("Exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[properties] illegal property value [" + cityProperty +
@@ -150,7 +145,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("database_file", "does-not-exist.mmdb.gz");
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("Exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[database_file] database file [does-not-exist.mmdb.gz] doesn't exist"));
@@ -171,7 +166,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("properties", fieldNames);
-        GeoIpProcessor processor = factory.create(config);
+        GeoIpProcessor processor = factory.create(null, config);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getProperties(), equalTo(properties));
     }
@@ -183,7 +178,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("properties", Collections.singletonList("invalid"));
         try {
-            factory.create(config);
+            factory.create(null, config);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[properties] illegal property value [invalid]. valid values are [IP, COUNTRY_ISO_CODE, " +
@@ -194,7 +189,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
         config.put("field", "_field");
         config.put("properties", "invalid");
         try {
-            factory.create(config);
+            factory.create("tag", config);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[properties] property isn't a list, but of type [java.lang.String]"));

test/framework/src/main/java/org/elasticsearch/ingest/IngestTestPlugin.java

@@ -28,7 +28,7 @@ import org.elasticsearch.plugins.Plugin;
 public class IngestTestPlugin extends Plugin {
 
     public void onModule(NodeModule nodeModule) {
-        nodeModule.registerProcessor("test", (registry) -> config ->
+        nodeModule.registerProcessor("test", (registry) -> (tag, config) ->
                 new TestProcessor("id", "test", doc -> {
                     doc.setFieldValue("processed", true);
                     if (doc.hasField("fail") && doc.getFieldValue("fail", Boolean.class)) {

test/framework/src/main/java/org/elasticsearch/ingest/TestProcessor.java

@@ -64,9 +64,9 @@ public class TestProcessor implements Processor {
         return invokedCounter.get();
     }
 
-    public static final class Factory extends AbstractProcessorFactory<TestProcessor> {
+    public static final class Factory implements Processor.Factory {
         @Override
-        public TestProcessor doCreate(String processorTag, Map<String, Object> config) throws Exception {
+        public TestProcessor create(String processorTag, Map<String, Object> config) throws Exception {
             return new TestProcessor(processorTag, "test-processor", ingestDocument -> {});
         }
     }