diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java
index 24c545a..df52d22 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java
@@ -29,12 +29,12 @@
 import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.processors.filters.jvm.processor.booleanfilter.BooleanFilterProcessor;
-import org.apache.streampipes.processors.filters.jvm.processor.compose.ComposeController;
-import org.apache.streampipes.processors.filters.jvm.processor.enrich.MergeByEnrichController;
+import org.apache.streampipes.processors.filters.jvm.processor.compose.ComposeProcessor;
+import org.apache.streampipes.processors.filters.jvm.processor.enrich.MergeByEnrichProcessor;
 import org.apache.streampipes.processors.filters.jvm.processor.limit.RateLimitController;
-import org.apache.streampipes.processors.filters.jvm.processor.merge.MergeByTimeController;
+import org.apache.streampipes.processors.filters.jvm.processor.merge.MergeByTimeProcessor;
 import org.apache.streampipes.processors.filters.jvm.processor.numericalfilter.NumericalFilterProcessor;
-import org.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter.NumericalTextFilterController;
+import org.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter.NumericalTextFilterProcessor;
 import org.apache.streampipes.processors.filters.jvm.processor.projection.ProjectionProcessor;
 import org.apache.streampipes.processors.filters.jvm.processor.schema.MergeBySchemaProcessor;
 import org.apache.streampipes.processors.filters.jvm.processor.textfilter.TextFilterProcessor;
@@ -58,11 +58,11 @@
                     new NumericalFilterProcessor(),
                     new ThresholdDetectionProcessor(),
                     new ProjectionProcessor(),
-                    new MergeByEnrichController(),
-                    new MergeByTimeController(),
+                    new MergeByEnrichProcessor(),
+                    new MergeByTimeProcessor(),
                     new MergeBySchemaProcessor(),
-                    new ComposeController(),
-                    new NumericalTextFilterController(),
+                    new ComposeProcessor(),
+                    new NumericalTextFilterProcessor(),
                     new RateLimitController())
             .registerMessagingFormats(
                     new JsonDataFormatFactory(),
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java
deleted file mode 100644
index 5cb9620..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.compose;
-
-import org.apache.streampipes.model.constants.PropertySelectorConstants;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.runtime.EventFactory;
-import org.apache.streampipes.model.schema.EventSchema;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class Compose implements EventProcessor<ComposeParameters> {
-
-  private Map<String, Event> lastEvents;
-  private EventSchema outputSchema;
-  private List<String> outputKeySelectors;
-
-
-  @Override
-  public void onInvocation(ComposeParameters composeParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
-    this.outputSchema = composeParameters.getGraph().getOutputStream().getEventSchema();
-    this.outputKeySelectors = composeParameters.getOutputKeySelectors();
-    this.lastEvents = new HashMap<>();
-  }
-
-  @Override
-  public void onDetach() {
-    this.lastEvents.clear();
-  }
-
-  @Override
-  public void onEvent(Event event, SpOutputCollector spOutputCollector) {
-    this.lastEvents.put(event.getSourceInfo().getSelectorPrefix(), event);
-    if (lastEvents.size() == 2) {
-      spOutputCollector.collect(buildOutEvent(event.getSourceInfo().getSelectorPrefix()));
-    }
-  }
-
-  private Event buildOutEvent(String currentSelectorPrefix) {
-    return EventFactory.fromEvents(lastEvents.get(currentSelectorPrefix), lastEvents.get
-            (getOtherSelectorPrefix(currentSelectorPrefix)), outputSchema).getSubset(outputKeySelectors);
-  }
-
-  private String getOtherSelectorPrefix(String currentSelectorPrefix) {
-    return currentSelectorPrefix.equals(PropertySelectorConstants.FIRST_STREAM_ID_PREFIX) ?
-            PropertySelectorConstants.SECOND_STREAM_ID_PREFIX : PropertySelectorConstants
-            .FIRST_STREAM_ID_PREFIX;
-  }
-
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java
deleted file mode 100644
index aa22d9f..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.compose;
-
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.helpers.OutputStrategies;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-import java.util.List;
-
-public class ComposeController extends StandaloneEventProcessingDeclarer<ComposeParameters> {
-
-  @Override
-  public DataProcessorDescription declareModel() {
-    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.compose")
-            .category(DataProcessorType.TRANSFORM)
-            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-            .withLocales(Locales.EN)
-            .requiredStream(StreamRequirementsBuilder
-                    .create()
-                    .requiredProperty(EpRequirements.anyProperty())
-                    .build())
-            .requiredStream(StreamRequirementsBuilder
-                    .create()
-                    .requiredProperty(EpRequirements.anyProperty())
-                    .build())
-            .outputStrategy(OutputStrategies.custom(true))
-            .build();
-  }
-
-  @Override
-  public ConfiguredEventProcessor<ComposeParameters> onInvocation(DataProcessorInvocation graph,
-                                                                  ProcessingElementParameterExtractor extractor) {
-
-    List<String> outputKeySelectors = extractor.outputKeySelectors();
-
-    ComposeParameters staticParam = new ComposeParameters(
-            graph, outputKeySelectors);
-
-    return new ConfiguredEventProcessor<>(staticParam, Compose::new);
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeParameters.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeParameters.java
deleted file mode 100644
index 2c0b429..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeParameters.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.compose;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-import java.util.List;
-
-public class ComposeParameters extends EventProcessorBindingParams {
-
-  private List<String> outputKeySelectors;
-
-  public ComposeParameters(DataProcessorInvocation graph, List<String> outputKeySelectors) {
-    super(graph);
-    this.outputKeySelectors = outputKeySelectors;
-  }
-
-  public List<String> getOutputKeySelectors() {
-    return outputKeySelectors;
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeProcessor.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeProcessor.java
new file mode 100644
index 0000000..341f26c
--- /dev/null
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeProcessor.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.streampipes.processors.filters.jvm.processor.compose;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.constants.PropertySelectorConstants;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ComposeProcessor extends StreamPipesDataProcessor {
+
+  private List<String> outputKeySelectors;
+  private Map<String, Event> lastEvents;
+  private EventSchema outputSchema;
+
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.compose")
+            .category(DataProcessorType.TRANSFORM)
+            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+            .withLocales(Locales.EN)
+            .requiredStream(StreamRequirementsBuilder
+                    .create()
+                    .requiredProperty(EpRequirements.anyProperty())
+                    .build())
+            .requiredStream(StreamRequirementsBuilder
+                    .create()
+                    .requiredProperty(EpRequirements.anyProperty())
+                    .build())
+            .outputStrategy(OutputStrategies.custom(true))
+            .build();
+  }
+
+  @Override
+  public void onInvocation(ProcessorParams processorParams, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext eventProcessorRuntimeContext) throws SpRuntimeException {
+    this.outputKeySelectors = processorParams.extractor().outputKeySelectors();
+    this.outputSchema = processorParams.getGraph().getOutputStream().getEventSchema();
+    this.lastEvents = new HashMap<>();
+  }
+
+  @Override
+  public void onEvent(Event event, SpOutputCollector spOutputCollector) throws SpRuntimeException {
+    this.lastEvents.put(event.getSourceInfo().getSelectorPrefix(), event);
+    if (lastEvents.size() == 2) {
+      spOutputCollector.collect(buildOutEvent(event.getSourceInfo().getSelectorPrefix()));
+    }
+  }
+
+  @Override
+  public void onDetach() throws SpRuntimeException {
+    this.lastEvents.clear();
+  }
+
+
+  private Event buildOutEvent(String currentSelectorPrefix) {
+    return EventFactory.fromEvents(lastEvents.get(currentSelectorPrefix), lastEvents.get
+            (getOtherSelectorPrefix(currentSelectorPrefix)), outputSchema).getSubset(outputKeySelectors);
+  }
+
+  private String getOtherSelectorPrefix(String currentSelectorPrefix) {
+    return currentSelectorPrefix.equals(PropertySelectorConstants.FIRST_STREAM_ID_PREFIX) ?
+            PropertySelectorConstants.SECOND_STREAM_ID_PREFIX : PropertySelectorConstants
+            .FIRST_STREAM_ID_PREFIX;
+  }
+
+}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrich.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrich.java
deleted file mode 100644
index 5f24c3e..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrich.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.enrich;
-
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.runtime.EventFactory;
-import org.apache.streampipes.model.schema.EventSchema;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-import java.util.List;
-
-public class MergeByEnrich implements EventProcessor<MergeByEnrichParameters> {
-
-  private EventSchema outputSchema;
-  private List<String> outputKeySelectors;
-  private String selectedStream;
-
-  private Event eventBuffer;
-
-  @Override
-  public void onInvocation(MergeByEnrichParameters composeParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
-    this.outputSchema = composeParameters.getGraph().getOutputStream().getEventSchema();
-    this.outputKeySelectors = composeParameters.getOutputKeySelectors();
-
-    if (composeParameters.getSelectedStream().equals("Stream 1")) {
-      this.selectedStream = "s0";
-    } else {
-      this.selectedStream = "s1";
-    }
-
-    this.eventBuffer = null;
-  }
-
-
-  @Override
-  public void onEvent(Event event, SpOutputCollector spOutputCollector) {
-    String streamId = event.getSourceInfo().getSelectorPrefix();
-
-    // Enrich the selected stream and store last event of other stream
-    if (this.selectedStream.equals(streamId)) {
-      if (this.eventBuffer != null) {
-        Event result = mergeEvents(event, this.eventBuffer);
-        spOutputCollector.collect(result);
-      }
-    } else {
-      this.eventBuffer = event;
-    }
-
-  }
-
-  @Override
-  public void onDetach() {
-    this.eventBuffer = null;
-  }
-
-  private Event mergeEvents(Event e1, Event e2) {
-    return EventFactory.fromEvents(e1, e2, outputSchema).getSubset(outputKeySelectors);
-  }
-
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichController.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichController.java
deleted file mode 100644
index 62dee80..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichController.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.enrich;
-
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.processors.filters.jvm.config.FiltersJvmConfig;
-import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.sdk.helpers.*;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-import java.util.List;
-
-public class MergeByEnrichController extends StandaloneEventProcessingDeclarer<MergeByEnrichParameters> {
-
-  private static final String SELECT_STREAM = "select-stream";
-
-  @Override
-  public DataProcessorDescription declareModel() {
-    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.enrich")
-            .category(DataProcessorType.TRANSFORM)
-            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-            .withLocales(Locales.EN)
-            .requiredStream(StreamRequirementsBuilder
-                    .create()
-                    .requiredProperty(EpRequirements.anyProperty())
-                    .build())
-            .requiredStream(StreamRequirementsBuilder
-                    .create()
-                    .requiredProperty(EpRequirements.anyProperty())
-                    .build())
-            .requiredSingleValueSelection(Labels.withId(SELECT_STREAM),
-                    Options.from("Stream 1", "Stream 2"))
-            .outputStrategy(OutputStrategies.custom(true))
-            .build();
-  }
-
-  @Override
-  public ConfiguredEventProcessor<MergeByEnrichParameters>
-  onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
-
-    List<String> outputKeySelectors = extractor.outputKeySelectors();
-
-    String selectedStream = extractor.selectedSingleValue(SELECT_STREAM, String.class);
-
-    MergeByEnrichParameters staticParam = new MergeByEnrichParameters(
-            graph, outputKeySelectors, selectedStream);
-
-    return new ConfiguredEventProcessor<>(staticParam, MergeByEnrich::new);
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichParameters.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichParameters.java
deleted file mode 100644
index 2a5881f..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichParameters.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.enrich;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-import java.util.List;
-
-public class MergeByEnrichParameters extends EventProcessorBindingParams {
-
-  private List<String> outputKeySelectors;
-  private String selectedStream;
-
-  public MergeByEnrichParameters(DataProcessorInvocation graph, List<String> outputKeySelectors, String selectedStream) {
-    super(graph);
-    this.outputKeySelectors = outputKeySelectors;
-    this.selectedStream = selectedStream;
-  }
-
-  public List<String> getOutputKeySelectors() {
-    return outputKeySelectors;
-  }
-
-  public String getSelectedStream() {
-    return selectedStream;
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichProcessor.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichProcessor.java
new file mode 100644
index 0000000..d7a4e78
--- /dev/null
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/enrich/MergeByEnrichProcessor.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.streampipes.processors.filters.jvm.processor.enrich;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.*;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+import java.util.List;
+
+public class MergeByEnrichProcessor extends StreamPipesDataProcessor {
+
+  private static final String SELECT_STREAM = "select-stream";
+
+  private List<String> outputKeySelectors;
+  private String selectedStream;
+  private EventSchema outputSchema;
+  private Event eventBuffer;
+
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.enrich")
+            .category(DataProcessorType.TRANSFORM)
+            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+            .withLocales(Locales.EN)
+            .requiredStream(StreamRequirementsBuilder
+                    .create()
+                    .requiredProperty(EpRequirements.anyProperty())
+                    .build())
+            .requiredStream(StreamRequirementsBuilder
+                    .create()
+                    .requiredProperty(EpRequirements.anyProperty())
+                    .build())
+            .requiredSingleValueSelection(Labels.withId(SELECT_STREAM),
+                    Options.from("Stream 1", "Stream 2"))
+            .outputStrategy(OutputStrategies.custom(true))
+            .build();
+  }
+
+  @Override
+  public void onInvocation(ProcessorParams processorParams, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext eventProcessorRuntimeContext) throws SpRuntimeException {
+    this.outputKeySelectors = processorParams.extractor().outputKeySelectors();
+
+    this.selectedStream = processorParams.extractor().selectedSingleValue(SELECT_STREAM, String.class);
+
+    this.outputSchema = processorParams.getGraph().getOutputStream().getEventSchema();
+
+    if (this.selectedStream.equals("Stream 1")) {
+      this.selectedStream = "s0";
+    } else {
+      this.selectedStream = "s1";
+    }
+
+    this.eventBuffer = null;
+  }
+
+  @Override
+  public void onEvent(Event event, SpOutputCollector spOutputCollector) throws SpRuntimeException {
+    String streamId = event.getSourceInfo().getSelectorPrefix();
+
+    // Enrich the selected stream and store last event of other stream
+    if (this.selectedStream.equals(streamId)) {
+      if (this.eventBuffer != null) {
+        Event result = mergeEvents(event, this.eventBuffer);
+        spOutputCollector.collect(result);
+      }
+    } else {
+      this.eventBuffer = event;
+    }
+  }
+
+  @Override
+  public void onDetach() throws SpRuntimeException {
+
+  }
+
+  private Event mergeEvents(Event e1, Event e2) {
+    return EventFactory.fromEvents(e1, e2, outputSchema).getSubset(outputKeySelectors);
+  }
+}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTime.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTime.java
deleted file mode 100644
index c53cb1c..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTime.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.merge;
-
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.runtime.EventFactory;
-import org.apache.streampipes.model.schema.EventSchema;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-import java.util.List;
-
-public class MergeByTime implements EventProcessor<MergeByTimeParameters> {
-
-  private EventSchema outputSchema;
-  private List<String> outputKeySelectors;
-
-  private String timestampFieldStream0;
-  private String timestampFieldStream1;
-  private Integer timeInterval;
-
-  private StreamBuffer streamBufferS0;
-  private StreamBuffer streamBufferS1;
-
-  @Override
-  public void onInvocation(MergeByTimeParameters composeParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
-    this.outputSchema = composeParameters.getGraph().getOutputStream().getEventSchema();
-    this.outputKeySelectors = composeParameters.getOutputKeySelectors();
-    this.timestampFieldStream0 = composeParameters.getTimestampFieldStream1();
-    this.timestampFieldStream1 = composeParameters.getTimestampFieldStream2();
-    this.timeInterval = composeParameters.getTimeInterval();
-
-    this.streamBufferS0 = new StreamBuffer(this.timestampFieldStream0);
-    this.streamBufferS1 = new StreamBuffer(this.timestampFieldStream1);
-  }
-
-
-  @Override
-  public void onEvent(Event event, SpOutputCollector spOutputCollector) {
-    String streamId = event.getSourceInfo().getSelectorPrefix();
-
-    // Decide to which buffer the event should be added
-    if ("s0".equals(streamId)) {
-      this.streamBufferS0.add(event);
-    } else {
-      this.streamBufferS1.add(event);
-    }
-
-    // Calculate matching events between data streams
-    for (Event e0 : this.streamBufferS0.getList()) {
-      long time0 = e0.getFieldBySelector(timestampFieldStream0).getAsPrimitive().getAsLong();
-      for (Event e1 : this.streamBufferS1.getList()) {
-        long time1 = e1.getFieldBySelector(timestampFieldStream1).getAsPrimitive().getAsLong();
-
-        if (time0 + timeInterval > time1 && time1 > time0 - timeInterval) {
-          Event resultingEvent = mergeEvents(e0, e1);
-          spOutputCollector.collect(resultingEvent);
-          this.streamBufferS0.removeOldEvents(time0);
-          this.streamBufferS1.removeOldEvents(time1);
-        }
-      }
-    }
-
-    // Clean up buffer if events do not match to avoid buffer overflow
-    if (this.streamBufferS0.getLength() > 0 && this.streamBufferS1.getLength() > 0) {
-      Event e0 = this.streamBufferS0.get(0);
-      Event e1 = this.streamBufferS1.get(0);
-
-      long time0 = e0.getFieldBySelector(this.timestampFieldStream0).getAsPrimitive().getAsLong();
-      long time1 = e1.getFieldBySelector(this.timestampFieldStream1).getAsPrimitive().getAsLong();
-
-      if (time0 > time1) {
-        this.streamBufferS0.removeOldEvents(time0);
-        this.streamBufferS1.removeOldEvents(time0);
-      } else {
-        this.streamBufferS0.removeOldEvents(time1);
-        this.streamBufferS1.removeOldEvents(time1);
-      }
-    }
-
-  }
-
-  @Override
-  public void onDetach() {
-    this.streamBufferS0.reset();
-    this.streamBufferS1.reset();
-  }
-
-  private Event mergeEvents(Event e1, Event e2) {
-    return EventFactory.fromEvents(e1, e2, outputSchema).getSubset(outputKeySelectors);
-  }
-
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeController.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeController.java
deleted file mode 100644
index e6446f6..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeController.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.merge;
-
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.processors.filters.jvm.config.FiltersJvmConfig;
-import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-import org.apache.streampipes.sdk.helpers.Labels;
-import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.helpers.OutputStrategies;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-import java.util.List;
-
-public class MergeByTimeController extends StandaloneEventProcessingDeclarer<MergeByTimeParameters> {
-
-  private static final String TIMESTAMP_MAPPING_STREAM_1_KEY = "timestamp_mapping_stream_1";
-  private static final String TIMESTAMP_MAPPING_STREAM_2_KEY = "timestamp_mapping_stream_2";
-  private static final String NUMBER_MAPPING = "number_mapping";
-  private static final String TIME_INTERVAL = "time-interval";
-
-  @Override
-  public DataProcessorDescription declareModel() {
-    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.merge")
-            .category(DataProcessorType.TRANSFORM)
-            .withAssets(Assets.DOCUMENTATION, Assets.ICON, "merge_description.png")
-            .withLocales(Locales.EN)
-            .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
-                    EpRequirements.timestampReq(),
-                    Labels.withId(TIMESTAMP_MAPPING_STREAM_1_KEY),
-                    PropertyScope.NONE).build())
-            .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
-                    EpRequirements.timestampReq(),
-                    Labels.withId(TIMESTAMP_MAPPING_STREAM_2_KEY),
-                    PropertyScope.NONE).build())
-            .requiredIntegerParameter(Labels.withId(TIME_INTERVAL), NUMBER_MAPPING)
-            .outputStrategy(OutputStrategies.custom(true))
-            .build();
-  }
-
-  @Override
-  public ConfiguredEventProcessor<MergeByTimeParameters>
-  onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
-
-    List<String> outputKeySelectors = extractor.outputKeySelectors();
-
-    String timestampFieldStream1 = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_STREAM_1_KEY);
-    String timestampFieldStream2 = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_STREAM_2_KEY);
-    Integer timeInterval = extractor.singleValueParameter(TIME_INTERVAL, Integer.class);
-
-    MergeByTimeParameters staticParam = new MergeByTimeParameters(
-            graph, outputKeySelectors, timestampFieldStream1, timestampFieldStream2, timeInterval);
-
-    return new ConfiguredEventProcessor<>(staticParam, MergeByTime::new);
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeParameters.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeParameters.java
deleted file mode 100644
index f39405c..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeParameters.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.merge;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-import java.util.List;
-
-public class MergeByTimeParameters extends EventProcessorBindingParams {
-
-  private List<String> outputKeySelectors;
-  private String timestampFieldStream1;
-  private String timestampFieldStream2;
-  private Integer timeInterval;
-
-  public MergeByTimeParameters(DataProcessorInvocation graph, List<String> outputKeySelectors,
-                               String timestampFieldStream1, String timestampFieldStream2, Integer timeInterval) {
-    super(graph);
-    this.outputKeySelectors = outputKeySelectors;
-    this.timestampFieldStream1 = timestampFieldStream1;
-    this.timestampFieldStream2 = timestampFieldStream2;
-    this.timeInterval = timeInterval;
-  }
-
-  public List<String> getOutputKeySelectors() {
-    return outputKeySelectors;
-  }
-
-  public String getTimestampFieldStream1() {
-    return timestampFieldStream1;
-  }
-
-  public String getTimestampFieldStream2() {
-    return timestampFieldStream2;
-  }
-
-  public Integer getTimeInterval() {
-    return timeInterval;
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
new file mode 100644
index 0000000..8d41aa7
--- /dev/null
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.streampipes.processors.filters.jvm.processor.merge;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+import java.util.List;
+
+public class MergeByTimeProcessor extends StreamPipesDataProcessor {
+
+  private static final String TIMESTAMP_MAPPING_STREAM_1_KEY = "timestamp_mapping_stream_1";
+  private static final String TIMESTAMP_MAPPING_STREAM_2_KEY = "timestamp_mapping_stream_2";
+  private static final String NUMBER_MAPPING = "number_mapping";
+  private static final String TIME_INTERVAL = "time-interval";
+
+  private List<String> outputKeySelectors;
+  private String timestampFieldStream0;
+  private String timestampFieldStream1;
+  private Integer timeInterval;
+  private EventSchema outputSchema;
+
+  private StreamBuffer streamBufferS0;
+  private StreamBuffer streamBufferS1;
+
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.merge")
+            .category(DataProcessorType.TRANSFORM)
+            .withAssets(Assets.DOCUMENTATION, Assets.ICON, "merge_description.png")
+            .withLocales(Locales.EN)
+            .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
+                    EpRequirements.timestampReq(),
+                    Labels.withId(TIMESTAMP_MAPPING_STREAM_1_KEY),
+                    PropertyScope.NONE).build())
+            .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
+                    EpRequirements.timestampReq(),
+                    Labels.withId(TIMESTAMP_MAPPING_STREAM_2_KEY),
+                    PropertyScope.NONE).build())
+            .requiredIntegerParameter(Labels.withId(TIME_INTERVAL), NUMBER_MAPPING)
+            .outputStrategy(OutputStrategies.custom(true))
+            .build();
+  }
+
+  @Override
+  public void onInvocation(ProcessorParams processorParams, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext eventProcessorRuntimeContext) throws SpRuntimeException {
+    this.outputSchema = processorParams.getGraph().getOutputStream().getEventSchema();
+    this.outputKeySelectors = processorParams.extractor().outputKeySelectors();
+    this.timestampFieldStream0 = processorParams.extractor().mappingPropertyValue(TIMESTAMP_MAPPING_STREAM_1_KEY);
+    this.timestampFieldStream1 = processorParams.extractor().mappingPropertyValue(TIMESTAMP_MAPPING_STREAM_2_KEY);
+
+    this.timeInterval = processorParams.extractor().singleValueParameter(TIME_INTERVAL,Integer.class);
+
+    this.streamBufferS0 = new StreamBuffer(this.timestampFieldStream0);
+    this.streamBufferS1 = new StreamBuffer(this.timestampFieldStream1);
+  }
+
+  @Override
+  public void onEvent(Event event, SpOutputCollector spOutputCollector) throws SpRuntimeException {
+    String streamId = event.getSourceInfo().getSelectorPrefix();
+
+    // Decide to which buffer the event should be added
+    if ("s0".equals(streamId)) {
+      this.streamBufferS0.add(event);
+    } else {
+      this.streamBufferS1.add(event);
+    }
+
+    // Calculate matching events between data streams
+    for (Event e0 : this.streamBufferS0.getList()) {
+      long time0 = e0.getFieldBySelector(timestampFieldStream0).getAsPrimitive().getAsLong();
+      for (Event e1 : this.streamBufferS1.getList()) {
+        long time1 = e1.getFieldBySelector(timestampFieldStream1).getAsPrimitive().getAsLong();
+
+        if (time0 + timeInterval > time1 && time1 > time0 - timeInterval) {
+          Event resultingEvent = mergeEvents(e0, e1);
+          spOutputCollector.collect(resultingEvent);
+          this.streamBufferS0.removeOldEvents(time0);
+          this.streamBufferS1.removeOldEvents(time1);
+        }
+      }
+    }
+
+    // Clean up buffer if events do not match to avoid buffer overflow
+    if (this.streamBufferS0.getLength() > 0 && this.streamBufferS1.getLength() > 0) {
+      Event e0 = this.streamBufferS0.get(0);
+      Event e1 = this.streamBufferS1.get(0);
+
+      long time0 = e0.getFieldBySelector(this.timestampFieldStream0).getAsPrimitive().getAsLong();
+      long time1 = e1.getFieldBySelector(this.timestampFieldStream1).getAsPrimitive().getAsLong();
+
+      if (time0 > time1) {
+        this.streamBufferS0.removeOldEvents(time0);
+        this.streamBufferS1.removeOldEvents(time0);
+      } else {
+        this.streamBufferS0.removeOldEvents(time1);
+        this.streamBufferS1.removeOldEvents(time1);
+      }
+    }
+
+  }
+
+  @Override
+  public void onDetach() throws SpRuntimeException {
+
+  }
+
+
+  private Event mergeEvents(Event e1, Event e2) {
+    return EventFactory.fromEvents(e1, e2, outputSchema).getSubset(outputKeySelectors);
+  }
+}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilter.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilter.java
deleted file mode 100644
index 27c3950..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilter.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter;
-
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-public class NumericalTextFilter implements EventProcessor<NumericalTextFilterParameters> {
-
-  private NumericalTextFilterParameters params;
-
-  @Override
-  public void onInvocation(NumericalTextFilterParameters numericalTextFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
-          runtimeContext) {
-    this.params = numericalTextFilterParameters;
-  }
-
-  @Override
-  public void onEvent(Event event, SpOutputCollector out) {
-    boolean satisfiesNumberFilter = false;
-    boolean satisfiesTextFilter = false;
-
-    Double numbervalue = event.getFieldBySelector(params.getNumberProperty())
-            .getAsPrimitive()
-            .getAsDouble();
-
-    String value = event.getFieldBySelector(params.getTextProperty())
-            .getAsPrimitive()
-            .getAsString();
-
-    //Double value = Double.parseDouble(String.valueOf(in.get(params.getFilterProperty())));
-    Double threshold = params.getNumberThreshold();
-
-    if (params.getNumericalOperator() == NumericalOperator.EQ) {
-      satisfiesNumberFilter = (Math.abs(numbervalue - threshold) < 0.000001);
-    } else if (params.getNumericalOperator() == NumericalOperator.GE) {
-      satisfiesNumberFilter = (numbervalue >= threshold);
-    } else if (params.getNumericalOperator() == NumericalOperator.GT) {
-      satisfiesNumberFilter = numbervalue > threshold;
-    } else if (params.getNumericalOperator() == NumericalOperator.LE) {
-      satisfiesNumberFilter = (numbervalue <= threshold);
-    } else if (params.getNumericalOperator() == NumericalOperator.LT) {
-      satisfiesNumberFilter = (numbervalue < threshold);
-    } else if (params.getNumericalOperator() == NumericalOperator.IE) {
-      satisfiesNumberFilter = (Math.abs(numbervalue - threshold) > 0.000001);
-    }
-
-    if (params.getTextOperator() == StringOperator.MATCHES) {
-      satisfiesTextFilter = (value.equals(params.getTextKeyword()));
-    } else if (params.getTextOperator() == StringOperator.CONTAINS) {
-      satisfiesTextFilter = (value.contains(params.getTextKeyword()));
-    }
-
-    if (satisfiesNumberFilter && satisfiesTextFilter) {
-      out.collect(event);
-    }
-  }
-
-  @Override
-  public void onDetach() {
-
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterController.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterController.java
deleted file mode 100644
index 32882d5..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterController.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter;
-
-import org.apache.streampipes.container.api.ResolvesContainerProvidedOptions;
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.model.staticproperty.Option;
-import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
-import org.apache.streampipes.sdk.helpers.*;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-import java.util.List;
-
-public class NumericalTextFilterController extends StandaloneEventProcessingDeclarer<NumericalTextFilterParameters> {
-
-  // number
-  private static final String NUMBER_MAPPING = "number-mapping";
-  private static final String NUMBER_OPERATION = "number-operation";
-  private static final String NUMBER_VALUE = "number-value";
-  // text
-  private static final String TEXT_MAPPING = "text-mapping";
-  private static final String TEXT_OPERATION = "text-operation";
-  private static final String TEXT_KEYWORD = "text-keyword";
-
-  @Override
-  public DataProcessorDescription declareModel() {
-    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.numericaltextfilter")
-            .category(DataProcessorType.FILTER)
-            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-            .withLocales(Locales.EN)
-            .requiredStream(StreamRequirementsBuilder
-                    .create()
-                    .requiredPropertyWithUnaryMapping(EpRequirements.numberReq(),
-                            Labels.withId(NUMBER_MAPPING),
-                            PropertyScope.MEASUREMENT_PROPERTY)
-                    .requiredPropertyWithUnaryMapping(EpRequirements.stringReq(),
-                            Labels.withId(TEXT_MAPPING), PropertyScope.NONE)
-                    .build())
-            .requiredSingleValueSelection(Labels.withId(NUMBER_OPERATION), Options.from("<", "<=", ">",
-                    ">=", "==", "!="))
-            .requiredFloatParameter(Labels.withId(NUMBER_VALUE), NUMBER_MAPPING)
-            .requiredSingleValueSelection(Labels.withId(TEXT_OPERATION), Options.from("MATCHES",
-                    "CONTAINS"))
-            .requiredTextParameter(Labels.withId(TEXT_KEYWORD), "text")
-            .outputStrategy(OutputStrategies.keep())
-            .build();
-
-  }
-
-  @Override
-  public ConfiguredEventProcessor<NumericalTextFilterParameters> onInvocation
-          (DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
-
-    // number
-    String numberProperty = extractor.mappingPropertyValue(NUMBER_MAPPING);
-    Double numberThreshold = extractor.singleValueParameter(NUMBER_VALUE, Double.class);
-    String numberOperation = extractor.selectedSingleValue(NUMBER_OPERATION, String.class);
-
-    // text
-    String textProperty = extractor.mappingPropertyValue(TEXT_MAPPING);
-    String textKeyword = extractor.singleValueParameter(TEXT_KEYWORD, String.class);
-    String textOperation = extractor.selectedSingleValue(TEXT_OPERATION, String.class);
-
-    String numOperation = "GT";
-
-    if (numberOperation.equals("<=")) {
-      numOperation = "LE";
-    } else if (numberOperation.equals("<")) {
-      numOperation = "LT";
-    } else if (numberOperation.equals(">=")) {
-      numOperation = "GE";
-    } else if (numberOperation.equals("==")) {
-      numOperation = "EQ";
-    } else if (numberOperation.equals("!=")) {
-      numOperation = "IE";
-    }
-
-
-    NumericalTextFilterParameters staticParam = new NumericalTextFilterParameters(
-            graph,
-            numberThreshold,
-            NumericalOperator.valueOf(numOperation),
-            numberProperty,
-            textKeyword,
-            StringOperator.valueOf(textOperation),
-            textProperty);
-
-    return new ConfiguredEventProcessor<>(staticParam, NumericalTextFilter::new);
-  }
-
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterParameters.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterParameters.java
deleted file mode 100644
index c669430..0000000
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterParameters.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class NumericalTextFilterParameters extends EventProcessorBindingParams {
-
-  private double numberThreshold;
-  private NumericalOperator numericalOperator;
-  private String numberProperty;
-  private String textKeyword;
-  private StringOperator textOperator;
-  private String textProperty;
-
-  public NumericalTextFilterParameters(DataProcessorInvocation graph, Double numberThreshold, NumericalOperator
-          NumericalOperator, String numberProperty, String textKeyword, StringOperator textOperator, String textProperty) {
-    super(graph);
-    this.numberThreshold = numberThreshold;
-    this.numericalOperator = NumericalOperator;
-    this.numberProperty = numberProperty;
-    this.textKeyword = textKeyword;
-    this.textOperator = textOperator;
-    this.textProperty = textProperty;
-  }
-
-  public double getNumberThreshold() {
-    return numberThreshold;
-  }
-
-  public NumericalOperator getNumericalOperator() {
-    return numericalOperator;
-  }
-
-  public String getNumberProperty() {
-    return numberProperty;
-  }
-
-  public String getTextKeyword() {
-    return textKeyword;
-  }
-
-  public StringOperator getTextOperator() {
-    return textOperator;
-  }
-
-  public String getTextProperty() {
-    return textProperty;
-  }
-}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterProcessor.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterProcessor.java
new file mode 100644
index 0000000..3fec57d
--- /dev/null
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericaltextfilter/NumericalTextFilterProcessor.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.*;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+
+public class NumericalTextFilterProcessor extends StreamPipesDataProcessor {
+
+  // number
+  private static final String NUMBER_MAPPING = "number-mapping";
+  private static final String NUMBER_OPERATION = "number-operation";
+  private static final String NUMBER_VALUE = "number-value";
+  // text
+  private static final String TEXT_MAPPING = "text-mapping";
+  private static final String TEXT_OPERATION = "text-operation";
+  private static final String TEXT_KEYWORD = "text-keyword";
+
+  private double numberThreshold;
+  private NumericalOperator numericalOperator;
+  private String numberProperty;
+  private String textKeyword;
+  private StringOperator textOperator;
+  private String textProperty;
+
+
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.numericaltextfilter")
+            .category(DataProcessorType.FILTER)
+            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+            .withLocales(Locales.EN)
+            .requiredStream(StreamRequirementsBuilder
+                    .create()
+                    .requiredPropertyWithUnaryMapping(EpRequirements.numberReq(),
+                            Labels.withId(NUMBER_MAPPING),
+                            PropertyScope.MEASUREMENT_PROPERTY)
+                    .requiredPropertyWithUnaryMapping(EpRequirements.stringReq(),
+                            Labels.withId(TEXT_MAPPING), PropertyScope.NONE)
+                    .build())
+            .requiredSingleValueSelection(Labels.withId(NUMBER_OPERATION), Options.from("<", "<=", ">",
+                    ">=", "==", "!="))
+            .requiredFloatParameter(Labels.withId(NUMBER_VALUE), NUMBER_MAPPING)
+            .requiredSingleValueSelection(Labels.withId(TEXT_OPERATION), Options.from("MATCHES",
+                    "CONTAINS"))
+            .requiredTextParameter(Labels.withId(TEXT_KEYWORD), "text")
+            .outputStrategy(OutputStrategies.keep())
+            .build();
+
+  }
+
+  @Override
+  public void onInvocation(ProcessorParams processorParams, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext eventProcessorRuntimeContext) throws SpRuntimeException {
+
+    // number
+    this.numberProperty = processorParams.extractor().mappingPropertyValue(NUMBER_MAPPING);
+    this.numberThreshold = processorParams.extractor().singleValueParameter(NUMBER_VALUE, Double.class);
+    String numberOperation = processorParams.extractor().selectedSingleValue(NUMBER_OPERATION, String.class);
+
+    // text
+    this.textProperty = processorParams.extractor().mappingPropertyValue(TEXT_MAPPING);
+    this.textKeyword = processorParams.extractor().singleValueParameter(TEXT_KEYWORD, String.class);
+    this.textOperator = StringOperator.valueOf(processorParams.extractor().selectedSingleValue(TEXT_OPERATION, String.class));
+
+    String numOperation = "GT";
+
+    if (numberOperation.equals("<=")) {
+      numOperation = "LE";
+    } else if (numberOperation.equals("<")) {
+      numOperation = "LT";
+    } else if (numberOperation.equals(">=")) {
+      numOperation = "GE";
+    } else if (numberOperation.equals("==")) {
+      numOperation = "EQ";
+    } else if (numberOperation.equals("!=")) {
+      numOperation = "IE";
+    }
+
+    this.numericalOperator = NumericalOperator.valueOf(numOperation);
+
+  }
+
+  @Override
+  public void onEvent(Event event, SpOutputCollector spOutputCollector) throws SpRuntimeException {
+    boolean satisfiesNumberFilter = false;
+    boolean satisfiesTextFilter = false;
+
+    Double numbervalue = event.getFieldBySelector(this.numberProperty)
+            .getAsPrimitive()
+            .getAsDouble();
+
+    String value = event.getFieldBySelector(this.textProperty)
+            .getAsPrimitive()
+            .getAsString();
+
+    Double threshold = this.numberThreshold;
+
+    if (this.numericalOperator == NumericalOperator.EQ) {
+      satisfiesNumberFilter = (Math.abs(numbervalue - threshold) < 0.000001);
+    } else if (this.numericalOperator  == NumericalOperator.GE) {
+      satisfiesNumberFilter = (numbervalue >= threshold);
+    } else if (this.numericalOperator  == NumericalOperator.GT) {
+      satisfiesNumberFilter = numbervalue > threshold;
+    } else if (this.numericalOperator  == NumericalOperator.LE) {
+      satisfiesNumberFilter = (numbervalue <= threshold);
+    } else if (this.numericalOperator  == NumericalOperator.LT) {
+      satisfiesNumberFilter = (numbervalue < threshold);
+    } else if (this.numericalOperator  == NumericalOperator.IE) {
+      satisfiesNumberFilter = (Math.abs(numbervalue - threshold) > 0.000001);
+    }
+
+    if (this.textOperator == StringOperator.MATCHES) {
+      satisfiesTextFilter = (value.equals(this.textKeyword));
+    } else if (this.textOperator == StringOperator.CONTAINS) {
+      satisfiesTextFilter = (value.contains(this.textKeyword));
+    }
+
+    if (satisfiesNumberFilter && satisfiesTextFilter) {
+      spOutputCollector.collect(event);
+    }
+  }
+
+  @Override
+  public void onDetach() throws SpRuntimeException {
+
+  }
+}