Repository: apache/streampipes
Pull request #2024 (2023-10-13 13:00:11)

fix: make data retrieval of IOLink sensor more robust
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
This PR makes the data retrieval within the IOLink adapter more robust. Prior to it, an error occurring for one port stopped the processing of all subsequent ports. This means, that in case the processing fails for port 2, e.g., since this port is not connected to a sensor and therefore does not deliver any data, parsing for port 3 and 4 is skipped. 
From now on, only the affected port will be skipped.

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/streampipes-client-python/poetry.lock b/streampipes-client-python/poetry.lock
index 1d2d712c5e..f1136c22de 100644
--- a/streampipes-client-python/poetry.lock
+++ b/streampipes-client-python/poetry.lock
@@ -1983,13 +1983,13 @@ testing = ["pytest", "pytest-benchmark"]
 
 [[package]]
 name = "pre-commit"
-version = "3.4.0"
+version = "3.5.0"
 description = "A framework for managing and maintaining multi-language pre-commit hooks."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "pre_commit-3.4.0-py2.py3-none-any.whl", hash = "sha256:96d529a951f8b677f730a7212442027e8ba53f9b04d217c4c67dc56c393ad945"},
-    {file = "pre_commit-3.4.0.tar.gz", hash = "sha256:6bbd5129a64cad4c0dfaeeb12cd8f7ea7e15b77028d985341478c8af3c759522"},
+    {file = "pre_commit-3.5.0-py2.py3-none-any.whl", hash = "sha256:841dc9aef25daba9a0238cd27984041fa0467b4199fc4852e27950664919f660"},
+    {file = "pre_commit-3.5.0.tar.gz", hash = "sha256:5804465c675b659b0862f07907f96295d490822a450c4c40e747d0b1c6ebcb32"},
 ]
 
 [package.dependencies]
@@ -3290,4 +3290,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = ">3.8.1,<3.12"
-content-hash = "f523de24f08c24e6ea7d26e792c7092d630b0451597f656a7ed19292d01201bb"
+content-hash = "41a26070e9f75bd6c9cabed820e02dc87fdae3e8053e4f20997be7d4796878ef"
diff --git a/streampipes-client-python/pyproject.toml b/streampipes-client-python/pyproject.toml
index 8b9cae9c83..3d6d73aabf 100644
--- a/streampipes-client-python/pyproject.toml
+++ b/streampipes-client-python/pyproject.toml
@@ -73,7 +73,7 @@ interrogate = { version = "1.5.0", extras = ["png"] }
 isort = "5.12.0"
 mypy = "1.6.0"
 ruff = "0.0.280"
-pre-commit = "3.4.0"
+pre-commit = "3.5.0"
 pytest = "7.4.0"
 pytest-cov = "4.1.0"
 pyupgrade = "3.15.0"
diff --git a/streampipes-client-python/streampipes/client/client.py b/streampipes-client-python/streampipes/client/client.py
index 7d613d8704..047e52cc89 100644
--- a/streampipes-client-python/streampipes/client/client.py
+++ b/streampipes-client-python/streampipes/client/client.py
@@ -65,6 +65,11 @@ class StreamPipesClient:
     dataStreamApi: DataStreamEndpoint
         Instance of the data stream endpoint
 
+    Raises
+    ------
+    AttributeError:
+        In case an invalid configuration of the `StreamPipesClientConfig` is passed
+
     Examples
     --------
 
@@ -117,6 +122,15 @@ class StreamPipesClient:
         client_config: StreamPipesClientConfig,
         logging_level: Optional[int] = logging.INFO,
     ):
+        # validate client config
+        # `https_disabled` and `port` 443 is an invalid configuration
+        if client_config.https_disabled and client_config.port == 443:
+            raise AttributeError(
+                "Invalid configuration passed! The given client configuration has "
+                "`https_disabled` set to `True` and `port` set to `443`.\n "
+                "If you want to connect to port 443, use `https_disabled=False` or "
+                "alternatively connect to port `80`."
+            )
         self.client_config = client_config
 
         # set up a requests session
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java
index e2606996d9..79b3dfe677 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java
@@ -54,16 +54,20 @@ public class IfmAlMqttAdapter implements StreamPipesAdapter {
 
   public static final String ID = "org.apache.streampipes.connect.iiot.adapters.iolink";
 
+  public static final String KEY_PORT_INFORMATION = "/iolinkmaster/port[%s]/iolinkdevice/pdin";
+
   private static final Logger LOG = LoggerFactory.getLogger(IfmAlMqttAdapter.class);
 
   private static final String PORTS = "ports";
   private static final String SENSOR_TYPE = "sensor_type";
 
+  private final IParser parser;
+  private boolean missingEventDataDetected = false;
+  private boolean missingPortInformationDetected = false;
+
   private MqttConsumer mqttConsumer;
   private MqttConfig mqttConfig;
 
-  private final IParser parser;
-
   private List<String> ports;
 
   public IfmAlMqttAdapter() {
@@ -110,25 +114,61 @@ public class IfmAlMqttAdapter implements StreamPipesAdapter {
 
               for (int i = 0; i < ports.size(); i++) {
 
-                var portResult = getMap(payload,
-                    "/iolinkmaster/port[%s]/iolinkdevice/pdin".formatted(ports.get(i)));
-                var eventData = (String) portResult.get("data");
-
-                var parsedEvent = sensor.parseEvent(eventData);
-                parsedEvent.put("timestamp", System.currentTimeMillis() + i);
-                parsedEvent.put("port", "port" + ports.get(i));
-                parsedEvent.put(SensorVVB001.IO_LINK_MASTER_SN, serialnumber);
-
-                collector.collect(parsedEvent);
+                String keyPortInformation = KEY_PORT_INFORMATION.formatted(ports.get(i));
+
+                Map<String, Object> portResult;
+                if (payload.containsKey(keyPortInformation)) {
+
+                  portResult = getMap(payload, keyPortInformation);
+
+                  try {
+                    String eventData;
+                    if (portResult.containsKey("data")) {
+                      eventData = (String) portResult.get("data");
+
+                      var parsedEvent = sensor.parseEvent(eventData);
+                      parsedEvent.put("timestamp", System.currentTimeMillis() + i);
+                      parsedEvent.put("port", "port" + ports.get(i));
+                      parsedEvent.put(SensorVVB001.IO_LINK_MASTER_SN, serialnumber);
+
+                      collector.collect(parsedEvent);
+                    } else {
+                      if (!missingEventDataDetected) {
+                        adapterRuntimeContext
+                                .getLogger()
+                                .warn("Payload for port %s does not contain event data".formatted(i), "");
+                        LOG.error(
+                                "IoLink event does not look like expected. "
+                                + "No port information found for port {}.", i);
+                        missingEventDataDetected = true;
+                      }
+                    }
+                  } catch (Exception e) {
+                    adapterRuntimeContext
+                            .getLogger()
+                            .error(e);
+                    LOG.error("Data from IOLink sensor could not be extracted for port {}: {}", i, e);
+                  }
+
+                } else {
+                  if (!missingPortInformationDetected) {
+                    adapterRuntimeContext
+                            .getLogger()
+                            .warn("Event does not contain information about port " + i, "");
+                    LOG.error("IoLink event does not look like expected. No port information found for port {}.", i);
+                    missingPortInformationDetected = true;
+                  }
+                }
               }
             });
-          } catch (Exception e) {
+          } catch (ParseException e) {
             adapterRuntimeContext
-                .getLogger()
-                .error(e);
-            LOG.error("Could not parse event", e);
+                    .getLogger()
+                    .error(e);
+            LOG.error("IOLink master event could not be parsed.", e);
           }
-        });
+        }
+        );
 
     Thread thread = new Thread(this.mqttConsumer);
     thread.start();
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
index 98bfabefdb..09dc0887bf 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/pom.xml
@@ -43,6 +43,12 @@
             <groupId>org.quartz-scheduler</groupId>
             <artifactId>quartz</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
index fcc5c7e229..0edb33b964 100644
--- a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/merge/MergeByTimeProcessor.java
@@ -40,10 +40,10 @@ 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";
+  protected static final String TIMESTAMP_MAPPING_STREAM_1_KEY = "timestamp_mapping_stream_1";
+  protected static final String TIMESTAMP_MAPPING_STREAM_2_KEY = "timestamp_mapping_stream_2";
+  protected static final String NUMBER_MAPPING = "number_mapping";
+  protected static final String TIME_INTERVAL = "time-interval";
 
   private List<String> outputKeySelectors;
   private String timestampFieldStream0;
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 3be0585a6d..bbd1db46e6 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -82,7 +82,7 @@
         "@cypress/webpack-preprocessor": "5.17.0",
         "@ngtools/webpack": "^15.2.9",
         "@types/angular": "^1.8.4",
-        "@types/jasmine": "~4.3.1",
+        "@types/jasmine": "~5.1.0",
         "@types/jqueryui": "^1.12.16",
         "@types/jshint": "^2.12.1",
         "@types/leaflet": "^1.9.0",
@@ -7088,9 +7088,9 @@
       }
     },
     "node_modules/@types/jasmine": {
-      "version": "4.3.1",
-      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-4.3.1.tgz",
-      "integrity": "sha512-Vu8l+UGcshYmV1VWwULgnV/2RDbBaO6i2Ptx7nd//oJPIZGhoI1YLST4VKagD2Pq/Bc2/7zvtvhM7F3p4SN7kQ==",
+      "version": "5.1.0",
+      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-5.1.0.tgz",
+      "integrity": "sha512-XOV0KsqXNX2gUSqk05RWeolIMgaAQ7+l/ozOBoQ8NGwLg+E7J9vgagODtNgfim4jCzEUP0oJ3gnXeC+Zv+Xi1A==",
       "dev": true
     },
     "node_modules/@types/jquery": {
@@ -33617,9 +33617,9 @@
       }
     },
     "@types/jasmine": {
-      "version": "4.3.1",
-      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-4.3.1.tgz",
-      "integrity": "sha512-Vu8l+UGcshYmV1VWwULgnV/2RDbBaO6i2Ptx7nd//oJPIZGhoI1YLST4VKagD2Pq/Bc2/7zvtvhM7F3p4SN7kQ==",
+      "version": "5.1.0",
+      "resolved": "https://registry.npmjs.org/@types/jasmine/-/jasmine-5.1.0.tgz",
+      "integrity": "sha512-XOV0KsqXNX2gUSqk05RWeolIMgaAQ7+l/ozOBoQ8NGwLg+E7J9vgagODtNgfim4jCzEUP0oJ3gnXeC+Zv+Xi1A==",
       "dev": true
     },
     "@types/jquery": {
diff --git a/ui/package.json b/ui/package.json
index 002515edb1..5197195e07 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -104,7 +104,7 @@
     "@cypress/webpack-preprocessor": "5.17.0",
     "@ngtools/webpack": "^15.2.9",
     "@types/angular": "^1.8.4",
-    "@types/jasmine": "~4.3.1",
+    "@types/jasmine": "~5.1.0",
     "@types/jqueryui": "^1.12.16",
     "@types/jshint": "^2.12.1",
     "@types/leaflet": "^1.9.0",


Test patch:
diff --git a/streampipes-client-python/tests/client/test_client.py b/streampipes-client-python/tests/client/test_client.py
index a9c1ffc265..260ba35ee4 100644
--- a/streampipes-client-python/tests/client/test_client.py
+++ b/streampipes-client-python/tests/client/test_client.py
@@ -79,6 +79,18 @@ class TestStreamPipesClient(TestCase):
         self.assertTrue(isinstance(result.dataLakeMeasureApi, DataLakeMeasureEndpoint))
         self.assertEqual(result.base_api_path, "https://localhost:443/streampipes-backend/")
 
+    def test_client_create_invalid_config(self):
+
+        with self.assertRaises(AttributeError):
+            StreamPipesClient.create(
+                client_config=StreamPipesClientConfig(
+                    credential_provider=StreamPipesApiKeyCredentials(username="user", api_key="key"),
+                    host_address="localhost",
+                    https_disabled=True,
+                    port=443,
+                )
+            )
+
     @patch("streampipes.client.client.logger", autospec=True)
     @patch("streampipes.endpoint.endpoint.APIEndpoint._make_request", autospec=True)
     def test_client_describe(self, make_request: MagicMock, mocked_logger: MagicMock):
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java
new file mode 100644
index 0000000000..30a5358308
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/booleanfilter/TestBooleanFilterProcessor.java
@@ -0,0 +1,153 @@
+/*
+ * 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.booleanfilter;
+
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.model.staticproperty.OneOfStaticProperty;
+import org.apache.streampipes.test.extensions.api.StoreEventCollector;
+import org.apache.streampipes.test.generator.EventStreamGenerator;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class TestBooleanFilterProcessor {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestBooleanFilterProcessor.class);
+
+  @org.junit.runners.Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+    List<Boolean> allTrue = Arrays.asList(true, true, true);
+    List<Boolean> allFalse = Arrays.asList(false, false, false);
+    List<Boolean> someTrueSomeFalse = Arrays.asList(true, false, true, false, true, false, true, false, true);
+    List<Boolean> empty = Arrays.asList();
+    return Arrays.asList(new Object[][]{
+        {"True", "Test", someTrueSomeFalse, 5},
+        {"True", "Test", allTrue, 3},
+        {"True", "Test", allFalse, 0},
+        {"True", "Test", empty, 0},
+        {"False", "Test", someTrueSomeFalse, 4},
+        {"False", "Test", allTrue, 0},
+        {"False", "Test", allFalse, 3},
+        {"False", "Test", empty, 0},
+    });
+  }
+
+  @org.junit.runners.Parameterized.Parameter
+  public String boolToKeep;
+
+  @org.junit.runners.Parameterized.Parameter(1)
+  public String fieldName;
+
+  @org.junit.runners.Parameterized.Parameter(2)
+  public List<Boolean> eventBooleans;
+
+  @org.junit.runners.Parameterized.Parameter(3)
+  public int expectedFilteredBooleanCount;
+
+  @Test
+  public void testBoolenFilter() {
+    BooleanFilterProcessor bfp = new BooleanFilterProcessor();
+    DataProcessorDescription originalGraph = bfp.declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    graph.setInputStreams(Collections
+        .singletonList(EventStreamGenerator
+            .makeStreamWithProperties(Collections.singletonList(fieldName))));
+
+    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList(fieldName)));
+
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+        .setActualTopicName("output-topic");
+
+    graph.getStaticProperties().stream()
+        .filter(p -> p instanceof MappingPropertyUnary)
+        .map((p -> (MappingPropertyUnary) p))
+        // Must hardcode since BOOLEAN_MAPPING is private
+        .filter(p -> p.getInternalName().equals("boolean-mapping"))
+        .findFirst().get().setSelectedProperty("s0::" + fieldName);
+    ProcessorParams params = new ProcessorParams(graph);
+    params.extractor().getStaticPropertyByName(BooleanFilterProcessor.VALUE, OneOfStaticProperty.class).getOptions()
+        .stream().filter(ot -> ot.getName().equals(boolToKeep)).findFirst()
+        .get().setSelected(true);
+    StoreEventCollector collector = new StoreEventCollector();
+
+    bfp.onInvocation(params, collector, null);
+
+    int result = sendEvents(bfp, collector);
+
+    LOG.info("Expected filtered boolean count is {}", expectedFilteredBooleanCount);
+    LOG.info("Actual filtered boolean count is {}", result);
+    assertEquals(expectedFilteredBooleanCount, result);
+  }
+
+  private int sendEvents(BooleanFilterProcessor processor, StoreEventCollector collector) {
+    List<Event> events = makeEvents();
+    for (Event event : events) {
+      LOG.info("Sending event with value "
+          + event.getFieldBySelector("s0::" + fieldName).getAsPrimitive().getAsBoolean());
+      processor.onEvent(event, collector);
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+    return collector.getEvents().size();
+  }
+
+  private List<Event> makeEvents() {
+    List<Event> events = new ArrayList<>();
+    for (Boolean eventSetting : eventBooleans) {
+      events.add(makeEvent(eventSetting));
+    }
+    return events;
+  }
+
+  private Event makeEvent(Boolean value) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(fieldName, value);
+    return EventFactory.fromMap(map, new SourceInfo("test" + "-topic", "s0"),
+        new SchemaInfo(null, new ArrayList<>()));
+  }
+}
diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
new file mode 100644
index 0000000000..e2efbd6957
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/test/java/org/apache/streampipes/processors/filters/jvm/processor/merge/TestMergeByTimeProcessor.java
@@ -0,0 +1,193 @@
+/*
+ * 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.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.test.extensions.api.StoreEventCollector;
+import org.apache.streampipes.test.generator.EventStreamGenerator;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public class TestMergeByTimeProcessor {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestMergeByTimeProcessor.class);
+
+  private static final Integer timeInterval = 100;
+  @org.junit.runners.Parameterized.Parameter
+  public String testName;
+  @org.junit.runners.Parameterized.Parameter(1)
+  public List<String> eventStrings;
+  @org.junit.runners.Parameterized.Parameter(2)
+  public List<String> expectedValue;
+
+  @org.junit.runners.Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+    return Arrays.asList(new Object[][]{
+        {"testWithInInterval", Arrays.asList("s0:0", "s1:90"), List.of("(90,0)")},
+        {"testNotWithInInterval", Arrays.asList("s0:0", "s1:110"), List.of()},
+        {"testWithInAndNotWithInInterval", Arrays.asList("s0:0", "s1:80", "s0:110", "s1:500"),
+            List.of("(80,0)")},
+        {"testFigGvnInDocs",
+            Arrays.asList("s1:0", "s0:10", "s0:110", "s1:115", "s0:120", "s1:230", "s0:340", "s0:500",
+                "s1:510"),
+            Arrays.asList("(0,10)", "(115,110)", "(510,500)")}
+    });
+  }
+
+  @Test
+  public void testMergeByTimeProcessor() {
+    MergeByTimeProcessor mergeByTimeProcessor = new MergeByTimeProcessor();
+    DataProcessorDescription originalGraph = mergeByTimeProcessor.declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph = InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+    graph.setInputStreams(Arrays.asList(
+        EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("in-stram0")),
+        EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("in-stream1"))
+    ));
+
+    graph.setOutputStream(
+        EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("out-stream"))
+    );
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+        .setActualTopicName("output-topic");
+
+    List<String> outputKeySelectors = graph.getOutputStrategies()
+        .stream()
+        .filter(CustomOutputStrategy.class::isInstance)
+        .map(o -> (CustomOutputStrategy) o)
+        .findFirst()
+        .map(CustomOutputStrategy::getSelectedPropertyKeys)
+        .orElse(new ArrayList<>());
+    outputKeySelectors.add("s0::timestamp_mapping_stream_1");
+    outputKeySelectors.add("s1::timestamp_mapping_stream_2");
+
+    List<MappingPropertyUnary> mappingPropertyUnaries = graph.getStaticProperties()
+        .stream()
+        .filter(p -> p instanceof MappingPropertyUnary)
+        .map((p -> (MappingPropertyUnary) p))
+        .filter(p -> Arrays.asList(
+                MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY,
+                MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY)
+            .contains(p.getInternalName()))
+        .collect(Collectors.toList());
+
+    assert mappingPropertyUnaries.size() == 2;
+    mappingPropertyUnaries.get(0)
+        .setSelectedProperty("s0::" + MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY);
+    mappingPropertyUnaries.get(1)
+        .setSelectedProperty("s1::" + MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY);
+
+    FreeTextStaticProperty fsp = graph.getStaticProperties().stream()
+        .filter(p -> p instanceof FreeTextStaticProperty)
+        .map((p -> (FreeTextStaticProperty) p))
+        .filter(p -> p.getInternalName().equals(MergeByTimeProcessor.TIME_INTERVAL))
+        .findFirst().orElse(null);
+    assert fsp != null;
+    fsp.setValue(String.valueOf(timeInterval));
+
+    ProcessorParams params = new ProcessorParams(graph);
+
+    StoreEventCollector collector = new StoreEventCollector();
+
+    mergeByTimeProcessor.onInvocation(params, collector, null);
+    sendEvents(mergeByTimeProcessor, collector);
+
+    List<String> actualCollectedEvents = collector.getEvents().stream()
+        .map(e -> formatMergedEvent(e))
+        .collect(Collectors.toList());
+
+    LOG.info("Expected merged event is {}", expectedValue);
+    LOG.info("Actual merged event is {}", actualCollectedEvents);
+    assertTrue(eventsEquals(expectedValue, actualCollectedEvents));
+  }
+
+  private boolean eventsEquals(List<String> expectedValue, List<String> actualCollectedEvents) {
+    if (expectedValue.size() != actualCollectedEvents.size()) {
+      return false;
+    }
+    for (int i = 0; i < expectedValue.size(); i++) {
+      if (!expectedValue.get(i).equalsIgnoreCase(actualCollectedEvents.get(i))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private String formatMergedEvent(Event mergedEvent) {
+    return String.format("(%s)", mergedEvent.getFields().values().stream()
+        .map(m -> m.getAsPrimitive().getAsString()).collect(Collectors.joining(",")));
+  }
+
+  private void sendEvents(MergeByTimeProcessor mergeByTimeProcessor, StoreEventCollector spOut) {
+    List<Event> events = makeEvents();
+    for (Event event : events) {
+      mergeByTimeProcessor.onEvent(event, spOut);
+    }
+
+  }
+
+  private List<Event> makeEvents() {
+    List<Event> events = Lists.newArrayList();
+    for (String eventString : eventStrings) {
+      events.add(makeEvent(eventString));
+    }
+    return events;
+  }
+
+  private Event makeEvent(String eventString) {
+    Map<String, Object> map = Maps.newHashMap();
+    String streamId = eventString.split(":")[0];
+    String timestamp = eventString.split(":")[1];
+    if (streamId.equals("s0")) {
+      map.put(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_1_KEY, timestamp);
+    } else {
+      map.put(MergeByTimeProcessor.TIMESTAMP_MAPPING_STREAM_2_KEY, timestamp);
+    }
+    return EventFactory.fromMap(map,
+        new SourceInfo("test", streamId),
+        new SchemaInfo(null, Lists.newArrayList()));
+  }
+}
\ No newline at end of file
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/TestChangedValueDetectionProcessor.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/TestChangedValueDetectionProcessor.java
index 001991630f..f7c38a6c3a 100644
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/TestChangedValueDetectionProcessor.java
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/TestChangedValueDetectionProcessor.java
@@ -18,9 +18,6 @@
 
 package org.apache.streampipes.processors.transformation.jvm.processor.value.change;
 
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
-import org.apache.streampipes.messaging.InternalEventProcessor;
 import org.apache.streampipes.model.graph.DataProcessorInvocation;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.runtime.EventFactory;
@@ -31,6 +28,7 @@ import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
 import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
 import org.apache.streampipes.sdk.builder.adapter.GuessSchemaBuilder;
 import org.apache.streampipes.sdk.utils.Datatypes;
+import org.apache.streampipes.test.extensions.api.StoreEventCollector;
 import org.apache.streampipes.test.generator.InvocationGraphGenerator;
 import org.apache.streampipes.wrapper.params.compat.ProcessorParams;
 
@@ -38,7 +36,6 @@ import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
@@ -60,8 +57,8 @@ public class TestChangedValueDetectionProcessor {
 
     // Create event with no DIMENSION_PROPERTY
     event.getSchemaInfo()
-      .getEventSchema()
-      .getEventProperties().get(0)
+        .getEventSchema()
+        .getEventProperties().get(0)
         .setPropertyScope(PropertyScope.MEASUREMENT_PROPERTY.name());
 
     assertEquals("l1", processor.getDimensionKey(event));
@@ -87,9 +84,9 @@ public class TestChangedValueDetectionProcessor {
     DataProcessorInvocation graph = InvocationGraphGenerator.makeEmptyInvocation(processor.declareModel());
 
     graph.getStaticProperties().stream()
-    .filter(p -> p instanceof MappingPropertyUnary)
-    .map((p -> (MappingPropertyUnary) p))
-    .findFirst().get().setSelectedProperty("s0::value");
+        .filter(p -> p instanceof MappingPropertyUnary)
+        .map((p -> (MappingPropertyUnary) p))
+        .findFirst().get().setSelectedProperty("s0::value");
 
     ProcessorParams params = new ProcessorParams(graph);
     processor.onInvocation(params, null, null);
@@ -117,9 +114,9 @@ public class TestChangedValueDetectionProcessor {
     DataProcessorInvocation graph = InvocationGraphGenerator.makeEmptyInvocation(processor.declareModel());
 
     graph.getStaticProperties().stream()
-    .filter(p -> p instanceof MappingPropertyUnary)
-    .map((p -> (MappingPropertyUnary) p))
-    .findFirst().get().setSelectedProperty("s0::value");
+        .filter(p -> p instanceof MappingPropertyUnary)
+        .map((p -> (MappingPropertyUnary) p))
+        .findFirst().get().setSelectedProperty("s0::value");
 
     ProcessorParams params = new ProcessorParams(graph);
     processor.onInvocation(params, null, null);
@@ -170,31 +167,4 @@ public class TestChangedValueDetectionProcessor {
 
     return EventFactory.fromMap(map, new SourceInfo("", "s0"), new SchemaInfo(eventSchema, new ArrayList<>()));
   }
-
-  class StoreEventCollector implements SpOutputCollector {
-
-    ArrayList<Event> events = new ArrayList<Event>();
-
-    @Override
-    public void registerConsumer(String routeId, InternalEventProcessor<Map<String, Object>> consumer) {}
-
-    @Override
-    public void unregisterConsumer(String routeId) {}
-
-    @Override
-    public void connect() throws SpRuntimeException {}
-
-    @Override
-    public void disconnect() throws SpRuntimeException {}
-
-    @Override
-    public void collect(Event event) {
-      events.add(event);
-    }
-
-    public List<Event> getEvents() {
-      return this.events;
-    }
-
-  }
 }
diff --git a/streampipes-test-utils/pom.xml b/streampipes-test-utils/pom.xml
index 75c3a80d1a..de133e551b 100644
--- a/streampipes-test-utils/pom.xml
+++ b/streampipes-test-utils/pom.xml
@@ -41,6 +41,18 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-messaging</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-extensions-api</artifactId>
+            <version>0.93.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/extensions/api/StoreEventCollector.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/extensions/api/StoreEventCollector.java
new file mode 100644
index 0000000000..fd4080d8c9
--- /dev/null
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/extensions/api/StoreEventCollector.java
@@ -0,0 +1,57 @@
+/*
+ * 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.test.extensions.api;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
+import org.apache.streampipes.messaging.InternalEventProcessor;
+import org.apache.streampipes.model.runtime.Event;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class StoreEventCollector implements SpOutputCollector {
+  private ArrayList<Event> events = new ArrayList<Event>();
+
+  @Override
+  public void registerConsumer(String routeId, InternalEventProcessor<Map<String, Object>> consumer) {
+  }
+
+  @Override
+  public void unregisterConsumer(String routeId) {
+  }
+
+  @Override
+  public void connect() throws SpRuntimeException {
+  }
+
+  @Override
+  public void disconnect() throws SpRuntimeException {
+  }
+
+  @Override
+  public void collect(Event event) {
+    events.add(event);
+  }
+
+  public List<Event> getEvents() {
+    return this.events;
+  }
+}
\ No newline at end of file