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.

52 files changed, 167 insertions, 237 deletions

diff --git a/core/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java b/core/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java
index 6924550d79..ce98767bcc 100644
--- a/core/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java
+++ b/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);
diff --git a/core/src/main/java/org/elasticsearch/ingest/AbstractProcessorFactory.java b/core/src/main/java/org/elasticsearch/ingest/AbstractProcessorFactory.java
deleted file mode 100644
index 0fb5022879..0000000000
--- a/core/src/main/java/org/elasticsearch/ingest/AbstractProcessorFactory.java
+++ /dev/null
@@ -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;
-}
diff --git a/core/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java b/core/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java
index 623c4f315a..54fbc4a023 100644
--- a/core/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java
+++ b/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;
-            processor = factory.create(config);
+            String tag = ConfigurationUtils.readOptionalStringProperty(null, null, config, TAG_KEY);
+            Processor processor = factory.create(tag, config);
 
             if (onFailureProcessorConfigs != null && onFailureProcessors.isEmpty()) {
                 throw newConfigurationException(processor.getType(), processor.getTag(), Pipeline.ON_FAILURE_KEY,
diff --git a/core/src/main/java/org/elasticsearch/ingest/Processor.java b/core/src/main/java/org/elasticsearch/ingest/Processor.java
index 92c18464e9..d4a5462c1c 100644
--- a/core/src/main/java/org/elasticsearch/ingest/Processor.java
+++ b/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;
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/ingest/ProcessorsRegistry.java b/core/src/main/java/org/elasticsearch/ingest/ProcessorsRegistry.java
index dd16898aea..461247b6d0 100644
--- a/core/src/main/java/org/elasticsearch/ingest/ProcessorsRegistry.java
+++ b/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) {
-            Function<ProcessorsRegistry, Processor.Factory<?>> previous = this.providers.putIfAbsent(name, provider);
+        public void registerProcessor(String name, Function<ProcessorsRegistry, Processor.Factory> provider) {
+            Function<ProcessorsRegistry, Processor.Factory> previous = this.providers.putIfAbsent(name, provider);
             if (previous != null) {
                 throw new IllegalArgumentException("Processor factory already registered for name [" + name + "]");
             }
diff --git a/core/src/main/java/org/elasticsearch/node/NodeModule.java b/core/src/main/java/org/elasticsearch/node/NodeModule.java
index 6d8be732dd..49de57a87b 100644
--- a/core/src/main/java/org/elasticsearch/node/NodeModule.java
+++ b/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);
     }
 }
diff --git a/core/src/test/java/org/elasticsearch/ingest/ConfigurationUtilsTests.java b/core/src/test/java/org/elasticsearch/ingest/ConfigurationUtilsTests.java
index fa78d5aa16..82886628c8 100644
--- a/core/src/test/java/org/elasticsearch/ingest/ConfigurationUtilsTests.java
+++ b/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));
 
 
diff --git a/core/src/test/java/org/elasticsearch/ingest/PipelineFactoryTests.java b/core/src/test/java/org/elasticsearch/ingest/PipelineFactoryTests.java
index cb7bd849a4..8fe7ddd84c 100644
--- a/core/src/test/java/org/elasticsearch/ingest/PipelineFactoryTests.java
+++ b/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,
diff --git a/core/src/test/java/org/elasticsearch/ingest/PipelineStoreTests.java b/core/src/test/java/org/elasticsearch/ingest/PipelineStoreTests.java
index 55ea4360ec..ec41eda47b 100644
--- a/core/src/test/java/org/elasticsearch/ingest/PipelineStoreTests.java
+++ b/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
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AbstractStringProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AbstractStringProcessor.java
index d9ffd34cac..f35f3fec53 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AbstractStringProcessor.java
+++ b/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);
     }
 }
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AppendProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AppendProcessor.java
index af163c3c18..26b0e66a63 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/AppendProcessor.java
+++ b/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));
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ConvertProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ConvertProcessor.java
index 015c56c72c..558bdde96a 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ConvertProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateIndexNameProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateIndexNameProcessor.java
index a94d4d048a..c750c84c57 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateIndexNameProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateProcessor.java
index b82b9c8b76..e61ed51311 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/DateProcessor.java
+++ b/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");
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/FailProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/FailProcessor.java
index 6c434d85d5..a24322f556 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/FailProcessor.java
+++ b/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));
         }
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ForEachProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ForEachProcessor.java
index b6d14d1b8c..05be47633b 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ForEachProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java
index aa9fbb905c..32edc665d5 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GrokProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GsubProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GsubProcessor.java
index 72bc9e7671..1dfc566670 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/GsubProcessor.java
+++ b/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");
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JoinProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JoinProcessor.java
index 8114d20f28..f8bc2afffd 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JoinProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/LowercaseProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/LowercaseProcessor.java
index 9f8ea7a561..e7a8f3f3e6 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/LowercaseProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RemoveProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RemoveProcessor.java
index 98c4e18a40..a7de33a7e2 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RemoveProcessor.java
+++ b/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));
         }
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RenameProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RenameProcessor.java
index 9143321c4a..ae81291b64 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RenameProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java
index e488136616..4733c1f586 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java
+++ b/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");
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SetProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SetProcessor.java
index a78701645a..ce328e34cb 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SetProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SortProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SortProcessor.java
index 706a1cef9c..5ab4016921 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SortProcessor.java
+++ b/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(
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SplitProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SplitProcessor.java
index f7c5e8befc..87526f6229 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/SplitProcessor.java
+++ b/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"));
         }
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/TrimProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/TrimProcessor.java
index a57a25125d..e852f887da 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/TrimProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/UppercaseProcessor.java b/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/UppercaseProcessor.java
index a5c817352a..5585a130ea 100644
--- a/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/UppercaseProcessor.java
+++ b/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);
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/AppendProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/AppendProcessorFactoryTests.java
index fbf77cc428..b49a44cc04 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/AppendProcessorFactoryTests.java
+++ b/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(config);
+        AppendProcessor appendProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ConvertProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ConvertProcessorFactoryTests.java
index 1ec5362af1..7dd8bbf97e 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ConvertProcessorFactoryTests.java
+++ b/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(config);
+        ConvertProcessor convertProcessor = factory.create(processorTag, 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(config);
+        ConvertProcessor convertProcessor = factory.create(processorTag, config);
         assertThat(convertProcessor.getTag(), equalTo(processorTag));
         assertThat(convertProcessor.getField(), equalTo("field1"));
         assertThat(convertProcessor.getTargetField(), equalTo("field2"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateIndexNameFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateIndexNameFactoryTests.java
index 42877236b8..cc272d0b12 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateIndexNameFactoryTests.java
+++ b/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"));
     }
 
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateProcessorFactoryTests.java
index 65dcdf6082..9c5352c7ee 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/DateProcessorFactoryTests.java
+++ b/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(config);
+        DateProcessor processor = factory.create(processorTag, 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));
     }
 }
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/FailProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/FailProcessorFactoryTests.java
index db16b78b31..a385a58ef5 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/FailProcessorFactoryTests.java
+++ b/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(config);
+        FailProcessor failProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorFactoryTests.java
index d45e98ab06..0bbad532fb 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorFactoryTests.java
+++ b/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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java
index a7a133b436..1287d06642 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GrokProcessorFactoryTests.java
+++ b/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(config);
+        GrokProcessor processor = factory.create(processorTag, 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"));
     }
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GsubProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GsubProcessorFactoryTests.java
index 60cceb3402..0086457f85 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/GsubProcessorFactoryTests.java
+++ b/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(config);
+        GsubProcessor gsubProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/JoinProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/JoinProcessorFactoryTests.java
index 970fd8b8b9..b2386c17a2 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/JoinProcessorFactoryTests.java
+++ b/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(config);
+        JoinProcessor joinProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/LowercaseProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/LowercaseProcessorFactoryTests.java
index 4dec115458..2e0682beb6 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/LowercaseProcessorFactoryTests.java
+++ b/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 = factory.create(config);
+        LowercaseProcessor uppercaseProcessor = (LowercaseProcessor)factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RemoveProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RemoveProcessorFactoryTests.java
index a5f88103e9..133ddbeebf 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RemoveProcessorFactoryTests.java
+++ b/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(config);
+        RemoveProcessor removeProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RenameProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RenameProcessorFactoryTests.java
index c078f09dd9..b969a3b624 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RenameProcessorFactoryTests.java
+++ b/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(config);
+        RenameProcessor renameProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ScriptProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ScriptProcessorFactoryTests.java
index ed47894d4d..79048cd776 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ScriptProcessorFactoryTests.java
+++ b/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"));
     }
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SetProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SetProcessorFactoryTests.java
index b8c97a379c..cbbf3f4090 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SetProcessorFactoryTests.java
+++ b/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(config);
+        SetProcessor setProcessor = factory.create(processorTag, 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(config);
+        SetProcessor setProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorFactoryTests.java
index c747807b71..4f85d61e62 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorFactoryTests.java
+++ b/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(config);
+        SplitProcessor splitProcessor = factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorTests.java
index 13d45dc126..a7e1313a09 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SplitProcessorTests.java
+++ b/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<>());
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/TrimProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/TrimProcessorFactoryTests.java
index 5490477547..fa9a33f41d 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/TrimProcessorFactoryTests.java
+++ b/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 = factory.create(config);
+        TrimProcessor uppercaseProcessor = (TrimProcessor)factory.create(processorTag, 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"));
diff --git a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/UppercaseProcessorFactoryTests.java b/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/UppercaseProcessorFactoryTests.java
index cd4d1faf76..91698d4fcc 100644
--- a/modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/UppercaseProcessorFactoryTests.java
+++ b/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 = factory.create(config);
+        UppercaseProcessor uppercaseProcessor = (UppercaseProcessor)factory.create(processorTag, 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"));
diff --git a/plugins/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java b/plugins/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java
index 40ebe2592a..fb4dc37d64 100644
--- a/plugins/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java
+++ b/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");
diff --git a/plugins/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorFactoryTests.java b/plugins/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorFactoryTests.java
index 6bd4e07702..8d01105685 100644
--- a/plugins/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorFactoryTests.java
+++ b/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(config);
+        AttachmentProcessor processor = factory.create(processorTag, 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]"));
diff --git a/plugins/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpProcessor.java b/plugins/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpProcessor.java
index 0b18cae25e..92c4785dba 100644
--- a/plugins/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpProcessor.java
+++ b/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");
diff --git a/plugins/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpProcessorFactoryTests.java b/plugins/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpProcessorFactoryTests.java
index 0472916272..28043171fc 100644
--- a/plugins/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpProcessorFactoryTests.java
+++ b/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]"));
diff --git a/test/framework/src/main/java/org/elasticsearch/ingest/IngestTestPlugin.java b/test/framework/src/main/java/org/elasticsearch/ingest/IngestTestPlugin.java
index b32a2eab99..a9585aa8d4 100644
--- a/test/framework/src/main/java/org/elasticsearch/ingest/IngestTestPlugin.java
+++ b/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)) {
diff --git a/test/framework/src/main/java/org/elasticsearch/ingest/TestProcessor.java b/test/framework/src/main/java/org/elasticsearch/ingest/TestProcessor.java
index e36d73a8d9..b309d94fe0 100644
--- a/test/framework/src/main/java/org/elasticsearch/ingest/TestProcessor.java
+++ b/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 -> {});
         }
     }