Repository: apache/streampipes
Pull request #1092 (2023-01-13 11:04:18)

Reduce warnings at startup, improve structure of resource configs (#1…
…091)

<!--
  ~ 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 eliminates the WADL Feature warning on service startup and lets the core service's resource config inherit from the `BaseResourceConfig` to facilitate code reuse.

### 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/.github/workflows/pr-validation.yml b/.github/workflows/pr-validation.yml
index 07ace7096f..90bc06e310 100644
--- a/.github/workflows/pr-validation.yml
+++ b/.github/workflows/pr-validation.yml
@@ -131,12 +131,10 @@ jobs:
       - name: Run style & code checks
         run: |
           cd ./streampipes-client-python
-          pre-commit run --files streampipes_client/*
-          pre-commit run --files tests/*
+          pre-commit run -a
           make mypy
           make lint
 
-
   python-unit-tests:
     runs-on: ${{ matrix.os }}
     strategy:
@@ -171,4 +169,34 @@ jobs:
       - name: Run unit-tests
         run: |
           cd ./streampipes-client-python
-          make unit-tests
\ No newline at end of file
+          make unit-tests
+
+  python-doc-checks:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
+
+      - name: Cache Python dependencies
+        uses: actions/cache@v2
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip${{ hashFiles('**/setup.py') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install Python dependencies
+        run: |
+          cd ./streampipes-client-python
+          python -m pip install --upgrade pip
+          pip install wheel
+          pip install -e ".[docs]"
+
+      - name: Build MkDocs
+        run: |
+          cd ./streampipes-client-python
+          make doc
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 03949bef45..a3afd577c1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,7 @@
     <parent>
         <groupId>org.apache</groupId>
         <artifactId>apache</artifactId>
-        <version>25</version>
+        <version>29</version>
     </parent>
 
     <groupId>org.apache.streampipes</groupId>
@@ -151,7 +151,7 @@
 	<commons-math3.version>3.6.1</commons-math3.version>
 	<commons-compress.version>1.21</commons-compress.version>
 	<commons-text.version>1.10.0</commons-text.version>
-	<commons-pool2.version>2.9.0</commons-pool2.version>
+	<commons-pool2.version>2.11.1</commons-pool2.version>
 	<ditto-client.version>1.0.0</ditto-client.version>
 	<data-generator.version>1.4.1</data-generator.version>
 	<ddogleg.version>0.16</ddogleg.version>
@@ -1436,7 +1436,6 @@
         <module>streampipes-model</module>
         <module>streampipes-model-client</module>
         <module>streampipes-model-shared</module>
-        <module>streampipes-performance-tests</module>
         <module>streampipes-platform-services</module>
         <module>streampipes-pipeline-management</module>
         <module>streampipes-sources</module>
diff --git a/streampipes-client-python/mkdocs.yml b/streampipes-client-python/mkdocs.yml
index bc129cea5d..dbff3772b7 100644
--- a/streampipes-client-python/mkdocs.yml
+++ b/streampipes-client-python/mkdocs.yml
@@ -23,7 +23,7 @@ copyright: "Apache License 2.0"
 
 theme:
   name: material
-  logo: https://streampipes.apache.org/img/sp-logo-white-right.png
+  logo: https://raw.githubusercontent.com/apache/streampipes/dev/ui/deployment/dev/img/logo-navigation.png
   favicon: https://streampipes.apache.org/img/favicon/favicon.png
   custom_dir: docs/overrides
   features:
@@ -82,4 +82,5 @@ nav:
     - Examples:
         - Introduction to the StreamPipes Python Client: examples/1-introduction-to-streampipes-python-client.ipynb
         - Extracting Data from the StreamPipes Data Lake: examples/2-extracting-data-from-the-streampipes-data-lake.ipynb
-    - Reference: reference/*
\ No newline at end of file
+        - Exploring Live Data from a StreamPipes Data Stream: examples/3-getting-live-data-from-the-streampipes-data-stream.ipynb
+    - Reference: reference/*
diff --git a/streampipes-client-python/setup.py b/streampipes-client-python/setup.py
index 462dd3d87c..5fd1ff9722 100644
--- a/streampipes-client-python/setup.py
+++ b/streampipes-client-python/setup.py
@@ -38,30 +38,30 @@ base_packages = [
 ]
 
 dev_packages = base_packages + [
-    "autoflake>=1.7.7",
-    "black>=22.10.0",
-    "flake8>=5.0.4",
-    "interrogate>=1.5.0",
-    "isort>=5.10.1",
-    "mypy>=0.990",
-    "pandas-stubs>=1.2.0.62",
-    "pre-commit>=2.20.0",
-    "pytest>=7.2.0",
-    "pytest-cov>=4.0.0",
-    "pyupgrade>=3.2.2",
-    "types-requests>=2.28.11.4",
+    "autoflake==2.0.0",
+    "black==22.12.0",
+    "flake8==6.0.0",
+    "interrogate==1.5.0",
+    "isort==5.11.4",
+    "mypy==0.991",
+    "pandas-stubs==1.5.2.230105",
+    "pre-commit==2.21.0",
+    "pytest==7.2.1",
+    "pytest-cov==4.0.0",
+    "pyupgrade==3.3.1",
+    "types-requests==2.28.11.7",
 ]
 
 docs_packages = [
-    "mkdocs>=1.2.3",
-    "mkdocs-awesome-pages-plugin>=2.7.0",
-    "mkdocs-material>=8.1.11",
-    "mkdocstrings[python]>=0.19.0",
-    "pytkdocs[numpy-style]>=0.5.0",
-    "mkdocs-gen-files>=0.3.5",
-    "mkdocs-literate-nav>=0.4.1",
-    "numpydoc>=1.2",
-    "mkdocs-jupyter>=0.22.0 "
+    "mkdocs==1.4.2",
+    "mkdocs-awesome-pages-plugin==2.8.0",
+    "mkdocs-material==8.5.11",  # < 9.x.y is required by mkdocs-jupyter
+    "mkdocstrings[python]==0.19.1",
+    "pytkdocs[numpy-style]>=0.16.1",
+    "mkdocs-gen-files==0.4.0",
+    "mkdocs-literate-nav==0.6.0",
+    "numpydoc==1.5.0",
+    "mkdocs-jupyter==0.22.0 "
 ]
 
 here = os.path.abspath(os.path.dirname(__file__))
@@ -111,7 +111,6 @@ setuptools.setup(
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
-        "Programming Language :: Python :: 3.11",
         "Topic :: Scientific/Engineering"
     ],
     ext_modules=[],
diff --git a/streampipes-client-python/streampipes_client/client/client.py b/streampipes-client-python/streampipes_client/client/client.py
index 21c3c856f0..e22ed5fcd6 100644
--- a/streampipes-client-python/streampipes_client/client/client.py
+++ b/streampipes-client-python/streampipes_client/client/client.py
@@ -28,7 +28,6 @@ import sys
 from typing import Dict, Optional
 
 from requests import Session
-
 from streampipes_client.client.client_config import StreamPipesClientConfig
 from streampipes_client.endpoint import DataLakeMeasureEndpoint, DataStreamEndpoint
 from streampipes_client.endpoint.endpoint import APIEndpoint
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapter.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapter.java
index 5ad006b5c6..e6e8f91c4c 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapter.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapter.java
@@ -32,11 +32,10 @@ public interface IAdapter<T extends AdapterDescription> extends Connector {
 
   void stopAdapter() throws AdapterException;
 
-  IAdapter getInstance(T adapterDescription);
+  IAdapter<T> getInstance(T adapterDescription);
 
   GuessSchema getSchema(T adapterDescription) throws AdapterException, ParseException;
 
   void changeEventGrounding(TransportProtocol transportProtocol);
 
-  boolean isDebug();
 }
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapterPipeline.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapterPipeline.java
index f3434897c5..5739396c47 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapterPipeline.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IAdapterPipeline.java
@@ -17,6 +17,8 @@
  */
 package org.apache.streampipes.extensions.api.connect;
 
+import org.apache.streampipes.model.schema.EventSchema;
+
 import java.util.List;
 import java.util.Map;
 
@@ -31,4 +33,6 @@ public interface IAdapterPipeline {
   void changePipelineSink(IAdapterPipelineElement pipelineSink);
 
   IAdapterPipelineElement getPipelineSink();
+
+  EventSchema getResultingEventSchema();
 }
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
index 2d99825362..8b4a69ba04 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
@@ -21,7 +21,6 @@ import org.apache.streampipes.extensions.api.connect.exception.AdapterException;
 import org.apache.streampipes.extensions.api.connect.exception.ParseException;
 import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
-import org.apache.streampipes.model.schema.EventSchema;
 
 public interface IProtocol extends Connector {
 
@@ -42,6 +41,4 @@ public interface IProtocol extends Connector {
 
   String getId();
 
-  //TODO remove
-  void setEventSchema(EventSchema eventSchema);
 }
diff --git a/streampipes-extensions-management/pom.xml b/streampipes-extensions-management/pom.xml
index 883d30fb46..0b6c9e15d7 100644
--- a/streampipes-extensions-management/pom.xml
+++ b/streampipes-extensions-management/pom.xml
@@ -96,6 +96,12 @@
             <artifactId>streampipes-service-discovery</artifactId>
             <version>0.91.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.91.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-vocabulary</artifactId>
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/Adapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/Adapter.java
index 3abf928305..4d87918d6d 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/Adapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/Adapter.java
@@ -35,29 +35,16 @@ import com.google.common.annotations.VisibleForTesting;
 
 public abstract class Adapter<T extends AdapterDescription> implements IAdapter<T> {
 
-  private boolean debug;
-
   protected AdapterPipeline adapterPipeline;
 
   protected T adapterDescription;
 
-  public Adapter(T adapterDescription, boolean debug) {
+  public Adapter(T adapterDescription) {
     this.adapterDescription = adapterDescription;
-    this.debug = debug;
     this.adapterPipeline = getAdapterPipeline(adapterDescription);
   }
 
-  public Adapter(T adapterDescription) {
-    this(adapterDescription, false);
-  }
-
-  public Adapter(boolean debug) {
-    this.debug = debug;
-  }
-
-  public Adapter() {
-    this(false);
-  }
+  public Adapter() { }
 
   @Override
   public void changeEventGrounding(TransportProtocol transportProtocol) {
@@ -96,11 +83,6 @@ public abstract class Adapter<T extends AdapterDescription> implements IAdapter<
     return new AdapterPipelineGenerator().generatePipeline(adapterDescription);
   }
 
-  @Override
-  public boolean isDebug() {
-    return debug;
-  }
-
   @VisibleForTesting
   public AdapterPipeline getAdapterPipeline() {
     return adapterPipeline;
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
index 60d409a5e6..32a913f996 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/AdapterPipelineGenerator.java
@@ -71,20 +71,21 @@ public class AdapterPipelineGenerator {
     }
     pipelineElements.add(transformStreamAdapterElement);
 
-    // TODO decide what was meant with this comment
-    // Needed when adapter is (
     if (adapterDescription.getEventGrounding() != null
         && adapterDescription.getEventGrounding().getTransportProtocol() != null
         && adapterDescription.getEventGrounding().getTransportProtocol().getBrokerHostname() != null) {
-      return new AdapterPipeline(pipelineElements, getAdapterSink(adapterDescription));
+      return new AdapterPipeline(
+          pipelineElements,
+          getAdapterSink(adapterDescription),
+          adapterDescription.getEventSchema());
     }
 
     DebugSinkRuleDescription debugSinkRuleDescription = getDebugRule(adapterDescription.getRules());
     if (debugSinkRuleDescription != null) {
-      return new AdapterPipeline(pipelineElements, new DebugAdapterSink());
+      return new AdapterPipeline(pipelineElements, new DebugAdapterSink(), adapterDescription.getEventSchema());
     }
 
-    return new AdapterPipeline(pipelineElements);
+    return new AdapterPipeline(pipelineElements, adapterDescription.getEventSchema());
   }
 
   public List<IAdapterPipelineElement> makeAdapterPipelineElements(List<TransformationRuleDescription> rules) {
@@ -178,7 +179,7 @@ public class AdapterPipelineGenerator {
   }
 
   private boolean isPrioritized(SpProtocol prioritizedProtocol,
-                                      Class<?> protocolClass) {
+                                Class<?> protocolClass) {
     return prioritizedProtocol.getProtocolClass().equals(protocolClass.getCanonicalName());
   }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParser.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParser.java
index a0a68d9632..12724b38a4 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParser.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParser.java
@@ -108,11 +108,11 @@ public class CsvParser extends Parser {
     EventSchema resultSchema = new EventSchema();
     for (int i = 0; i < keys.length; i++) {
       EventPropertyPrimitive p = new EventPropertyPrimitive();
-      var runtimeType = DatatypeUtils.getXsdDatatype(data[i], false);
+      var runtimeType = DatatypeUtils.getXsdDatatype(data[i], true);
       var convertedValue = DatatypeUtils.convertValue(data[i], runtimeType);
       p.setRuntimeName(keys[i]);
       p.setRuntimeType(runtimeType);
-      sample.put(keys[i], new GuessTypeInfo(DatatypeUtils.getCanonicalTypeClassName(data[i], false), convertedValue));
+      sample.put(keys[i], new GuessTypeInfo(DatatypeUtils.getCanonicalTypeClassName(data[i], true), convertedValue));
       resultSchema.addEventProperty(p);
     }
 
@@ -165,11 +165,6 @@ public class CsvParser extends Parser {
 
           inQuotes = true;
 
-          //Fixed : allow "" in empty quote enclosed
-//                    if (chars[0] != '"' && customQuote == '\"') {
-//                        curVal.append('"');
-//                    }
-
           //double quotes in column will hit this!
           if (startCollectChar) {
             curVal.append('"');
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericAdapter.java
index 68e9591c4c..569cbcf7f2 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericAdapter.java
@@ -29,7 +29,6 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.adapter.GenericAdapterDescription;
 import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
-import org.apache.streampipes.model.schema.EventSchema;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,10 +42,6 @@ public abstract class GenericAdapter<T extends AdapterDescription> extends Adapt
     super(adapterDescription);
   }
 
-  public GenericAdapter(T adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
-
   public GenericAdapter() {
     super();
   }
@@ -63,15 +58,11 @@ public abstract class GenericAdapter<T extends AdapterDescription> extends Adapt
     IParser parser = getParser(adapterDescription);
     IFormat format = getFormat(adapterDescription);
 
-    ProtocolDescription protocolDescription = ((GenericAdapterDescription) adapterDescription).getProtocolDescription();
+    ProtocolDescription protocolDescription = (adapterDescription).getProtocolDescription();
 
     IProtocol protocolInstance = this.protocol.getInstance(protocolDescription, parser, format);
     this.protocol = protocolInstance;
 
-    //TODO remove
-    EventSchema eventSchema = adapterDescription.getEventSchema();
-    this.protocol.setEventSchema(eventSchema);
-
     logger.debug("Start adatper with format: " + format.getId() + " and " + protocol.getId());
 
     protocolInstance.run(adapterPipeline);
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataSetAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataSetAdapter.java
index 5a20453233..9902524974 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataSetAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataSetAdapter.java
@@ -23,24 +23,14 @@ import org.apache.streampipes.extensions.management.connect.adapter.Adapter;
 import org.apache.streampipes.model.connect.adapter.GenericAdapterDescription;
 import org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class GenericDataSetAdapter extends GenericAdapter<GenericAdapterSetDescription> {
 
   public static final String ID = GenericAdapterSetDescription.ID;
 
-  Logger logger = LoggerFactory.getLogger(Adapter.class);
-
   public GenericDataSetAdapter() {
     super();
   }
 
-
-  public GenericDataSetAdapter(GenericAdapterSetDescription adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
-
   public GenericDataSetAdapter(GenericAdapterSetDescription adapterDescription) {
     super(adapterDescription);
   }
@@ -48,8 +38,6 @@ public class GenericDataSetAdapter extends GenericAdapter<GenericAdapterSetDescr
   @Override
   public GenericAdapterSetDescription declareModel() {
     GenericAdapterSetDescription adapterDescription = new GenericAdapterSetDescription();
-//        adapterDescription.setAdapterId(GenericAdapterSetDescription.ID);
-//        adapterDescription.setUri(GenericAdapterSetDescription.ID);
     adapterDescription.setAppId(GenericAdapterSetDescription.ID);
     return adapterDescription;
   }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataStreamAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataStreamAdapter.java
index 6bf97463e5..ce5e85643e 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataStreamAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/GenericDataStreamAdapter.java
@@ -18,28 +18,19 @@
 
 package org.apache.streampipes.extensions.management.connect.adapter.model.generic;
 
+import org.apache.streampipes.extensions.api.connect.IAdapter;
 import org.apache.streampipes.extensions.api.connect.IProtocol;
-import org.apache.streampipes.extensions.management.connect.adapter.Adapter;
 import org.apache.streampipes.model.connect.adapter.GenericAdapterDescription;
 import org.apache.streampipes.model.connect.adapter.GenericAdapterStreamDescription;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class GenericDataStreamAdapter extends GenericAdapter<GenericAdapterStreamDescription> {
 
   public static final String ID = GenericAdapterStreamDescription.ID;
 
-  Logger logger = LoggerFactory.getLogger(Adapter.class);
-
   public GenericDataStreamAdapter() {
     super();
   }
 
-  public GenericDataStreamAdapter(GenericAdapterStreamDescription adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
-
   public GenericDataStreamAdapter(GenericAdapterStreamDescription adapterDescription) {
     super(adapterDescription);
   }
@@ -47,14 +38,12 @@ public class GenericDataStreamAdapter extends GenericAdapter<GenericAdapterStrea
   @Override
   public GenericAdapterStreamDescription declareModel() {
     GenericAdapterStreamDescription adapterDescription = new GenericAdapterStreamDescription();
-//        adapterDescription.setAdapterId(GenericAdapterStreamDescription.ID);
-//        adapterDescription.setUri(GenericAdapterStreamDescription.ID);
     adapterDescription.setAppId(GenericAdapterStreamDescription.ID);
     return adapterDescription;
   }
 
   @Override
-  public Adapter getInstance(GenericAdapterStreamDescription adapterDescription) {
+  public IAdapter<GenericAdapterStreamDescription> getInstance(GenericAdapterStreamDescription adapterDescription) {
     return new GenericDataStreamAdapter(adapterDescription);
   }
 
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/Protocol.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/Protocol.java
index 9b5a7870d2..d7d24ca4c5 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/Protocol.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/generic/Protocol.java
@@ -21,16 +21,12 @@ package org.apache.streampipes.extensions.management.connect.adapter.model.gener
 import org.apache.streampipes.extensions.api.connect.IFormat;
 import org.apache.streampipes.extensions.api.connect.IParser;
 import org.apache.streampipes.extensions.api.connect.IProtocol;
-import org.apache.streampipes.model.schema.EventSchema;
 
 public abstract class Protocol implements IProtocol {
 
   protected IParser parser;
   protected IFormat format;
 
-  //TODO remove
-  protected EventSchema eventSchema;
-
   public Protocol() {
 
   }
@@ -39,10 +35,4 @@ public abstract class Protocol implements IProtocol {
     this.parser = parser;
     this.format = format;
   }
-
-  //TODO remove
-  @Override
-  public void setEventSchema(EventSchema eventSchema) {
-    this.eventSchema = eventSchema;
-  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterEventPreviewPipeline.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterEventPreviewPipeline.java
index 3367d21551..17a70bfcd0 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterEventPreviewPipeline.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterEventPreviewPipeline.java
@@ -24,6 +24,7 @@ import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
 import org.apache.streampipes.extensions.management.connect.adapter.AdapterPipelineGenerator;
 import org.apache.streampipes.model.connect.guess.AdapterEventPreview;
 import org.apache.streampipes.model.connect.guess.GuessTypeInfo;
+import org.apache.streampipes.model.schema.EventSchema;
 
 import java.util.List;
 import java.util.Map;
@@ -79,4 +80,9 @@ public class AdapterEventPreviewPipeline implements IAdapterPipeline {
         .collect(Collectors.toMap(Map.Entry::getKey,
             e -> new GuessTypeInfo(e.getValue().getClass().getCanonicalName(), e.getValue())));
   }
+
+  @Override
+  public EventSchema getResultingEventSchema() {
+    return null;
+  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterPipeline.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterPipeline.java
index a055c4f3a6..7edd585ecb 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterPipeline.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/pipeline/AdapterPipeline.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.extensions.management.connect.adapter.model.pipel
 
 import org.apache.streampipes.extensions.api.connect.IAdapterPipeline;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
+import org.apache.streampipes.model.schema.EventSchema;
 
 import java.util.List;
 import java.util.Map;
@@ -29,25 +30,25 @@ public class AdapterPipeline implements IAdapterPipeline {
   private List<IAdapterPipelineElement> pipelineElements;
   private IAdapterPipelineElement pipelineSink;
 
+  private EventSchema resultingEventSchema;
 
-  public AdapterPipeline(List<IAdapterPipelineElement> pipelineElements) {
+  public AdapterPipeline(List<IAdapterPipelineElement> pipelineElements,
+                         EventSchema resultingEventSchema) {
     this.pipelineElements = pipelineElements;
+    this.resultingEventSchema = resultingEventSchema;
   }
 
-  public AdapterPipeline(List<IAdapterPipelineElement> pipelineElements, IAdapterPipelineElement pipelineSink) {
+  public AdapterPipeline(List<IAdapterPipelineElement> pipelineElements,
+                         IAdapterPipelineElement pipelineSink,
+                         EventSchema resultingEventSchema) {
     this.pipelineElements = pipelineElements;
     this.pipelineSink = pipelineSink;
+    this.resultingEventSchema = resultingEventSchema;
   }
 
   @Override
   public void process(Map<String, Object> event) {
 
-    // TODO remove, just for performance tests
-    if ("true".equals(System.getenv("SP_DEBUG_CONNECT"))) {
-      event.put("internal_t1", System.currentTimeMillis());
-    }
-
-
     for (IAdapterPipelineElement pipelineElement : pipelineElements) {
       event = pipelineElement.process(event);
     }
@@ -76,4 +77,9 @@ public class AdapterPipeline implements IAdapterPipeline {
   public IAdapterPipelineElement getPipelineSink() {
     return pipelineSink;
   }
+
+  @Override
+  public EventSchema getResultingEventSchema() {
+    return resultingEventSchema;
+  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificAdapter.java
index bb4f667eba..4311045c70 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificAdapter.java
@@ -31,7 +31,4 @@ public abstract class SpecificAdapter<T extends AdapterDescription> extends Adap
     super(adapterDescription);
   }
 
-  public SpecificAdapter(T adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataSetAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataSetAdapter.java
index 80b7f85c4f..8ae5b69d06 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataSetAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataSetAdapter.java
@@ -30,7 +30,4 @@ public abstract class SpecificDataSetAdapter extends SpecificAdapter<SpecificAda
     super(adapterDescription);
   }
 
-  public SpecificDataSetAdapter(SpecificAdapterSetDescription adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataStreamAdapter.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataStreamAdapter.java
index 6019ba2352..a55ed8c387 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataStreamAdapter.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/model/specific/SpecificDataStreamAdapter.java
@@ -30,8 +30,4 @@ public abstract class SpecificDataStreamAdapter extends SpecificAdapter<Specific
     super(adapterDescription);
   }
 
-  public SpecificDataStreamAdapter(SpecificAdapterStreamDescription adapterDescription, boolean debug) {
-    super(adapterDescription, debug);
-  }
-
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerReplayAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerReplayAdapterSink.java
deleted file mode 100644
index 812e7d61ec..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerReplayAdapterSink.java
+++ /dev/null
@@ -1,97 +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.extensions.management.connect.adapter.preprocessing.elements;
-
-import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.Util;
-
-import java.util.List;
-import java.util.Map;
-
-public class SendToBrokerReplayAdapterSink implements IAdapterPipelineElement {
-
-  private final SendToBrokerAdapterSink sendToBrokerAdapterSink;
-  private long lastEventTimestamp;
-  private final List<String> timestampKeys;
-  private final boolean replaceTimestamp;
-  private final float speedUp;
-
-
-  public SendToBrokerReplayAdapterSink(SendToBrokerAdapterSink sendToBrokerAdapterSink,
-                                       String timestampKey, boolean replaceTimestamp, float speedUp) {
-    this.sendToBrokerAdapterSink = sendToBrokerAdapterSink;
-    this.lastEventTimestamp = -1;
-    this.timestampKeys = Util.toKeyArray(timestampKey);
-    this.replaceTimestamp = replaceTimestamp;
-    this.speedUp = speedUp;
-  }
-
-  @Override
-  public Map<String, Object> process(Map<String, Object> event) {
-    if ((event != null) && (lastEventTimestamp != -1)) {
-      long actualEventTimestamp = getTimestampInEvent(event);
-      try {
-        if ((actualEventTimestamp - lastEventTimestamp) > 0) {
-          Thread.sleep((long) ((actualEventTimestamp - lastEventTimestamp) / speedUp));
-        }
-      } catch (InterruptedException e) {
-        e.printStackTrace();
-      }
-      if (replaceTimestamp) {
-        setTimestampInEvent(event, System.currentTimeMillis());
-      }
-      lastEventTimestamp = actualEventTimestamp;
-
-    } else if (lastEventTimestamp == -1) {
-      lastEventTimestamp = getTimestampInEvent(event);
-      if (replaceTimestamp) {
-        setTimestampInEvent(event, System.currentTimeMillis());
-      }
-    }
-    return sendToBrokerAdapterSink.process(event);
-  }
-
-  private long getTimestampInEvent(Map<String, Object> event) {
-    if (timestampKeys.size() == 1) {
-      try {
-        return (long) event.get(timestampKeys.get(0));
-      } catch (ClassCastException e) {
-        return lastEventTimestamp;
-      }
-    }
-    Map<String, Object> subEvent = event;
-    for (int i = 0; i < timestampKeys.size() - 1; i++) {
-      subEvent = (Map<String, Object>) subEvent.get(timestampKeys.get(i));
-    }
-    return (long) subEvent.get(timestampKeys.get(timestampKeys.size() - 1));
-
-  }
-
-  private void setTimestampInEvent(Map<String, Object> event, long timestamp) {
-    if (timestampKeys.size() == 1) {
-      event.put(timestampKeys.get(0), timestamp);
-    } else {
-      Map<String, Object> subEvent = event;
-      for (int i = 0; i < timestampKeys.size() - 1; i++) {
-        subEvent = (Map<String, Object>) subEvent.get(timestampKeys.get(i));
-      }
-      subEvent.put(timestampKeys.get(timestampKeys.size() - 1), timestamp);
-    }
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtils.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtils.java
index 20655f112d..4e1933ee0f 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtils.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtils.java
@@ -81,17 +81,29 @@ public class DatatypeUtils {
   }
 
   public static Class<?> getTypeClass(String value,
-                                      boolean preferFloat) {
+                                      boolean preferFloatingPointNumber) {
+    var targetClass = String.class;
     if (NumberUtils.isParsable(value)) {
+      Class<?> numberClass;
       try {
-        Integer.parseInt(value);
-        return preferFloat ? Float.class : Integer.class;
+        long longValue = Long.parseLong(value);
+        numberClass = longValue > Integer.MAX_VALUE ? Long.class : Integer.class;
+        if (preferFloatingPointNumber) {
+          return numberClass == Long.class ? Double.class : Float.class;
+        } else {
+          return numberClass;
+        }
       } catch (NumberFormatException ignored) {
       }
 
       try {
-        Long.parseLong(value);
-        return preferFloat ? Float.class : Long.class;
+        double doubleValue = Double.parseDouble(value);
+        numberClass = doubleValue > Float.MAX_VALUE ? Double.class : Float.class;
+        if (preferFloatingPointNumber) {
+          return numberClass == Double.class ? Double.class : Float.class;
+        } else {
+          return numberClass;
+        }
       } catch (NumberFormatException ignored) {
       }
 
@@ -107,13 +119,7 @@ public class DatatypeUtils {
       return Boolean.class;
     }
 
-    return String.class;
+    return targetClass;
   }
 
-  public static void main(String[] args) {
-    long max = Long.MAX_VALUE;
-    String className = getCanonicalTypeClassName(String.valueOf(max), true);
-    System.out.println(className);
-    System.out.println(convertValue(max, className));
-  }
 }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/EventSchemaUtils.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/EventSchemaUtils.java
new file mode 100644
index 0000000000..f0a7020447
--- /dev/null
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/util/EventSchemaUtils.java
@@ -0,0 +1,61 @@
+/*
+ * 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.extensions.management.util;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyNested;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.vocabulary.SO;
+
+import java.net.URI;
+import java.util.List;
+import java.util.Optional;
+
+public class EventSchemaUtils {
+
+  /**
+   * Returns the timestamp property of an event schema as an {@code Optional}.
+   *
+   * <p> The method checks all properties if they are of type {@code EventPropertyPrimitive} and if their domain
+   * properties contains the uri http://schema.org/DateTime </p>
+   *
+   * @param eventSchema the event schema for which the timestamp property is to be returned
+   * @return an {@code Optional} containing the timestamp property, or an empty {@code Optional} if
+   * no such property was found
+   */
+  public static Optional<EventPropertyPrimitive> getTimestampProperty(EventSchema eventSchema) {
+    return getTimstampProperty(eventSchema.getEventProperties());
+  }
+
+
+  private static Optional<EventPropertyPrimitive> getTimstampProperty(List<EventProperty> eventProperties) {
+    for (EventProperty ep : eventProperties) {
+      if (ep instanceof EventPropertyPrimitive && ep.getDomainProperties().contains(URI.create(SO.DATE_TIME))) {
+        return Optional.of((EventPropertyPrimitive) ep);
+      }
+
+      if (ep instanceof EventPropertyNested) {
+        return getTimstampProperty(((EventPropertyNested) ep).getEventProperties());
+      }
+    }
+
+    return Optional.empty();
+  }
+}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
index d47a0f4b44..4b588f882f 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
@@ -19,24 +19,18 @@
 package org.apache.streampipes.connect.iiot.protocol.stream;
 
 import org.apache.streampipes.connect.iiot.utils.FileProtocolUtils;
+import org.apache.streampipes.extensions.api.connect.EmitBinaryEvent;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipeline;
 import org.apache.streampipes.extensions.api.connect.IFormat;
 import org.apache.streampipes.extensions.api.connect.IParser;
+import org.apache.streampipes.extensions.api.connect.exception.AdapterException;
 import org.apache.streampipes.extensions.api.connect.exception.ParseException;
-import org.apache.streampipes.extensions.management.connect.SendToPipeline;
 import org.apache.streampipes.extensions.management.connect.adapter.guess.SchemaGuesser;
 import org.apache.streampipes.extensions.management.connect.adapter.model.generic.Protocol;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToBrokerReplayAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToJmsAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToKafkaAdapterSink;
-import org.apache.streampipes.extensions.management.connect.adapter.preprocessing.elements.SendToMqttAdapterSink;
+import org.apache.streampipes.extensions.management.util.EventSchemaUtils;
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventPropertyList;
-import org.apache.streampipes.model.schema.EventPropertyNested;
-import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.sdk.builder.adapter.ProtocolDescriptionBuilder;
 import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
@@ -53,12 +47,14 @@ import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class FileStreamProtocol extends Protocol {
 
-  private static Logger logger = LoggerFactory.getLogger(FileStreamProtocol.class);
+  private static final Logger logger = LoggerFactory.getLogger(FileStreamProtocol.class);
 
   public static final String ID = "org.apache.streampipes.connect.iiot.protocol.stream.file";
 
@@ -67,15 +63,17 @@ public class FileStreamProtocol extends Protocol {
   private float speedUp;
   private int timeBetweenReplay;
 
-  private Thread task;
-  private boolean running;
-
+  private ScheduledExecutorService executor;
 
   public FileStreamProtocol() {
   }
 
-  public FileStreamProtocol(IParser parser, IFormat format, String selectedFileName,
-                            boolean replaceTimestamp, float speedUp, int timeBetweenReplay) {
+  public FileStreamProtocol(IParser parser,
+                            IFormat format,
+                            String selectedFileName,
+                            boolean replaceTimestamp,
+                            float speedUp,
+                            int timeBetweenReplay) {
     super(parser, format);
     this.selectedFileName = selectedFileName;
     this.replaceTimestamp = replaceTimestamp;
@@ -84,66 +82,77 @@ public class FileStreamProtocol extends Protocol {
   }
 
   @Override
-  public void run(IAdapterPipeline adapterPipeline) {
-    String timestampKey = getTimestampKey(eventSchema.getEventProperties(), "");
-
-    // exchange adapter pipeline sink with special purpose replay sink for file replay
-    if (adapterPipeline.getPipelineSink() instanceof SendToKafkaAdapterSink) {
-      adapterPipeline.changePipelineSink(new SendToBrokerReplayAdapterSink(
-          (SendToKafkaAdapterSink) adapterPipeline.getPipelineSink(),
-          timestampKey,
-          replaceTimestamp,
-          speedUp));
-
-    } else if (adapterPipeline.getPipelineSink() instanceof SendToMqttAdapterSink) {
-      adapterPipeline.changePipelineSink(new SendToBrokerReplayAdapterSink(
-          (SendToMqttAdapterSink) adapterPipeline.getPipelineSink(),
-          timestampKey,
-          replaceTimestamp,
-          speedUp));
-
-    } else if (adapterPipeline.getPipelineSink() instanceof SendToJmsAdapterSink) {
-      adapterPipeline.changePipelineSink(new SendToBrokerReplayAdapterSink(
-          (SendToJmsAdapterSink) adapterPipeline.getPipelineSink(),
-          timestampKey,
-          replaceTimestamp,
-          speedUp));
+  public void run(IAdapterPipeline adapterPipeline) throws AdapterException {
+    String timestampKey = getTimestampKey(adapterPipeline.getResultingEventSchema());
+
+
+    executor = Executors.newScheduledThreadPool(1);
+    var eventProcessor = new LocalEventProcessor(adapterPipeline, timestampKey);
+
+    executor.scheduleAtFixedRate(() -> {
+      try (InputStream dataInputStream = getDataFromEndpoint()) {
+        format.reset();
+        parser.parse(dataInputStream, eventProcessor);
+      } catch (ParseException | IOException e) {
+        logger.error("Error while parsing: " + e.getMessage());
+      }
+    }, 0, timeBetweenReplay, TimeUnit.SECONDS);
+  }
+
+  private class LocalEventProcessor implements EmitBinaryEvent {
+
+    private final IAdapterPipeline adapterPipeline;
+    private final String timestampKey;
+
+    private long lastEventTimestamp = -1;
+
+    /**
+     * This local class is responsible to parse the events and set the timestamp accordign to the selected replay
+     * configurations
+     * @param adapterPipeline that transforms the events and send them to the message broker in the end
+     * @param timestampKey the runtimeName of the timestamp field
+     */
+    public LocalEventProcessor(IAdapterPipeline adapterPipeline,
+                               String timestampKey) {
+      this.adapterPipeline = adapterPipeline;
+      this.timestampKey = timestampKey;
+      format.reset();
     }
 
-    running = true;
-    task = new Thread() {
-      @Override
-      public void run() {
-        while (running) {
-
-          format.reset();
-          SendToPipeline stk = new SendToPipeline(format, adapterPipeline);
-          InputStream dataInputStream = getDataFromEndpoint();
-          try {
-            if (dataInputStream != null) {
-              parser.parse(dataInputStream, stk);
-            } else {
-              logger.warn("Could not read data from file.");
+    @Override
+    public Boolean emit(byte[] event) {
+
+      var eventMap = format.parse(event);
+      if (eventMap != null) {
+        long actualEventTimestamp = (long) eventMap.get(timestampKey);
+
+        if (lastEventTimestamp != -1) {
+          long sleepTime = (long) ((actualEventTimestamp - lastEventTimestamp) / speedUp);
+          if (sleepTime > 0) {
+            try {
+              Thread.sleep(sleepTime);
+            } catch (InterruptedException e) {
+              e.printStackTrace();
             }
-          } catch (ParseException e) {
-            logger.error("Error while parsing: " + e.getMessage());
           }
+        }
 
-          try {
-            Thread.sleep(timeBetweenReplay * 1000);
-          } catch (InterruptedException e) {
-            logger.error("Error while waiting for next replay round" + e.getMessage());
-          }
+        if (replaceTimestamp) {
+          eventMap.put(timestampKey, System.currentTimeMillis());
         }
+        lastEventTimestamp = actualEventTimestamp;
+
+        adapterPipeline.process(eventMap);
+
       }
-    };
-    task.start();
+      return true;
+    }
   }
 
-
   @Override
   public void stop() {
-    running = false;
+    executor.shutdownNow();
+    logger.info("Stopped file stream adapter for file " + selectedFileName);
   }
 
   private InputStream getDataFromEndpoint() throws ParseException {
@@ -172,29 +181,14 @@ public class FileStreamProtocol extends Protocol {
     return new FileStreamProtocol(parser, format, fileName, replaceTimestamp, speedUp, timeBetweenReplay);
   }
 
-  private String getTimestampKey(List<EventProperty> eventProperties, String prefixKey) {
-    String result = null;
-    for (EventProperty eventProperty : eventProperties) {
-      if (eventProperty instanceof EventPropertyPrimitive && eventProperty.getDomainProperties() != null) {
-        for (int i = eventProperty.getDomainProperties().size() - 1; i >= 0; i--) {
-          if (eventProperty.getDomainProperties().get(0).toString().equals("http://schema.org/DateTime")) {
-            result = prefixKey + eventProperty.getRuntimeName();
-          }
-        }
-      } else if (eventProperty instanceof EventPropertyNested
-          && ((EventPropertyNested) eventProperty).getEventProperties() != null) {
-        result = getTimestampKey(((EventPropertyNested) eventProperty).getEventProperties(),
-            prefixKey + eventProperty.getRuntimeName() + ".");
-      } else if (eventProperty instanceof EventPropertyList
-          && ((EventPropertyList) eventProperty).getEventProperty() != null) {
-        result = getTimestampKey(Arrays.asList(((EventPropertyList) eventProperty).getEventProperty()),
-            prefixKey + eventProperty.getRuntimeName() + ".");
-      }
-      if (result != null) {
-        return result;
-      }
+  private String getTimestampKey(EventSchema eventSchema) throws AdapterException {
+    var timestampProperty = EventSchemaUtils.getTimestampProperty(eventSchema);
+    if (timestampProperty.isPresent()) {
+      return timestampProperty.get().getRuntimeName();
+    } else {
+      throw new AdapterException("No timestamp present in event schema");
     }
-    return result;
+
   }
 
   @Override
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
index efbcb7e1c8..c685b709ea 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
@@ -30,6 +30,7 @@ import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 import org.apache.streampipes.model.grounding.MqttTransportProtocol;
 import org.apache.streampipes.model.grounding.SimpleTopicDefinition;
 import org.apache.streampipes.model.grounding.TransportProtocol;
+import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.model.shared.annotation.TsModel;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.model.util.Cloner;
@@ -252,4 +253,6 @@ public abstract class AdapterDescription extends NamedStreamPipesEntity {
   public void setCorrespondingDataStreamElementId(String correspondingDataStreamElementId) {
     this.correspondingDataStreamElementId = correspondingDataStreamElementId;
   }
+
+  public abstract EventSchema getEventSchema();
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterDescription.java
index 7468b79fa2..f7e2eb8a83 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterDescription.java
@@ -21,7 +21,6 @@ package org.apache.streampipes.model.connect.adapter;
 import org.apache.streampipes.model.connect.grounding.FormatDescription;
 import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
 import org.apache.streampipes.model.connect.rules.TransformationRuleDescription;
-import org.apache.streampipes.model.schema.EventSchema;
 
 import java.util.List;
 
@@ -32,8 +31,5 @@ public interface GenericAdapterDescription {
   FormatDescription getFormatDescription();
 
   List<TransformationRuleDescription> getRules();
-
-  EventSchema getEventSchema();
-
 }
 
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterSetDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterSetDescription.java
index 0779f7ae10..45e292bbfc 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterSetDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterSetDescription.java
@@ -30,8 +30,6 @@ public class GenericAdapterSetDescription extends AdapterSetDescription implemen
 
   public static final String ID = ElementIdGenerator.makeFixedElementId(GenericAdapterSetDescription.class);
 
-//    private String sourceType = "org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription";
-
   private FormatDescription formatDescription;
 
   private ProtocolDescription protocolDescription;
@@ -66,14 +64,6 @@ public class GenericAdapterSetDescription extends AdapterSetDescription implemen
     this.formatDescription = formatDescription;
   }
 
-  @Override
-  public EventSchema getEventSchema() {
-    if (this.getDataSet() != null) {
-      return this.getDataSet().getEventSchema();
-    }
-    return null;
-  }
-
   public ProtocolDescription getProtocolDescription() {
     return protocolDescription;
   }
@@ -82,11 +72,8 @@ public class GenericAdapterSetDescription extends AdapterSetDescription implemen
     this.protocolDescription = protocolDescription;
   }
 
-//    public String getSourceType() {
-//        return sourceType;
-//    }
-//
-//    public void setSourceType(String sourceType) {
-//        this.sourceType = sourceType;
-//    }
+  public EventSchema getEventSchema() {
+    return this.getDataSet().getEventSchema();
+  }
+
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterStreamDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterStreamDescription.java
index 67a88aa87a..b25cf6b5d4 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterStreamDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/GenericAdapterStreamDescription.java
@@ -63,14 +63,6 @@ public class GenericAdapterStreamDescription extends AdapterStreamDescription im
     this.formatDescription = formatDescription;
   }
 
-  @Override
-  public EventSchema getEventSchema() {
-    if (this.getDataStream() != null) {
-      return this.getDataStream().getEventSchema();
-    }
-    return null;
-  }
-
   public ProtocolDescription getProtocolDescription() {
     return protocolDescription;
   }
@@ -78,5 +70,7 @@ public class GenericAdapterStreamDescription extends AdapterStreamDescription im
   public void setProtocolDescription(ProtocolDescription protocolDescription) {
     this.protocolDescription = protocolDescription;
   }
-
+  public EventSchema getEventSchema() {
+    return this.getDataStream().getEventSchema();
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterSetDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterSetDescription.java
index 6052a85a37..d48aa3c86e 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterSetDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterSetDescription.java
@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.model.connect.adapter;
 
+import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.model.shared.annotation.TsModel;
 
 @TsModel
@@ -29,4 +30,8 @@ public class SpecificAdapterSetDescription extends AdapterSetDescription {
   public SpecificAdapterSetDescription(AdapterSetDescription other) {
     super(other);
   }
+
+  public EventSchema getEventSchema() {
+    return this.getDataSet().getEventSchema();
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterStreamDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterStreamDescription.java
index d5f2aa50fd..13759dbd23 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterStreamDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/SpecificAdapterStreamDescription.java
@@ -18,6 +18,7 @@
 
 package org.apache.streampipes.model.connect.adapter;
 
+import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.model.shared.annotation.TsModel;
 
 @TsModel
@@ -29,4 +30,8 @@ public class SpecificAdapterStreamDescription extends AdapterStreamDescription {
   public SpecificAdapterStreamDescription(AdapterStreamDescription other) {
     super(other);
   }
+
+  public EventSchema getEventSchema() {
+    return this.getDataStream().getEventSchema();
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/util/SchemaUtils.java b/streampipes-model/src/main/java/org/apache/streampipes/model/util/SchemaUtils.java
index 655037aea6..48738c1431 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/util/SchemaUtils.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/util/SchemaUtils.java
@@ -48,4 +48,5 @@ public class SchemaUtils {
     }
     return properties;
   }
+
 }
diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
index 03d2b9dfac..a59ce31ac5 100644
--- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
@@ -47,7 +47,6 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource {
   @JacksonSerialized
   @Produces(MediaType.APPLICATION_JSON)
   @Consumes(MediaType.APPLICATION_JSON)
-  @Path("/")
   public Response addDataLake(DataLakeMeasure dataLakeMeasure) {
     DataLakeMeasure result = this.dataLakeManagement.addDataLake(dataLakeMeasure);
     return ok(result);
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
index be3ee329a2..5cb0209860 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
@@ -75,7 +75,6 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
 
   @GET
   @Produces(MediaType.APPLICATION_JSON)
-  @Path("/")
   @JacksonSerialized
   @Operation(summary = "Get all pipelines of the current user",
       tags = {"Pipeline"},
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
index 76f0b16bf4..4cd90e99c6 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
@@ -55,7 +55,6 @@ public class DataStreamResource extends AbstractAuthGuardedRestResource {
   }
 
   @GET
-  @Path("/")
   @Produces({MediaType.APPLICATION_JSON, SpMediaType.JSONLD})
   @JacksonSerialized
   @PreAuthorize(AuthConstants.HAS_READ_PIPELINE_ELEMENT_PRIVILEGE)
diff --git a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/rest/BaseResourceConfig.java b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/rest/BaseResourceConfig.java
index 3eb8c92794..64c8f043bd 100644
--- a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/rest/BaseResourceConfig.java
+++ b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/rest/BaseResourceConfig.java
@@ -18,19 +18,25 @@
 package org.apache.streampipes.service.base.rest;
 
 import org.glassfish.jersey.server.ResourceConfig;
-import org.glassfish.jersey.servlet.ServletProperties;
+import org.glassfish.jersey.server.ServerProperties;
 
-import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
 
 public abstract class BaseResourceConfig extends ResourceConfig {
 
   public BaseResourceConfig() {
-    property(ServletProperties.FILTER_FORWARD_ON_404, true);
-    getClassesToRegister()
-        .forEach(set -> set.forEach(this::register));
-    register(ServiceHealthResource.class);
+    var customConfigs = new HashMap<String, Object>();
+    addAdditionalConfigs(customConfigs);
+    property(ServerProperties.WADL_FEATURE_DISABLE, true);
+    addProperties(customConfigs);
+    getClassesToRegister().forEach(this::register);
+
   }
 
-  public abstract List<List<Class<?>>> getClassesToRegister();
+  public abstract Set<Class<?>> getClassesToRegister();
+
+  public abstract void addAdditionalConfigs(Map<String, Object> configs);
 
 }
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java
index 67dd942736..1a4f253b7b 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesResourceConfig.java
@@ -83,104 +83,107 @@ import org.apache.streampipes.rest.impl.pe.DataProcessorResource;
 import org.apache.streampipes.rest.impl.pe.DataSinkResource;
 import org.apache.streampipes.rest.impl.pe.DataStreamResource;
 import org.apache.streampipes.rest.shared.serializer.JacksonSerializationProvider;
+import org.apache.streampipes.service.base.rest.BaseResourceConfig;
 import org.apache.streampipes.service.base.rest.ServiceHealthResource;
 
 import io.swagger.v3.jaxrs2.integration.resources.OpenApiResource;
 import org.glassfish.jersey.media.multipart.MultiPartFeature;
-import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.server.ServerProperties;
 import org.springframework.context.annotation.Configuration;
 
 import jakarta.ws.rs.ApplicationPath;
 
-import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
 
 @Configuration
 @ApplicationPath("/api")
-public class StreamPipesResourceConfig extends ResourceConfig {
+public class StreamPipesResourceConfig extends BaseResourceConfig {
 
-  public StreamPipesResourceConfig() {
-    setProperties(Collections.singletonMap("jersey.config.server.response.setStatusOverSendError", true));
-    register(AccountActivationResource.class);
-    register(AdapterMonitoringResource.class);
-    register(Authentication.class);
-    register(AssetDashboardResource.class);
-    register(AssetManagementResource.class);
-    register(AutoComplete.class);
-    register(CategoryResource.class);
-    register(ConsulConfig.class);
-    register(ContainerProvidedOptions.class);
-    register(DashboardWidget.class);
-    register(Dashboard.class);
-    register(DataExportResource.class);
-    register(DataImportResource.class);
-    register(DataLakeImageResource.class);
-    register(DataLakeResourceV3.class);
-    register(DataLakeMeasureResourceV3.class);
-    register(DataLakeMeasureResourceV4.class);
-    register(DataStream.class);
-    register(EmailConfigurationResource.class);
-    register(EmailResource.class);
-    register(ExtensionsServiceEndpointResource.class);
-    register(FunctionsResource.class);
-    register(GeneralConfigurationResource.class);
-    register(GenericStorageResource.class);
-    register(LabelResource.class);
-    register(MeasurementUnitResource.class);
-    register(Notification.class);
-    register(OntologyMeasurementUnit.class);
-    register(PermissionResource.class);
-    register(PersistedDataStreamResource.class);
-    register(PipelineCanvasMetadataCache.class);
-    register(PipelineCanvasMetadataResource.class);
-    register(PipelineCache.class);
-    register(PipelineCategory.class);
-    register(PipelineElementAsset.class);
-    register(PipelineElementCategory.class);
-    register(PipelineElementFile.class);
-    register(PipelineElementImport.class);
-    register(PipelineElementPreview.class);
-    register(PipelineElementRuntimeInfo.class);
-    register(PipelineMonitoring.class);
-    register(PipelineResource.class);
-    register(PipelineTemplate.class);
-    register(DataSinkResource.class);
-    register(DataProcessorResource.class);
-    register(DataStreamResource.class);
-    register(Setup.class);
-    register(ResetResource.class);
-    register(RestorePasswordResource.class);
-    register(ServiceHealthResource.class);
-    register(UserResource.class);
-    register(Version.class);
-    register(PipelineElementAsset.class);
-    register(DataLakeDashboardResource.class);
-    register(DataLakeWidgetResource.class);
-    register(DataLakeResourceV3.class);
-    register(PipelineElementFile.class);
-    register(DashboardWidget.class);
-    register(Dashboard.class);
-    register(VisualizablePipelineResource.class);
-    register(UserGroupResource.class);
+  @Override
+  public Set<Class<?>> getClassesToRegister() {
+    return Set.of(
+        AccountActivationResource.class,
+        AdapterMonitoringResource.class,
+        Authentication.class,
+        AssetDashboardResource.class,
+        AssetManagementResource.class,
+        AutoComplete.class,
+        CategoryResource.class,
+        ConsulConfig.class,
+        ContainerProvidedOptions.class,
+        DashboardWidget.class,
+        Dashboard.class,
+        DataExportResource.class,
+        DataImportResource.class,
+        DataLakeDashboardResource.class,
+        DataLakeWidgetResource.class,
+        DataLakeImageResource.class,
+        DataLakeResourceV3.class,
+        DataLakeMeasureResourceV3.class,
+        DataLakeMeasureResourceV4.class,
+        DataStream.class,
+        EmailConfigurationResource.class,
+        EmailResource.class,
+        ExtensionsServiceEndpointResource.class,
+        FunctionsResource.class,
+        GeneralConfigurationResource.class,
+        GenericStorageResource.class,
+        LabelResource.class,
+        MeasurementUnitResource.class,
+        Notification.class,
+        OntologyMeasurementUnit.class,
+        PermissionResource.class,
+        PersistedDataStreamResource.class,
+        PipelineCanvasMetadataCache.class,
+        PipelineCanvasMetadataResource.class,
+        PipelineCache.class,
+        PipelineCategory.class,
+        PipelineElementAsset.class,
+        PipelineElementCategory.class,
+        PipelineElementFile.class,
+        PipelineElementImport.class,
+        PipelineElementPreview.class,
+        PipelineElementRuntimeInfo.class,
+        PipelineMonitoring.class,
+        PipelineResource.class,
+        PipelineTemplate.class,
+        DataSinkResource.class,
+        DataProcessorResource.class,
+        DataStreamResource.class,
+        Setup.class,
+        ResetResource.class,
+        RestorePasswordResource.class,
+        ServiceHealthResource.class,
+        UserResource.class,
+        Version.class,
 
-    // Serializers
-    register(JacksonSerializationProvider.class);
-    register(MultiPartFeature.class);
+        VisualizablePipelineResource.class,
+        UserGroupResource.class,
 
-    // Platform Services
-    register(PipelineElementTemplateResource.class);
-    register(DataLakeResourceV4.class);
-    register(OpenApiResource.class);
+        // Serializers
+        JacksonSerializationProvider.class,
+        MultiPartFeature.class,
 
+        // Platform Services
+        PipelineElementTemplateResource.class,
+        DataLakeResourceV4.class,
+        OpenApiResource.class,
 
-    // Connect Master
-    register(AdapterResource.class);
-    register(DescriptionResource.class);
-    register(SourcesResource.class);
-    register(GuessResource.class);
-//    register(MultiPartFeature.class);
-    register(UnitResource.class);
-    register(WorkerAdministrationResource.class);
-    register(RuntimeResolvableResource.class);
+        // Connect Master
+        AdapterResource.class,
+        DescriptionResource.class,
+        SourcesResource.class,
+        GuessResource.class,
+
+        UnitResource.class,
+        WorkerAdministrationResource.class,
+        RuntimeResolvableResource.class
+    );
   }
 
+  @Override
+  public void addAdditionalConfigs(Map<String, Object> configs) {
+    configs.put(ServerProperties.RESPONSE_SET_STATUS_OVER_SEND_ERROR, true);
+  }
 }
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsResourceConfig.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsResourceConfig.java
index e03725d571..beaba53928 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsResourceConfig.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/ExtensionsResourceConfig.java
@@ -31,46 +31,42 @@ import org.apache.streampipes.rest.extensions.pe.DataStreamPipelineElementResour
 import org.apache.streampipes.rest.extensions.pe.PipelineTemplateResource;
 import org.apache.streampipes.rest.shared.serializer.JacksonSerializationProvider;
 import org.apache.streampipes.service.base.rest.BaseResourceConfig;
+import org.apache.streampipes.service.base.rest.ServiceHealthResource;
 
 import org.glassfish.jersey.media.multipart.MultiPartFeature;
+import org.glassfish.jersey.servlet.ServletProperties;
 import org.springframework.stereotype.Component;
 
-import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 @Component
 public class ExtensionsResourceConfig extends BaseResourceConfig {
 
   @Override
-  public List<List<Class<?>>> getClassesToRegister() {
-    return List.of(
-        getAdapterResourceClasses(),
-        getBaseResourceClasses(),
-        getPipelineElementResourceClasses()
-    );
-  }
-
-  private List<Class<?>> getAdapterResourceClasses() {
-    return List.of(
+  public Set<Class<?>> getClassesToRegister() {
+    return Set.of(
         GuessResource.class,
         RuntimeResolvableResource.class,
         AdapterWorkerResource.class,
         MultiPartFeature.class,
         AdapterAssetResource.class,
-        HttpServerAdapterResource.class);
-  }
-
-  private List<Class<?>> getBaseResourceClasses() {
-    return List.of(
-        JacksonSerializationProvider.class,
-        MonitoringResource.class);
-  }
+        HttpServerAdapterResource.class,
 
-  private List<Class<?>> getPipelineElementResourceClasses() {
-    return List.of(
         DataSinkPipelineElementResource.class,
         DataProcessorPipelineElementResource.class,
         DataStreamPipelineElementResource.class,
         WelcomePage.class,
-        PipelineTemplateResource.class);
+        PipelineTemplateResource.class,
+
+        ServiceHealthResource.class,
+        JacksonSerializationProvider.class,
+        MonitoringResource.class
+    );
+  }
+
+  @Override
+  public void addAdditionalConfigs(Map<String, Object> configs) {
+    configs.put(ServletProperties.FILTER_FORWARD_ON_404, true);
   }
 }
diff --git a/ui/.eslintignore b/ui/.eslintignore
index e55ff402d3..c980966c1a 100644
--- a/ui/.eslintignore
+++ b/ui/.eslintignore
@@ -21,8 +21,6 @@ cypress/fixtures/**/*.csv
 
 # Remove these in the future to lint additional modules
 # Please also see .prettierignore
-src/app/core-model
-src/app/core-services
 src/app/core-ui
 src/app/CustomMaterial
 src/app/dashboard
@@ -31,5 +29,3 @@ src/app/editor
 src/app/files
 src/app/info
 src/app/login
-src/app/notifications
-src/app/NS
\ No newline at end of file
diff --git a/ui/.prettierignore b/ui/.prettierignore
index 134205dda5..3930e91eeb 100644
--- a/ui/.prettierignore
+++ b/ui/.prettierignore
@@ -21,8 +21,6 @@ cypress/fixtures/**/*.csv
 
 # Remove these in the future to format additional modules
 # Please also see .eslintignore
-src/app/core-model
-src/app/core-services
 src/app/core-ui
 src/app/CustomMaterial
 src/app/dashboard
@@ -31,5 +29,3 @@ src/app/editor
 src/app/files
 src/app/info
 src/app/login
-src/app/notifications
-src/app/NS
\ No newline at end of file
diff --git a/ui/src/app/NS/XS.service.spec.ts b/ui/src/app/NS/XS.service.spec.ts
index aab2c38783..6db8f577a6 100644
--- a/ui/src/app/NS/XS.service.spec.ts
+++ b/ui/src/app/NS/XS.service.spec.ts
@@ -19,16 +19,16 @@
 import { XsService } from './xs.service';
 
 describe('XSService', () => {
-  const service: XsService = new XsService();
+    const service: XsService = new XsService();
 
-  it('should get string XS String', () => {
-    expect(service.XS_STRING).toBe('xs:string');
-  });
+    it('should get string XS String', () => {
+        expect(service.XS_STRING).toBe('xs:string');
+    });
 
-  it('should check for numbers', () => {
-    expect(service.isNumber(service.XS_DOUBLE)).toBeTrue();
-    expect(service.isNumber(service.XS_INTEGER)).toBeTrue();
-    expect(service.isNumber(service.XS_NUMBER)).toBeTrue();
-    expect(service.isNumber(service.XS_STRING)).toBeFalse();
-  });
+    it('should check for numbers', () => {
+        expect(service.isNumber(service.XS_DOUBLE)).toBeTrue();
+        expect(service.isNumber(service.XS_INTEGER)).toBeTrue();
+        expect(service.isNumber(service.XS_NUMBER)).toBeTrue();
+        expect(service.isNumber(service.XS_STRING)).toBeFalse();
+    });
 });
diff --git a/ui/src/app/NS/xs.service.ts b/ui/src/app/NS/xs.service.ts
index 77986e0d63..a8eea881e3 100644
--- a/ui/src/app/NS/xs.service.ts
+++ b/ui/src/app/NS/xs.service.ts
@@ -20,19 +20,19 @@ import { Injectable } from '@angular/core';
 
 @Injectable()
 export class XsService {
+    XS_STRING = 'xs:string';
+    XS_INTEGER = 'http://www.w3.org/2001/XMLSchema#integer';
+    XS_DOUBLE = 'http://www.w3.org/2001/XMLSchema#double';
+    XS_BOOLEAN = 'http://www.w3.org/2001/XMLSchema#boolean';
+    XS_NUMBER = 'http://www.w3.org/2001/XMLSchema#number';
+    XS_STRING1 = 'http://www.w3.org/2001/XMLSchema#string';
+    SO_URL = 'https://schema.org/URL';
 
-  XS_STRING = 'xs:string';
-  XS_INTEGER = 'http://www.w3.org/2001/XMLSchema#integer';
-  XS_DOUBLE = 'http://www.w3.org/2001/XMLSchema#double';
-  XS_BOOLEAN = 'http://www.w3.org/2001/XMLSchema#boolean';
-  XS_NUMBER = 'http://www.w3.org/2001/XMLSchema#number';
-  XS_STRING1 = 'http://www.w3.org/2001/XMLSchema#string';
-  SO_URL = 'https://schema.org/URL';
-
-  isNumber(datatype: string): boolean {
-    return datatype === this.XS_DOUBLE ||
-      datatype === this.XS_INTEGER ||
-      datatype === this.XS_NUMBER;
-  }
-
+    isNumber(datatype: string): boolean {
+        return (
+            datatype === this.XS_DOUBLE ||
+            datatype === this.XS_INTEGER ||
+            datatype === this.XS_NUMBER
+        );
+    }
 }
diff --git a/ui/src/app/connect/components/adapter-configuration/schema-editor/event-schema-preview/event-schema-preview.component.html b/ui/src/app/connect/components/adapter-configuration/schema-editor/event-schema-preview/event-schema-preview.component.html
index a55ad96cec..6bb8b51e36 100644
--- a/ui/src/app/connect/components/adapter-configuration/schema-editor/event-schema-preview/event-schema-preview.component.html
+++ b/ui/src/app/connect/components/adapter-configuration/schema-editor/event-schema-preview/event-schema-preview.component.html
@@ -22,6 +22,7 @@
             <pre
                 [innerHTML]="originalField | jsonpretty"
                 class="preview-text"
+                data-cy="schema-preview-original-event"
             ></pre>
         </sp-basic-inner-panel>
     </div>
diff --git a/ui/src/app/core-model/base/UserErrorMessage.ts b/ui/src/app/core-model/base/UserErrorMessage.ts
index 2fc7f5c72a..56bfc061a5 100644
--- a/ui/src/app/core-model/base/UserErrorMessage.ts
+++ b/ui/src/app/core-model/base/UserErrorMessage.ts
@@ -17,13 +17,13 @@
  */
 
 export class UserErrorMessage {
-  public title: string;
-  public content: string;
-  public level: string;
+    public title: string;
+    public content: string;
+    public level: string;
 
-  constructor(title: string, content: string, level = 'error') {
-    this.title = title;
-    this.content = content;
-    this.level = level;
-  }
+    constructor(title: string, content: string, level = 'error') {
+        this.title = title;
+        this.content = content;
+        this.level = level;
+    }
 }
diff --git a/ui/src/app/core-model/coco/Annotation.ts b/ui/src/app/core-model/coco/Annotation.ts
index 9a3d779aeb..9eefe6d4c8 100644
--- a/ui/src/app/core-model/coco/Annotation.ts
+++ b/ui/src/app/core-model/coco/Annotation.ts
@@ -17,27 +17,25 @@
  */
 
 export class Annotation {
-  id: number;
-  image_id: number;
-  category_id: number;
-  segmentation: [any]; // RLE or [polygon] -> polygon [[x1,y1,x2,y2,... xn, yn]]
-  area: number;
-  bbox: [number, number, number, number]; // [x,y,width,height]
-  iscrowd: number; // (iscrowd=0 in which case polygons are used) or a collection of objects (iscrowd=1 in which case RLE is used)
-  brushSize: number [];
+    id: number;
+    image_id: number;
+    category_id: number;
+    segmentation: [any]; // RLE or [polygon] -> polygon [[x1,y1,x2,y2,... xn, yn]]
+    area: number;
+    bbox: [number, number, number, number]; // [x,y,width,height]
+    iscrowd: number; // (iscrowd=0 in which case polygons are used) or a collection of objects (iscrowd=1 in which case RLE is used)
+    brushSize: number[];
 
+    // For UI
+    isSelected = false;
+    isHovered = false;
+    category_name: string;
+    color: string;
+    label_name: string;
 
-  // For UI
-  isSelected = false;
-  isHovered = false;
-  category_name: string;
-  color: string;
-  label_name: string;
-
-  constructor() {
-    this.segmentation = undefined;
-    this.bbox = undefined;
-    this.brushSize = undefined;
-  }
-
+    constructor() {
+        this.segmentation = undefined;
+        this.bbox = undefined;
+        this.brushSize = undefined;
+    }
 }
diff --git a/ui/src/app/core-model/coco/Category.ts b/ui/src/app/core-model/coco/Category.ts
index f066776224..1578ed5359 100644
--- a/ui/src/app/core-model/coco/Category.ts
+++ b/ui/src/app/core-model/coco/Category.ts
@@ -17,16 +17,20 @@
  */
 
 export class Category {
+    constructor(
+        id: number,
+        name: string,
+        supercategory: string,
+        label_name: string,
+    ) {
+        this.id = id;
+        this.name = name;
+        this.supercategory = supercategory;
+        this.label_name = label_name;
+    }
 
-  constructor(id: number, name: string, supercategory: string, label_name: string) {
-    this.id = id;
-    this.name = name;
-    this.supercategory = supercategory;
-    this.label_name = label_name;
-  }
-
-  id: number;
-  name: string;
-  supercategory: string;
-  label_name: string;
+    id: number;
+    name: string;
+    supercategory: string;
+    label_name: string;
 }
diff --git a/ui/src/app/core-model/coco/Coco.format.ts b/ui/src/app/core-model/coco/Coco.format.ts
index 7b9da43197..e5d720debb 100644
--- a/ui/src/app/core-model/coco/Coco.format.ts
+++ b/ui/src/app/core-model/coco/Coco.format.ts
@@ -21,20 +21,18 @@ import { Category } from './Category';
 import { Image } from './Image';
 
 export class CocoFormat {
+    info: {
+        year: number;
+        version: string;
+        description: string;
+        contributor: string;
+        url: string;
+        date_created: Date;
+    };
+    licenses: [];
+    images: Image[] = [];
+    annotations: Annotation[] = [];
+    categories: Category[] = [];
 
-  info: {
-    'year': number,
-    'version': string,
-    'description': string,
-    'contributor': string,
-    'url': string,
-    'date_created': Date,
-  };
-  licenses: [];
-  images: Image[] = [];
-  annotations: Annotation[] = [];
-  categories: Category[] = [];
-
-  constructor() { }
-
+    constructor() {}
 }
diff --git a/ui/src/app/core-model/coco/Image.ts b/ui/src/app/core-model/coco/Image.ts
index 51aeee3953..05bf61a248 100644
--- a/ui/src/app/core-model/coco/Image.ts
+++ b/ui/src/app/core-model/coco/Image.ts
@@ -17,12 +17,12 @@
  */
 
 export class Image {
-  id: number;
-  width: number;
-  height: number;
-  file_name: Date;
-  license: number;
-  flickr_url: Date;
-  coco_url: Date;
-  date_captured: Date;
+    id: number;
+    width: number;
+    height: number;
+    file_name: Date;
+    license: number;
+    flickr_url: Date;
+    coco_url: Date;
+    date_captured: Date;
 }
diff --git a/ui/src/app/core-services/semantic-type/semantic-type-utils.service.ts b/ui/src/app/core-services/semantic-type/semantic-type-utils.service.ts
index ca4020cb3f..a56a3f48c7 100644
--- a/ui/src/app/core-services/semantic-type/semantic-type-utils.service.ts
+++ b/ui/src/app/core-services/semantic-type/semantic-type-utils.service.ts
@@ -21,24 +21,22 @@ import { EventProperty } from '@streampipes/platform-services';
 
 @Injectable()
 export class SemanticTypeUtilsService {
-
     public TIMESTAMP = 'http://schema.org/DateTime';
     public IMAGE = 'https://image.com';
     public NUMBER = 'http://schema.org/Number';
 
-    constructor() {
-    }
+    constructor() {}
 
     public getValue(inputValue, semanticType) {
         if (semanticType === this.TIMESTAMP) {
-            return new Date(inputValue).toLocaleString() ;
+            return new Date(inputValue).toLocaleString();
         } else {
             return inputValue;
         }
     }
 
     public isTimestamp(property: EventProperty): boolean {
-       return property.domainProperties.includes(this.TIMESTAMP);
+        return property.domainProperties.includes(this.TIMESTAMP);
     }
 
     public is(property: EventProperty, uri: string): boolean {
diff --git a/ui/src/app/core-services/template/PipelineInvocationBuilder.ts b/ui/src/app/core-services/template/PipelineInvocationBuilder.ts
index cec4cf053b..03715cf48a 100644
--- a/ui/src/app/core-services/template/PipelineInvocationBuilder.ts
+++ b/ui/src/app/core-services/template/PipelineInvocationBuilder.ts
@@ -15,54 +15,62 @@
  * limitations under the License.
  *
  */
-import { FreeTextStaticProperty, MappingPropertyUnary, PipelineTemplateInvocation } from '@streampipes/platform-services';
+import {
+    FreeTextStaticProperty,
+    MappingPropertyUnary,
+    PipelineTemplateInvocation,
+} from '@streampipes/platform-services';
 
 export class PipelineInvocationBuilder {
+    private pipelineTemplateInvocation: PipelineTemplateInvocation;
 
-  private pipelineTemplateInvocation: PipelineTemplateInvocation;
+    constructor(pipelineTemplateInvocation: PipelineTemplateInvocation) {
+        this.pipelineTemplateInvocation = pipelineTemplateInvocation;
+    }
 
-  constructor(pipelineTemplateInvocation: PipelineTemplateInvocation) {
-    this.pipelineTemplateInvocation = pipelineTemplateInvocation;
-  }
+    public static create(
+        pipelineTemplateInvocation: PipelineTemplateInvocation,
+    ) {
+        return new PipelineInvocationBuilder(pipelineTemplateInvocation);
+    }
 
-  public static create(pipelineTemplateInvocation: PipelineTemplateInvocation) {
-    return new PipelineInvocationBuilder(pipelineTemplateInvocation);
-  }
+    public setTemplateId(id: string) {
+        this.pipelineTemplateInvocation.pipelineTemplateId = id;
+        return this;
+    }
 
-  public setTemplateId(id: string) {
-    this.pipelineTemplateInvocation.pipelineTemplateId = id;
-    return this;
-  }
+    public setName(name: string) {
+        this.pipelineTemplateInvocation.kviName = name;
+        return this;
+    }
 
-  public setName(name: string) {
-    this.pipelineTemplateInvocation.kviName = name;
-    return this;
-  }
+    public setFreeTextStaticProperty(name: string, value: string) {
+        this.pipelineTemplateInvocation.staticProperties.forEach(property => {
+            if (
+                property instanceof FreeTextStaticProperty &&
+                'jsplumb_domId2' + name === property.internalName
+            ) {
+                property.value = value;
+            }
+        });
 
-  public setFreeTextStaticProperty(name: string, value: string) {
-    this.pipelineTemplateInvocation.staticProperties.forEach(property => {
-      if (property instanceof FreeTextStaticProperty && 'jsplumb_domId2' + name === property.internalName) {
-        property.value = value;
-      }
-    });
+        return this;
+    }
 
-    return this;
-  }
-
-  public setMappingPropertyUnary(name: string, value: string) {
-    this.pipelineTemplateInvocation.staticProperties.forEach(property => {
-      if (property instanceof MappingPropertyUnary && 'jsplumb_domId2' + name === property.internalName) {
-        property.selectedProperty = value;
-      }
-    });
-
-    return this;
-  }
-
-  public build() {
-    return this.pipelineTemplateInvocation;
-  }
+    public setMappingPropertyUnary(name: string, value: string) {
+        this.pipelineTemplateInvocation.staticProperties.forEach(property => {
+            if (
+                property instanceof MappingPropertyUnary &&
+                'jsplumb_domId2' + name === property.internalName
+            ) {
+                property.selectedProperty = value;
+            }
+        });
 
+        return this;
+    }
 
+    public build() {
+        return this.pipelineTemplateInvocation;
+    }
 }
-
diff --git a/ui/src/app/notifications/components/notification-item.component.html b/ui/src/app/notifications/components/notification-item.component.html
index 1f8786d943..6b7cbc151a 100644
--- a/ui/src/app/notifications/components/notification-item.component.html
+++ b/ui/src/app/notifications/components/notification-item.component.html
@@ -18,11 +18,12 @@
 <div class="notification-item">
     <div class="i">
         <div class="head">
-            <span class="time">{{notification.createdAt | date: 'yyyy-MM-dd HH:mm:ss'}}</span>
+            <span class="time">{{
+                notification.createdAt | date : 'yyyy-MM-dd HH:mm:ss'
+            }}</span>
         </div>
         <div class="notification-box sp-accent-bg">
             <div [innerHTML]="sanitizedMessage"></div>
         </div>
     </div>
-
 </div>
diff --git a/ui/src/app/notifications/components/notification-item.component.scss b/ui/src/app/notifications/components/notification-item.component.scss
index bbb3b9270d..b74594c201 100644
--- a/ui/src/app/notifications/components/notification-item.component.scss
+++ b/ui/src/app/notifications/components/notification-item.component.scss
@@ -52,4 +52,4 @@
 
 .notification-item {
     margin-top: 30px;
-}
\ No newline at end of file
+}
diff --git a/ui/src/app/notifications/components/notification-item.component.ts b/ui/src/app/notifications/components/notification-item.component.ts
index b363383bf2..5cf6d949a1 100644
--- a/ui/src/app/notifications/components/notification-item.component.ts
+++ b/ui/src/app/notifications/components/notification-item.component.ts
@@ -20,20 +20,19 @@ import { NotificationItem } from '../model/notifications.model';
 import { DomSanitizer } from '@angular/platform-browser';
 
 @Component({
-    selector: 'notification-item',
+    selector: 'sp-notification-item',
     templateUrl: './notification-item.component.html',
-    styleUrls: ['./notification-item.component.scss']
+    styleUrls: ['./notification-item.component.scss'],
 })
 export class NotificationItemComponent implements OnInit {
-
     @Input() notification: NotificationItem;
     sanitizedMessage;
 
-    constructor(private domSanitizer: DomSanitizer) {
-
-    }
+    constructor(private domSanitizer: DomSanitizer) {}
 
     ngOnInit(): void {
-        this.sanitizedMessage = this.domSanitizer.bypassSecurityTrustHtml(this.notification.message);
+        this.sanitizedMessage = this.domSanitizer.bypassSecurityTrustHtml(
+            this.notification.message,
+        );
     }
 }
diff --git a/ui/src/app/notifications/notifications.component.html b/ui/src/app/notifications/notifications.component.html
index e91ac59452..cafa23a3ea 100644
--- a/ui/src/app/notifications/notifications.component.html
+++ b/ui/src/app/notifications/notifications.component.html
@@ -17,7 +17,7 @@
   -->
 
 <div fxLayout="column" class="page-container page-container-max-height">
-    <div fxLayout="row" style="padding:0px;" class="sp-tab-bg">
+    <div fxLayout="row" style="padding: 0px" class="sp-tab-bg">
         <div fxFlex="100" class="page-container-nav">
             <div fxFlex="100" fxLayout="row">
                 <div fxFlex fxLayoutAlign="start center">
@@ -25,51 +25,108 @@
                         <mat-tab label="My Notifications"></mat-tab>
                     </mat-tab-group>
                 </div>
-                <div fxFlex fxLayoutAlign="end center" class="mr-20">
-
-                </div>
+                <div fxFlex fxLayoutAlign="end center" class="mr-20"></div>
             </div>
         </div>
     </div>
-    <div class="container-fluid marketplace-container" *ngIf="notificationsLoading" fxFlex="100" fxLayoutAlign="center center" fxLayout="column">
+    <div
+        class="container-fluid marketplace-container"
+        *ngIf="notificationsLoading"
+        fxFlex="100"
+        fxLayoutAlign="center center"
+        fxLayout="column"
+    >
         <mat-spinner [mode]="'indeterminate'" [diameter]="20"></mat-spinner>
         <h4>Loading notifications...</h4>
     </div>
-    <div class="fixed-height page-container-padding-inner" *ngIf="!notificationsLoading && !pipelinesWithNotificationsPresent" fxFlex="100" fxLayoutAlign="center center" fxLayout="row">
-        <h4>No notifications available. Create a new pipeline using the Notification Sink to create your first notification!</h4>
+    <div
+        class="fixed-height page-container-padding-inner"
+        *ngIf="!notificationsLoading && !pipelinesWithNotificationsPresent"
+        fxFlex="100"
+        fxLayoutAlign="center center"
+        fxLayout="row"
+    >
+        <h4>
+            No notifications available. Create a new pipeline using the
+            Notification Sink to create your first notification!
+        </h4>
     </div>
-    <div class="fixed-height page-container-padding-inner" fxLayout="row" fxFlex="100" *ngIf="!notificationsLoading && pipelinesWithNotificationsPresent">
+    <div
+        class="fixed-height page-container-padding-inner"
+        fxLayout="row"
+        fxFlex="100"
+        *ngIf="!notificationsLoading && pipelinesWithNotificationsPresent"
+    >
         <div fxFlex="30" class="notifications-overview scrolling-auto">
             <mat-list>
-                <mat-list-item *ngFor="let existingNotification of existingNotifications"
-                               (click)="selectNotification(existingNotification)" class="list-item"
-                               [ngClass]="{'selected-notification':
-                               existingNotification.notificationId === currentlySelectedNotificationId}">
-                    <div mat-list-avatar
-                         class="notification-avatar sp-accent-bg">{{elementIconText.getElementIconText(existingNotification.pipelineName)}}
+                <mat-list-item
+                    *ngFor="let existingNotification of existingNotifications"
+                    (click)="selectNotification(existingNotification)"
+                    class="list-item"
+                    [ngClass]="{
+                        'selected-notification':
+                            existingNotification.notificationId ===
+                            currentlySelectedNotificationId
+                    }"
+                >
+                    <div
+                        mat-list-avatar
+                        class="notification-avatar sp-accent-bg"
+                    >
+                        {{
+                            elementIconText.getElementIconText(
+                                existingNotification.pipelineName
+                            )
+                        }}
                     </div>
-                    <h4 mat-line>{{existingNotification.pipelineName}}</h4>
-                    <p mat-line>{{existingNotification.notificationTitle}} </p>
+                    <h4 mat-line>{{ existingNotification.pipelineName }}</h4>
+                    <p mat-line>{{ existingNotification.notificationTitle }}</p>
                     <div class="new-notification-info-panel">
-                        <div class="new-notification-info"
-                             *ngIf="currentlySelectedNotificationId != existingNotification.notificationId &&
-                             newNotificationInfo[existingNotification.notificationId]"></div>
+                        <div
+                            class="new-notification-info"
+                            *ngIf="
+                                currentlySelectedNotificationId !==
+                                    existingNotification.notificationId &&
+                                newNotificationInfo[
+                                    existingNotification.notificationId
+                                ]
+                            "
+                        ></div>
                     </div>
                 </mat-list-item>
             </mat-list>
         </div>
         <div fxFlex="70" class="notifications-details">
             <div class="notification-details-wrapper">
-                <div class="notification-header" fxLayout="column" fxLayoutAlign="center start">
-                    <div class="notification-header-pipeline-name">{{currentlySelectedNotification.pipelineName}}</div>
-                    <div class="notification-header-notification-name">{{currentlySelectedNotification.notificationTitle}}</div>
-                    <hr class="header-divider"/>
+                <div
+                    class="notification-header"
+                    fxLayout="column"
+                    fxLayoutAlign="center start"
+                >
+                    <div class="notification-header-pipeline-name">
+                        {{ currentlySelectedNotification.pipelineName }}
+                    </div>
+                    <div class="notification-header-notification-name">
+                        {{ currentlySelectedNotification.notificationTitle }}
+                    </div>
+                    <hr class="header-divider" />
                 </div>
-                <div #notificationPane class="notification-pane" (scroll)="onScroll()" *ngIf="currentlySelectedNotificationId">
-                    <notification-item [notification]="notification" *ngFor="let notification of allNotifications.get(currentlySelectedNotificationId)"></notification-item>
+                <div
+                    #notificationPane
+                    class="notification-pane"
+                    (scroll)="onScroll()"
+                    *ngIf="currentlySelectedNotificationId"
+                >
+                    <sp-notification-item
+                        [notification]="notification"
+                        *ngFor="
+                            let notification of allNotifications.get(
+                                currentlySelectedNotificationId
+                            )
+                        "
+                    ></sp-notification-item>
                 </div>
             </div>
-
         </div>
     </div>
 </div>
diff --git a/ui/src/app/notifications/notifications.component.scss b/ui/src/app/notifications/notifications.component.scss
index 079936714e..4e2ec2c606 100644
--- a/ui/src/app/notifications/notifications.component.scss
+++ b/ui/src/app/notifications/notifications.component.scss
@@ -32,7 +32,7 @@
 }
 
 .mr-20 {
-    margin-right:20px;
+    margin-right: 20px;
 }
 
 .notifications-overview {
@@ -60,11 +60,11 @@
 }
 
 .header-divider {
-    display:flex;
+    display: flex;
     align-self: center;
     margin: 10px 20px;
     width: 100%;
-    border-top: 2px solid $sp-color-accent
+    border-top: 2px solid $sp-color-accent;
 }
 
 .notification-header-pipeline-name {
@@ -83,17 +83,17 @@
 }
 
 .list-item:hover {
-    background: #E0E0E0;
+    background: #e0e0e0;
     cursor: pointer;
 }
 
 .selected-notification {
-    background: #E0E0E0;
+    background: #e0e0e0;
     cursor: pointer;
 }
 
 .list-item {
-    border-bottom: 1px solid #BDBDBD;
+    border-bottom: 1px solid #bdbdbd;
 }
 
 .notification-pane {
@@ -111,7 +111,7 @@
 
 .scrolling-auto {
     overflow: auto;
-    width:100%;
+    width: 100%;
 }
 
 .new-notification-info {
@@ -120,4 +120,3 @@
     height: 10px;
     border-radius: 50%;
 }
-
diff --git a/ui/src/app/notifications/notifications.component.ts b/ui/src/app/notifications/notifications.component.ts
index 4d31082e13..515885c953 100644
--- a/ui/src/app/notifications/notifications.component.ts
+++ b/ui/src/app/notifications/notifications.component.ts
@@ -16,211 +16,294 @@
  *
  */
 
-import { Component, ElementRef, OnDestroy, OnInit, ViewChild } from '@angular/core';
-import { ExistingNotification, NotificationItem } from './model/notifications.model';
+import {
+    Component,
+    ElementRef,
+    OnDestroy,
+    OnInit,
+    ViewChild,
+} from '@angular/core';
+import {
+    ExistingNotification,
+    NotificationItem,
+} from './model/notifications.model';
 import { ElementIconText } from '../services/get-element-icon-text.service';
 import { NotificationsService } from './service/notifications.service';
 import { Subscription, timer } from 'rxjs';
 import { NotificationUtils } from './utils/notifications.utils';
 import { NotificationCountService } from '../services/notification-count-service';
-import { FreeTextStaticProperty, Pipeline, PipelineService } from '@streampipes/platform-services';
+import {
+    FreeTextStaticProperty,
+    Pipeline,
+    PipelineService,
+} from '@streampipes/platform-services';
 import { AuthService } from '../services/auth.service';
 import { filter, switchMap } from 'rxjs/operators';
 import { SpBreadcrumbService } from '@streampipes/shared-ui';
 
 @Component({
-  selector: 'notifications',
-  templateUrl: './notifications.component.html',
-  styleUrls: ['./notifications.component.scss']
+    selector: 'sp-notifications',
+    templateUrl: './notifications.component.html',
+    styleUrls: ['./notifications.component.scss'],
 })
 export class NotificationsComponent implements OnInit, OnDestroy {
+    static readonly NOTIFICATIONS_APP_ID =
+        'org.apache.streampipes.sinks.internal.jvm.notification';
+    static readonly NOTIFICATION_TITLE_KEY = 'title';
 
-  static readonly NOTIFICATIONS_APP_ID = 'org.apache.streampipes.sinks.internal.jvm.notification';
-  static readonly NOTIFICATION_TITLE_KEY = 'title';
-
-  @ViewChild('notificationPane') private notificationContainer: ElementRef;
-
-  allNotifications: Map<string, NotificationItem[]> = new Map();
-  unreadNotifications: any;
-  existingNotifications: ExistingNotification[] = [];
-  currentlySelectedNotification: ExistingNotification;
-  currentlySelectedNotificationId: string;
-
-  pipelinesWithNotificationsPresent = false;
-  notificationsLoading = false;
-
-  currentOffset = 0;
-  liveOffset = 0;
-  previousScrollHeight: number;
-
-  subscription: Subscription;
-  notificationTopic: string;
-
-  newNotificationInfo: boolean[] = [];
-
-  newEventArriving = false;
-  lastFetchTime = new Date().getTime();
-
-  constructor(private authService: AuthService,
-              private pipelineService: PipelineService,
-              public elementIconText: ElementIconText,
-              private notificationService: NotificationsService,
-              private notificationCountService: NotificationCountService,
-              private breadcrumbService: SpBreadcrumbService) {
-    this.unreadNotifications = [];
-  }
-
-  ngOnInit() {
-    this.breadcrumbService.updateBreadcrumb([{label: 'Notifications'}]);
-    this.getPipelinesWithNotifications();
-    this.subscription = timer(0, 5000).pipe(
-      filter(() => (this.currentlySelectedNotification !== undefined && this.allNotifications.size > 0)),
-      switchMap(() => this.notificationService.getNotificationsFromTime(
-        this.lastFetchTime)))
-      .subscribe((notifications) => {
-        let scrollToBottom = false;
-        if (notifications.length > 0) {
-          if ((this.notificationContainer.nativeElement.scrollHeight - this.notificationContainer.nativeElement.scrollTop) <=
-            (this.notificationContainer.nativeElement.clientHeight + 10) &&
-            (this.notificationContainer.nativeElement.scrollHeight - this.notificationContainer.nativeElement.scrollTop) >=
-            (this.notificationContainer.nativeElement.clientHeight - 10)) {
-            scrollToBottom = true;
-          }
-          this.newEventArriving = true;
-          notifications.forEach(notification => {
-            const notificationId = NotificationUtils.makeNotificationId(notification.correspondingPipelineId, notification.title);
-            const existingNots = this.allNotifications.get(notificationId);
-            existingNots.push(notification);
-            this.allNotifications.set(notificationId, existingNots);
-            if (this.currentlySelectedNotificationId === notificationId) {
-              this.liveOffset++;
-              notification.read = true;
-              setTimeout(() => {
-                this.notificationService.updateNotification(notification).subscribe();
-              }, 500);
-            } else {
-              this.newNotificationInfo[notificationId] = true;
-            }
-          });
+    @ViewChild('notificationPane') private notificationContainer: ElementRef;
+
+    allNotifications: Map<string, NotificationItem[]> = new Map();
+    unreadNotifications: any;
+    existingNotifications: ExistingNotification[] = [];
+    currentlySelectedNotification: ExistingNotification;
+    currentlySelectedNotificationId: string;
+
+    pipelinesWithNotificationsPresent = false;
+    notificationsLoading = false;
+
+    currentOffset = 0;
+    liveOffset = 0;
+    previousScrollHeight: number;
+
+    subscription: Subscription;
+    notificationTopic: string;
+
+    newNotificationInfo: boolean[] = [];
+
+    newEventArriving = false;
+    lastFetchTime = new Date().getTime();
+
+    constructor(
+        private authService: AuthService,
+        private pipelineService: PipelineService,
+        public elementIconText: ElementIconText,
+        private notificationService: NotificationsService,
+        private notificationCountService: NotificationCountService,
+        private breadcrumbService: SpBreadcrumbService,
+    ) {
+        this.unreadNotifications = [];
+    }
 
-          if (scrollToBottom) {
-            setTimeout(() => {
-              this.scrollToBottom();
+    ngOnInit() {
+        this.breadcrumbService.updateBreadcrumb([{ label: 'Notifications' }]);
+        this.getPipelinesWithNotifications();
+        this.subscription = timer(0, 5000)
+            .pipe(
+                filter(
+                    () =>
+                        this.currentlySelectedNotification !== undefined &&
+                        this.allNotifications.size > 0,
+                ),
+                switchMap(() =>
+                    this.notificationService.getNotificationsFromTime(
+                        this.lastFetchTime,
+                    ),
+                ),
+            )
+            .subscribe(notifications => {
+                let scrollToBottom = false;
+                if (notifications.length > 0) {
+                    if (
+                        this.notificationContainer.nativeElement.scrollHeight -
+                            this.notificationContainer.nativeElement
+                                .scrollTop <=
+                            this.notificationContainer.nativeElement
+                                .clientHeight +
+                                10 &&
+                        this.notificationContainer.nativeElement.scrollHeight -
+                            this.notificationContainer.nativeElement
+                                .scrollTop >=
+                            this.notificationContainer.nativeElement
+                                .clientHeight -
+                                10
+                    ) {
+                        scrollToBottom = true;
+                    }
+                    this.newEventArriving = true;
+                    notifications.forEach(notification => {
+                        const notificationId =
+                            NotificationUtils.makeNotificationId(
+                                notification.correspondingPipelineId,
+                                notification.title,
+                            );
+                        const existingNots =
+                            this.allNotifications.get(notificationId);
+                        existingNots.push(notification);
+                        this.allNotifications.set(notificationId, existingNots);
+                        if (
+                            this.currentlySelectedNotificationId ===
+                            notificationId
+                        ) {
+                            this.liveOffset++;
+                            notification.read = true;
+                            setTimeout(() => {
+                                this.notificationService
+                                    .updateNotification(notification)
+                                    .subscribe();
+                            }, 500);
+                        } else {
+                            this.newNotificationInfo[notificationId] = true;
+                        }
+                    });
+
+                    if (scrollToBottom) {
+                        setTimeout(() => {
+                            this.scrollToBottom();
+                        });
+                    }
+                }
+                this.lastFetchTime = new Date().getTime();
+                this.newEventArriving = false;
             });
-          }
-        }
-        this.lastFetchTime = new Date().getTime();
-        this.newEventArriving = false;
-      });
-  }
-
-  scrollToBottom() {
-   setTimeout(() => {
-     this.notificationContainer.nativeElement.scrollTop = this.notificationContainer.nativeElement.scrollHeight + 100;
-   }, 200);
-  }
-
-  getPipelinesWithNotifications() {
-    this.notificationsLoading = true;
-    this.pipelineService.getOwnPipelines().subscribe(pipelines => {
-      this.existingNotifications = this.getAllExistingNotifications(pipelines);
-      this.notificationsLoading = false;
-      if (this.existingNotifications.length > 0) {
-        this.pipelinesWithNotificationsPresent = true;
-        this.selectNotification(this.existingNotifications[0]);
-        this.existingNotifications.forEach(notification => {
-          this.getNotifications(notification, this.currentOffset, 10, true);
+    }
+
+    scrollToBottom() {
+        setTimeout(() => {
+            this.notificationContainer.nativeElement.scrollTop =
+                this.notificationContainer.nativeElement.scrollHeight + 100;
+        }, 200);
+    }
+
+    getPipelinesWithNotifications() {
+        this.notificationsLoading = true;
+        this.pipelineService.getOwnPipelines().subscribe(pipelines => {
+            this.existingNotifications =
+                this.getAllExistingNotifications(pipelines);
+            this.notificationsLoading = false;
+            if (this.existingNotifications.length > 0) {
+                this.pipelinesWithNotificationsPresent = true;
+                this.selectNotification(this.existingNotifications[0]);
+                this.existingNotifications.forEach(notification => {
+                    this.getNotifications(
+                        notification,
+                        this.currentOffset,
+                        10,
+                        true,
+                    );
+                });
+            }
         });
-      }
-    });
-  }
-
-  getAllExistingNotifications(pipelines: Pipeline[]): ExistingNotification[] {
-    const existingNotifications = [];
-    pipelines.forEach(pipeline => {
-      const notificationActions = pipeline.actions.filter(sink => sink.appId === NotificationsComponent.NOTIFICATIONS_APP_ID);
-      notificationActions.forEach(notificationAction => {
-        const notificationName = notificationAction
-          .staticProperties
-          .filter(sp => sp.internalName === NotificationsComponent.NOTIFICATION_TITLE_KEY)
-          .map(sp => (sp as FreeTextStaticProperty).value)[0];
-        const pipelineName = pipeline.name;
-        existingNotifications.push({
-          notificationTitle: notificationName,
-          pipelineName,
-          pipelineId: pipeline._id,
-          notificationId: NotificationUtils.makeNotificationId(pipeline._id, notificationName)
+    }
+
+    getAllExistingNotifications(pipelines: Pipeline[]): ExistingNotification[] {
+        const existingNotifications = [];
+        pipelines.forEach(pipeline => {
+            const notificationActions = pipeline.actions.filter(
+                sink =>
+                    sink.appId === NotificationsComponent.NOTIFICATIONS_APP_ID,
+            );
+            notificationActions.forEach(notificationAction => {
+                const notificationName = notificationAction.staticProperties
+                    .filter(
+                        sp =>
+                            sp.internalName ===
+                            NotificationsComponent.NOTIFICATION_TITLE_KEY,
+                    )
+                    .map(sp => (sp as FreeTextStaticProperty).value)[0];
+                const pipelineName = pipeline.name;
+                existingNotifications.push({
+                    notificationTitle: notificationName,
+                    pipelineName,
+                    pipelineId: pipeline._id,
+                    notificationId: NotificationUtils.makeNotificationId(
+                        pipeline._id,
+                        notificationName,
+                    ),
+                });
+            });
         });
-      });
-    });
-    return existingNotifications;
-  }
-
-  getNotifications(notification: ExistingNotification,
-                   offset: number,
-                   count: number,
-                   scrollToBottom: boolean) {
-    this.notificationService.getNotifications(notification, offset, count).subscribe(notifications => {
-      notifications.sort((a, b) => {
-        return (a.createdAtTimestamp - b.createdAtTimestamp);
-      });
-      const notificationId = NotificationUtils.makeNotificationId(notification.pipelineId, notification.notificationTitle);
-      if (!this.allNotifications.has(notificationId)) {
-        this.allNotifications.set(notificationId, []);
-      }
-      this.allNotifications.get(notificationId).unshift(...notifications);
-      if (scrollToBottom) {
-        setTimeout(() => {
-          this.scrollToBottom();
-        }, 500);
-      } else {
-        setTimeout(() => {
-          this.notificationContainer.nativeElement.scrollTop =
-            this.notificationContainer.nativeElement.scrollHeight - this.previousScrollHeight;
+        return existingNotifications;
+    }
+
+    getNotifications(
+        notification: ExistingNotification,
+        offset: number,
+        count: number,
+        scrollToBottom: boolean,
+    ) {
+        this.notificationService
+            .getNotifications(notification, offset, count)
+            .subscribe(notifications => {
+                notifications.sort((a, b) => {
+                    return a.createdAtTimestamp - b.createdAtTimestamp;
+                });
+                const notificationId = NotificationUtils.makeNotificationId(
+                    notification.pipelineId,
+                    notification.notificationTitle,
+                );
+                if (!this.allNotifications.has(notificationId)) {
+                    this.allNotifications.set(notificationId, []);
+                }
+                this.allNotifications
+                    .get(notificationId)
+                    .unshift(...notifications);
+                if (scrollToBottom) {
+                    setTimeout(() => {
+                        this.scrollToBottom();
+                    }, 500);
+                } else {
+                    setTimeout(() => {
+                        this.notificationContainer.nativeElement.scrollTop =
+                            this.notificationContainer.nativeElement
+                                .scrollHeight - this.previousScrollHeight;
+                    });
+                }
+                this.updateUnreadNotifications(notifications);
+            });
+    }
+
+    updateUnreadNotifications(notifications: NotificationItem[]) {
+        notifications.forEach(n => {
+            if (!n.read) {
+                n.read = true;
+                this.notificationCountService.decreaseNotificationCount();
+                this.notificationService.updateNotification(n).subscribe();
+            }
         });
-      }
-      this.updateUnreadNotifications(notifications);
-    });
-  }
-
-  updateUnreadNotifications(notifications: NotificationItem[]) {
-    notifications.forEach(n => {
-      if (!n.read) {
-        n.read = true;
-        this.notificationCountService.decreaseNotificationCount();
-        this.notificationService.updateNotification(n).subscribe();
-      }
-    });
-  }
-
-  selectNotification(notification: ExistingNotification) {
-    this.currentOffset = 0;
-    this.liveOffset = 0;
-    this.currentlySelectedNotification = notification;
-    this.currentlySelectedNotificationId = NotificationUtils.makeNotificationIdFromNotification(notification);
-    if (this.allNotifications.has(this.currentlySelectedNotificationId)) {
-      this.updateUnreadNotifications(this.allNotifications.get(this.currentlySelectedNotificationId));
     }
-    this.newNotificationInfo[this.currentlySelectedNotificationId] = false;
-    this.notificationCountService.lockIncreaseUpdateForId(this.currentlySelectedNotificationId);
-    if (this.notificationContainer) {
-      this.scrollToBottom();
+
+    selectNotification(notification: ExistingNotification) {
+        this.currentOffset = 0;
+        this.liveOffset = 0;
+        this.currentlySelectedNotification = notification;
+        this.currentlySelectedNotificationId =
+            NotificationUtils.makeNotificationIdFromNotification(notification);
+        if (this.allNotifications.has(this.currentlySelectedNotificationId)) {
+            this.updateUnreadNotifications(
+                this.allNotifications.get(this.currentlySelectedNotificationId),
+            );
+        }
+        this.newNotificationInfo[this.currentlySelectedNotificationId] = false;
+        this.notificationCountService.lockIncreaseUpdateForId(
+            this.currentlySelectedNotificationId,
+        );
+        if (this.notificationContainer) {
+            this.scrollToBottom();
+        }
     }
-  }
-
-  onScroll() {
-    if (this.notificationContainer.nativeElement.scrollTop === 0) {
-      this.currentOffset += 10;
-      this.previousScrollHeight = this.notificationContainer.nativeElement.scrollHeight;
-      const currentNotifications = this.allNotifications.get(this.currentlySelectedNotificationId);
-      this.getNotifications(this.currentlySelectedNotification, currentNotifications.length + this.liveOffset, 10, false);
+
+    onScroll() {
+        if (this.notificationContainer.nativeElement.scrollTop === 0) {
+            this.currentOffset += 10;
+            this.previousScrollHeight =
+                this.notificationContainer.nativeElement.scrollHeight;
+            const currentNotifications = this.allNotifications.get(
+                this.currentlySelectedNotificationId,
+            );
+            this.getNotifications(
+                this.currentlySelectedNotification,
+                currentNotifications.length + this.liveOffset,
+                10,
+                false,
+            );
+        }
     }
-  }
 
-  ngOnDestroy(): void {
-    if (this.subscription) {
-      this.subscription.unsubscribe();
+    ngOnDestroy(): void {
+        if (this.subscription) {
+            this.subscription.unsubscribe();
+        }
+        this.notificationCountService.unlockIncreaseUpdate();
     }
-    this.notificationCountService.unlockIncreaseUpdate();
-  }
 }
diff --git a/ui/src/app/notifications/notifications.module.ts b/ui/src/app/notifications/notifications.module.ts
index 1882aac787..83cc795474 100644
--- a/ui/src/app/notifications/notifications.module.ts
+++ b/ui/src/app/notifications/notifications.module.ts
@@ -38,23 +38,12 @@ import { PlatformServicesModule } from '@streampipes/platform-services';
         FlexLayoutModule,
         CustomMaterialModule,
         FormsModule,
-        PlatformServicesModule
+        PlatformServicesModule,
     ],
-    declarations: [
-        NotificationsComponent,
-        NotificationItemComponent
-    ],
-    providers: [
-        NotificationsService
-    ],
-    exports: [
-        NotificationsComponent
-    ]
+    declarations: [NotificationsComponent, NotificationItemComponent],
+    providers: [NotificationsService],
+    exports: [NotificationsComponent],
 })
 export class NotificationModule {
-
-  constructor() {
-  }
-
+    constructor() {}
 }
-
diff --git a/ui/src/app/notifications/service/notifications.service.ts b/ui/src/app/notifications/service/notifications.service.ts
index f359b472de..8433f4584b 100644
--- a/ui/src/app/notifications/service/notifications.service.ts
+++ b/ui/src/app/notifications/service/notifications.service.ts
@@ -19,7 +19,7 @@ import { HttpClient } from '@angular/common/http';
 import { Observable } from 'rxjs';
 import {
     ExistingNotification,
-    NotificationItem
+    NotificationItem,
 } from '../model/notifications.model';
 import { Injectable } from '@angular/core';
 import { NotificationUtils } from '../utils/notifications.utils';
@@ -28,41 +28,61 @@ import { PlatformServicesCommons } from '@streampipes/platform-services';
 
 @Injectable()
 export class NotificationsService {
+    constructor(
+        private http: HttpClient,
+        private platformServicesCommons: PlatformServicesCommons,
+    ) {}
 
-    constructor(private http: HttpClient,
-                private platformServicesCommons: PlatformServicesCommons) {
-    }
-
-    getNotificationsFromTime(startTime: number): Observable<NotificationItem[]> {
+    getNotificationsFromTime(
+        startTime: number,
+    ): Observable<NotificationItem[]> {
         return this.http
-          .get(this.notificationUrl
-            + '/time'
-            + '?'
-            + 'startTime=' + startTime)
-          .pipe(map(data => {
-              return data as NotificationItem[];
-          }));
+            .get(
+                this.notificationUrl + '/time' + '?' + 'startTime=' + startTime,
+            )
+            .pipe(
+                map(data => {
+                    return data as NotificationItem[];
+                }),
+            );
     }
 
-    getNotifications(existingNotification: ExistingNotification, offset: number, limit: number): Observable<NotificationItem[]> {
+    getNotifications(
+        existingNotification: ExistingNotification,
+        offset: number,
+        limit: number,
+    ): Observable<NotificationItem[]> {
         return this.http
-            .get(this.notificationUrl
-                + '/offset'
-                + '?'
-                + 'notificationType=' + NotificationUtils.makeNotificationIdFromNotification(existingNotification)
-                + '&'
-                + 'offset=' + offset
-                + '&'
-                + 'count=' + limit)
-            .pipe(map(data => {
-                return data as NotificationItem[];
-            }));
+            .get(
+                this.notificationUrl +
+                    '/offset' +
+                    '?' +
+                    'notificationType=' +
+                    NotificationUtils.makeNotificationIdFromNotification(
+                        existingNotification,
+                    ) +
+                    '&' +
+                    'offset=' +
+                    offset +
+                    '&' +
+                    'count=' +
+                    limit,
+            )
+            .pipe(
+                map(data => {
+                    return data as NotificationItem[];
+                }),
+            );
     }
 
     updateNotification(notificationItem: NotificationItem): Observable<any> {
-        return this.http.put(this.notificationUrl + '/' + notificationItem._id, notificationItem, {
-            headers: { ignoreLoadingBar: '' }
-        });
+        return this.http.put(
+            this.notificationUrl + '/' + notificationItem._id,
+            notificationItem,
+            {
+                headers: { ignoreLoadingBar: '' },
+            },
+        );
     }
 
     private get notificationUrl() {
diff --git a/ui/src/app/notifications/utils/notifications.utils.ts b/ui/src/app/notifications/utils/notifications.utils.ts
index 8fb001357c..da7cda8399 100644
--- a/ui/src/app/notifications/utils/notifications.utils.ts
+++ b/ui/src/app/notifications/utils/notifications.utils.ts
@@ -18,9 +18,13 @@
 import { ExistingNotification } from '../model/notifications.model';
 
 export class NotificationUtils {
-
-    static makeNotificationIdFromNotification(existingNotification: ExistingNotification): string {
-       return this.makeNotificationId(existingNotification.pipelineId, existingNotification.notificationTitle);
+    static makeNotificationIdFromNotification(
+        existingNotification: ExistingNotification,
+    ): string {
+        return this.makeNotificationId(
+            existingNotification.pipelineId,
+            existingNotification.notificationTitle,
+        );
     }
 
     static makeNotificationId(pipelineId: string, notificationTitle: string) {


Test patch:
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java
similarity index 54%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java
rename to streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java
index 52a3ce6126..f525319ad9 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java
+++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java
@@ -16,11 +16,30 @@
  *
  */
 
-package org.apache.streampipes.performance.dataprovider;
+package org.apache.streampipes.extensions.management.connect.adapter.format.csv;
 
+
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 
-public interface DataProvider<T> {
+import static org.junit.Assert.assertEquals;
+
+public class CsvParserTest {
+
+  @Test
+  public void getSchemaAndSampleTimestamp() {
+    var csvValue = List.of(
+        "timestamp,value".getBytes(StandardCharsets.UTF_8),
+        "1667904471000,1".getBytes(StandardCharsets.UTF_8)
+    );
+
+    var csvParser = new CsvParser(",", true);
+    var result = csvParser.getSchemaAndSample(csvValue);
+
+    assertEquals(result.getEventPreview().size(), 1);
+    assertEquals(1667904471000.0, result.getEventPreview().get(0).get("timestamp").getValue());
+  }
 
-  List<T> getPreparedItems();
-}
+}
\ No newline at end of file
diff --git a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtilsTest.java b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtilsTest.java
new file mode 100644
index 0000000000..c5d273e071
--- /dev/null
+++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/util/DatatypeUtilsTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.extensions.management.connect.adapter.util;
+
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.junit.Test;
+
+import java.util.Locale;
+
+import static org.junit.Assert.assertEquals;
+
+public class DatatypeUtilsTest {
+
+  @Test
+  /**
+   * This test ensures that timestamps represented as strings are correctly parsed.
+   * Often they are first parsed into floating point number before transformed back to long.
+   * The data type for those values should be Double and not Float, because the transformation to Float might change
+   * the value
+   */
+  public void convertTimestampValue() {
+    var inputValue = "1667904471000";
+
+    var floatValue = DatatypeUtils.convertValue(inputValue, XSD.DOUBLE.toString());
+    var longValue = DatatypeUtils.convertValue(floatValue, XSD.LONG.toString());
+
+    assertEquals(Long.parseLong(inputValue), longValue);
+  }
+
+  String booleanInputValue = "true";
+  @Test
+  public void getTypeClassNoPrefereFloatingPointBoolean() {
+    var result = DatatypeUtils.getTypeClass(booleanInputValue, false);
+    assertEquals(Boolean.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointBoolean() {
+    var result = DatatypeUtils.getTypeClass(booleanInputValue, true);
+    assertEquals(Boolean.class, result);
+  }
+
+  String integerInputValue = "1";
+  @Test
+  public void getTypeClassNoPrefereFloatingPointInteger() {
+    var result = DatatypeUtils.getTypeClass(integerInputValue, false);
+    assertEquals(Integer.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointInteger() {
+    var result = DatatypeUtils.getTypeClass(integerInputValue, true);
+    assertEquals(Float.class, result);
+  }
+
+  String floatInputValue = "1.0";
+  @Test
+  public void getTypeClassNoPrefereFloatingPointFloat() {
+    var result = DatatypeUtils.getTypeClass(floatInputValue, false);
+    assertEquals(Float.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointFloat() {
+    var result = DatatypeUtils.getTypeClass(floatInputValue, true);
+    assertEquals(Float.class, result);
+  }
+
+
+  String doubleInputValue = String.format(Locale.US, "%.2f", Double.MAX_VALUE);
+  @Test
+  public void getTypeClassNoPrefereFloatingPointDouble() {
+    var result = DatatypeUtils.getTypeClass(doubleInputValue, false);
+    assertEquals(Double.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointDouble() {
+    var result = DatatypeUtils.getTypeClass(doubleInputValue, true);
+    assertEquals(Double.class, result);
+  }
+
+
+  String longInputValue = "1667904471000";
+  @Test
+  public void getTypeClassNoPrefereFloatingPointLong() {
+    var result = DatatypeUtils.getTypeClass(longInputValue, false);
+    assertEquals(Long.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointLong() {
+    var result = DatatypeUtils.getTypeClass(longInputValue, true);
+    assertEquals(Double.class, result);
+  }
+
+  String stringInputValue = "one";
+  @Test
+  public void getTypeClassNoPrefereFloatingPointString() {
+    var result = DatatypeUtils.getTypeClass(stringInputValue, false);
+
+    assertEquals(String.class, result);
+  }
+
+  @Test
+  public void getTypeClassWithPrefereFloatingPointString() {
+    var result = DatatypeUtils.getTypeClass(stringInputValue, true);
+
+    assertEquals(String.class, result);
+  }
+
+
+}
\ No newline at end of file
diff --git a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/util/EventSchemaUtilsTest.java b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/util/EventSchemaUtilsTest.java
new file mode 100644
index 0000000000..58797f97fe
--- /dev/null
+++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/util/EventSchemaUtilsTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.extensions.management.util;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.test.generator.EventPropertyNestedTestBuilder;
+import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder;
+import org.apache.streampipes.test.generator.EventSchemaTestBuilder;
+import org.apache.streampipes.vocabulary.SO;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class EventSchemaUtilsTest {
+
+  EventProperty timestampProperty = EventPropertyPrimitiveTestBuilder.create()
+      .withSemanticType(SO.DATE_TIME)
+      .withRuntimeName("timestamp")
+      .build();
+
+  @Test
+  public void noTimestampPropery() {
+
+    var eventSchema = EventSchemaTestBuilder.create()
+        .withEventProperty(
+            EventPropertyPrimitiveTestBuilder.create().build())
+        .build();
+
+    var result = EventSchemaUtils.getTimestampProperty(eventSchema);
+
+    assertFalse(result.isPresent());
+  }
+
+  @Test
+  public void getTimestampProperty() {
+    var eventSchema = EventSchemaTestBuilder.create()
+        .withEventProperty(timestampProperty)
+        .build();
+
+    var result = EventSchemaUtils.getTimestampProperty(eventSchema);
+
+    assertTrue(result.isPresent());
+    assertEquals(result.get(), timestampProperty);
+  }
+
+  @Test
+  public void getNestedTimestampProperty() {
+
+    var eventSchema = EventSchemaTestBuilder.create()
+        .withEventProperty(
+            EventPropertyNestedTestBuilder.create()
+                .withEventProperty(timestampProperty)
+                .build())
+        .build();
+
+    var result = EventSchemaUtils.getTimestampProperty(eventSchema);
+
+    assertTrue(result.isPresent());
+    assertEquals(result.get(), timestampProperty);
+  }
+}
\ No newline at end of file
diff --git a/streampipes-model/src/test/java/org/apache/streampipes/model/connect/adapter/AdapterDescriptionTest.java b/streampipes-model/src/test/java/org/apache/streampipes/model/connect/adapter/AdapterDescriptionTest.java
index dd8ce6e089..bd529a9acf 100644
--- a/streampipes-model/src/test/java/org/apache/streampipes/model/connect/adapter/AdapterDescriptionTest.java
+++ b/streampipes-model/src/test/java/org/apache/streampipes/model/connect/adapter/AdapterDescriptionTest.java
@@ -37,7 +37,7 @@ public class AdapterDescriptionTest {
 
   @Before
   public void init() {
-    adapterDescription = new AdapterDescription() {
+    adapterDescription = new SpecificAdapterStreamDescription() {
     };
 
     List rules = new ArrayList<>();
diff --git a/streampipes-performance-tests/pom.xml b/streampipes-performance-tests/pom.xml
deleted file mode 100644
index 49c1271c8a..0000000000
--- a/streampipes-performance-tests/pom.xml
+++ /dev/null
@@ -1,64 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>streampipes-parent</artifactId>
-        <groupId>org.apache.streampipes</groupId>
-        <version>0.91.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>streampipes-performance-tests</artifactId>
-
-    <dependencies>
-        <!-- StreamPipes dependencies -->
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-messaging-kafka</artifactId>
-            <version>0.91.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-model</artifactId>
-            <version>0.91.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-pipeline-management</artifactId>
-            <version>0.91.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-json</artifactId>
-            <version>0.91.0-SNAPSHOT</version>
-        </dependency>
-
-        <!-- External dependencies -->
-    </dependencies>
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-</project>
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java
deleted file mode 100644
index 93626cee3b..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java
+++ /dev/null
@@ -1,26 +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.performance;
-
-public class PerformanceTest {
-
-  public static void main(String[] args) {
-
-  }
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PipelinePerformanceTest.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PipelinePerformanceTest.java
deleted file mode 100644
index 703b6e2990..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PipelinePerformanceTest.java
+++ /dev/null
@@ -1,57 +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.performance;
-
-import org.apache.streampipes.performance.model.PerformanceTestSettings;
-import org.apache.streampipes.performance.util.ParameterTool;
-
-public class PipelinePerformanceTest {
-
-  /**
-   * configuration parameters:
-   * - number of timestamp enrichment epas
-   * - parallelism of enrichment epas
-   * - parallelism of elastic epas
-   * - total number of events to be simulated
-   * - wait time between events in ms
-   * - number of producer threads
-   * - backend URL (optional)
-   */
-
-  public static void main(String[] args) {
-    if (args.length < 6) {
-      System.out.println("Usage: java -jar streampipes-performance-tests.jar [NUMBER OF TIMESTAMP ENRICH EPAS] "
-          + "[PARALELLISM OF ENRICHMENT EPAS], [PARALELLISM OF ELASTIC EPAs], [TOTAL NUMBER OF EVENTS], [WAIT TIME "
-          + "BETWEEN EVENTS], [NUMBER OF PRODUCER THREADS]");
-    } else {
-      PerformanceTestSettings settings = ParameterTool.fromArgs(args);
-
-
-    }
-
-  }
-
-  // SOURCE
-
-  // EPA
-
-  // SINK
-  //Kafka Publisher
-
-
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/JsonDataProvider.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/JsonDataProvider.java
deleted file mode 100644
index 3dbd811111..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/JsonDataProvider.java
+++ /dev/null
@@ -1,110 +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.performance.dataprovider;
-
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventPropertyList;
-import org.apache.streampipes.model.schema.EventPropertyPrimitive;
-import org.apache.streampipes.model.schema.EventSchema;
-
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-import org.apache.commons.lang3.RandomStringUtils;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-
-public class JsonDataProvider implements DataProvider<String> {
-
-  private EventSchema schema;
-  private Long numberOfEvents;
-  private List<String> outputEvents;
-  private Random random;
-
-  public JsonDataProvider(EventSchema schema, Long numberOfEvents) {
-    this.schema = schema;
-    this.numberOfEvents = numberOfEvents;
-    this.outputEvents = new ArrayList<>();
-    this.random = new Random();
-  }
-
-  @Override
-  public List<String> getPreparedItems() {
-    for (Long i = 0L; i < numberOfEvents; i++) {
-      outputEvents.add(makeEvent());
-    }
-
-    return outputEvents;
-  }
-
-  private String makeEvent() {
-    JsonObject jsonObject = new JsonObject();
-    for (EventProperty property : schema.getEventProperties()) {
-      jsonObject.add(property.getRuntimeName(), makeValue(property));
-    }
-
-    return jsonObject.toString();
-  }
-
-  private JsonElement makeValue(EventProperty property) {
-    if (property instanceof EventPropertyPrimitive) {
-//      switch (((EventPropertyPrimitive) property).getRuntimeType()) {
-//        case XSD._integer.toString():
-//          return new JsonPrimitive(getRandomInteger());
-//        case XSD._string:
-//          return new JsonPrimitive(getRandomString());
-//        case XSD.LONG:
-//          return new JsonPrimitive(getRandomLong());
-//        case XSD.DOUBLE:
-//          return new JsonPrimitive(getRandomDouble());
-//        case XSD.FLOAT:
-//          return new JsonPrimitive(getRandomFloat());
-//      }
-    } else if (property instanceof EventPropertyList) {
-      // TODO
-      return null;
-    }
-    // TODO: nested property
-    return null;
-
-  }
-
-
-  private Float getRandomFloat() {
-    return random.nextFloat();
-  }
-
-  private Double getRandomDouble() {
-    return random.nextDouble();
-  }
-
-  private Long getRandomLong() {
-    return random.nextLong();
-  }
-
-  private String getRandomString() {
-    return RandomStringUtils.randomAlphabetic(20);
-  }
-
-  private Integer getRandomInteger() {
-    return random.nextInt(1000);
-  }
-
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/StreamProvider.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/StreamProvider.java
deleted file mode 100644
index 2a9fc50880..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/StreamProvider.java
+++ /dev/null
@@ -1,26 +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.performance.dataprovider;
-
-import org.apache.streampipes.model.SpDataStream;
-
-public interface StreamProvider {
-
-  SpDataStream getStreamDescription();
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/model/PerformanceTestSettings.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/model/PerformanceTestSettings.java
deleted file mode 100644
index d06836d21b..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/model/PerformanceTestSettings.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.performance.model;
-
-/**
- * configuration parameters:
- * - number of timestamp enrichment epas
- * - parallelism of enrichment epas
- * - parallelism of elastic epas
- * - total number of events to be simulated
- * - wait time between events in ms
- * - number of producer threads
- * - backend URL (optional)
- */
-public class PerformanceTestSettings {
-
-  private Integer numberOfTimestampEnrichmentEpas;
-  private Integer parallelismOfEnrichmentEpas;
-  private Integer parallelismOfElasticEpas;
-  private Long totalNumberofEvents;
-  private Long waitTimeBetweenEventsInMs;
-  private Integer numProducerThreads;
-  private String elasticUrl;
-
-  public PerformanceTestSettings(Integer numberOfTimestampEnrichmentEpas, Integer parallelismOfEnrichmentEpas,
-                                 Integer parallelismOfElasticEpas, Long totalNumberofEvents,
-                                 Long waitTimeBetweenEventsInMs, Integer numProducerThreads, String elasticUrl) {
-    this.numberOfTimestampEnrichmentEpas = numberOfTimestampEnrichmentEpas;
-    this.parallelismOfEnrichmentEpas = parallelismOfEnrichmentEpas;
-    this.parallelismOfElasticEpas = parallelismOfElasticEpas;
-    this.totalNumberofEvents = totalNumberofEvents;
-    this.waitTimeBetweenEventsInMs = waitTimeBetweenEventsInMs;
-    this.numProducerThreads = numProducerThreads;
-    this.elasticUrl = elasticUrl;
-  }
-
-  public Integer getNumberOfTimestampEnrichmentEpas() {
-    return numberOfTimestampEnrichmentEpas;
-  }
-
-  public Integer getParallelismOfEnrichmentEpas() {
-    return parallelismOfEnrichmentEpas;
-  }
-
-  public Integer getParallelismOfElasticEpas() {
-    return parallelismOfElasticEpas;
-  }
-
-  public Long getTotalNumberofEvents() {
-    return totalNumberofEvents;
-  }
-
-  public Long getWaitTimeBetweenEventsInMs() {
-    return waitTimeBetweenEventsInMs;
-  }
-
-  public Integer getNumProducerThreads() {
-    return numProducerThreads;
-  }
-
-  public String getElasticUrl() {
-    return elasticUrl;
-  }
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionFetcher.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionFetcher.java
deleted file mode 100644
index 15e9e925dd..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionFetcher.java
+++ /dev/null
@@ -1,21 +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.performance.pipeline;
-
-public class PipelineDescriptionFetcher {
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionParser.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionParser.java
deleted file mode 100644
index 5dbefa11d2..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/pipeline/PipelineDescriptionParser.java
+++ /dev/null
@@ -1,21 +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.performance.pipeline;
-
-public class PipelineDescriptionParser {
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/producer/DataSimulator.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/producer/DataSimulator.java
deleted file mode 100644
index ac0eb3e18d..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/producer/DataSimulator.java
+++ /dev/null
@@ -1,74 +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.performance.producer;
-
-import org.apache.streampipes.messaging.kafka.SpKafkaProducer;
-import org.apache.streampipes.performance.simulation.DataReplayStatusNotifier;
-
-import java.util.Collections;
-import java.util.Random;
-import java.util.UUID;
-
-public class DataSimulator implements Runnable {
-
-  private static final String topic = "";
-
-  private String threadId;
-  private SpKafkaProducer kafkaProducer;
-
-  private Long totalNumberOfEvents;
-  private Long waitTimeBetweenEvents;
-  private Random random;
-
-  private DataReplayStatusNotifier statusNotifier;
-
-  public DataSimulator(String kafkaUrl, Long totalNumberOfEvents, Long waitTimeBetweenEvents, String threadId,
-                       DataReplayStatusNotifier statusNotifier) {
-    this.kafkaProducer = new SpKafkaProducer(kafkaUrl, topic, Collections.emptyList());
-    this.threadId = threadId;
-
-    this.totalNumberOfEvents = totalNumberOfEvents;
-    this.waitTimeBetweenEvents = waitTimeBetweenEvents;
-
-    this.statusNotifier = statusNotifier;
-    this.random = new Random();
-  }
-
-  @Override
-  public void run() {
-    Long eventCount = 0L;
-    do {
-      kafkaProducer.publish(nextEvent());
-      eventCount++;
-    } while (eventCount < totalNumberOfEvents);
-
-    this.kafkaProducer.disconnect();
-    statusNotifier.onFinished(threadId);
-  }
-
-  private byte[] nextEvent() {
-    StringBuilder builder = new StringBuilder();
-    builder.append("{");
-    builder.append("\"timestamp\" : ").append(System.currentTimeMillis()).append(",");
-    builder.append("\"sensorId\" : \"").append(UUID.randomUUID().toString()).append("\",");
-    builder.append("\"pressure\" : ").append(random.nextDouble());
-    builder.append("}");
-
-    return builder.toString().getBytes();
-  }
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/DataReplayStatusNotifier.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/DataReplayStatusNotifier.java
deleted file mode 100644
index 7c39635862..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/DataReplayStatusNotifier.java
+++ /dev/null
@@ -1,24 +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.performance.simulation;
-
-public interface DataReplayStatusNotifier {
-
-  void onFinished(String threadId);
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationManager.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationManager.java
deleted file mode 100644
index bc68c9942d..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationManager.java
+++ /dev/null
@@ -1,69 +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.performance.simulation;
-
-import org.apache.streampipes.performance.model.PerformanceTestSettings;
-import org.apache.streampipes.performance.producer.DataSimulator;
-
-import org.apache.commons.lang3.RandomStringUtils;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class SimulationManager {
-
-  private static final String kafkaUrl = "kafka:9092";
-
-  private PerformanceTestSettings settings;
-  private SimulationStatusNotifier notifier;
-
-  private Map<String, Boolean> statusMap;
-
-  public SimulationManager(PerformanceTestSettings settings, SimulationStatusNotifier notifier) {
-    this.settings = settings;
-    this.notifier = notifier;
-    this.statusMap = new HashMap<>();
-  }
-
-  public void initSimulation() {
-
-    List<Thread> threads = new ArrayList<>();
-
-    for (Integer i = 0; i < settings.getNumProducerThreads(); i++) {
-      String threadId = RandomStringUtils.randomAlphanumeric(6);
-      statusMap.put(threadId, false);
-      threads.add(new Thread(new DataSimulator(kafkaUrl, settings.getTotalNumberofEvents(), settings
-          .getWaitTimeBetweenEventsInMs(), threadId, threadId1 -> {
-        statusMap.put(threadId1, true);
-        checkFinished();
-      })));
-    }
-
-    for (Thread thread : threads) {
-      thread.start();
-    }
-  }
-
-  private void checkFinished() {
-    if (statusMap.keySet().stream().allMatch(key -> statusMap.get(key))) {
-      notifier.onFinished();
-    }
-  }
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationStatusNotifier.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationStatusNotifier.java
deleted file mode 100644
index 9da2e57b19..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/simulation/SimulationStatusNotifier.java
+++ /dev/null
@@ -1,24 +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.performance.simulation;
-
-public interface SimulationStatusNotifier {
-
-  void onFinished();
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/tests/serializer/JsonSerializerTest.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/tests/serializer/JsonSerializerTest.java
deleted file mode 100644
index 8f6b67bbee..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/tests/serializer/JsonSerializerTest.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.performance.tests.serializer;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
-import org.apache.streampipes.performance.dataprovider.JsonDataProvider;
-import org.apache.streampipes.performance.dataprovider.SimpleSchemaProvider;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-public class JsonSerializerTest {
-
-  private static final Logger LOG = LoggerFactory.getLogger(JsonSerializerTest.class);
-
-  public static void main(String[] args) {
-    List<String> data = new JsonDataProvider(new SimpleSchemaProvider().getSchema(), 1000000L).getPreparedItems();
-    List<Map<String, Object>> events = new ArrayList<>();
-    JsonDataFormatDefinition formatter = new JsonDataFormatDefinition();
-
-    LOG.info("Total events: " + data.size());
-    LOG.info("Starting deserializer test");
-    Long startTime = System.currentTimeMillis();
-    for (String item : data) {
-      try {
-        events.add(formatter.toMap(item.getBytes()));
-      } catch (SpRuntimeException e) {
-        e.printStackTrace();
-      }
-    }
-    Long endTime = System.currentTimeMillis();
-
-    LOG.info("Total time: " + String.valueOf(endTime - startTime));
-    LOG.info("Per event: " + String.valueOf((endTime - startTime) / data.size()));
-
-    LOG.info("Starting serializer test");
-    startTime = System.currentTimeMillis();
-    for (Map<String, Object> event : events) {
-      try {
-        formatter.fromMap(event);
-      } catch (SpRuntimeException e) {
-        e.printStackTrace();
-      }
-    }
-    endTime = System.currentTimeMillis();
-
-    LOG.info("Total time: " + String.valueOf(endTime - startTime));
-    LOG.info("Per event: " + String.valueOf((endTime - startTime) / data.size()));
-  }
-}
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java b/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java
deleted file mode 100644
index bca098b63f..0000000000
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java
+++ /dev/null
@@ -1,36 +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.performance.util;
-
-import org.apache.streampipes.performance.model.PerformanceTestSettings;
-
-public class ParameterTool {
-
-  public static PerformanceTestSettings fromArgs(String[] args) {
-    return new PerformanceTestSettings(toInt(args[0]), toInt(args[1]), toInt(args[2]), toLong(args[3]), toLong
-        (args[4]), toInt(args[5]), args[6]);
-  }
-
-  private static Long toLong(String arg) {
-    return Long.parseLong(arg);
-  }
-
-  private static Integer toInt(String arg) {
-    return Integer.parseInt(arg);
-  }
-}
diff --git a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyNestedTestBuilder.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyNestedTestBuilder.java
new file mode 100644
index 0000000000..aecb3b24ca
--- /dev/null
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyNestedTestBuilder.java
@@ -0,0 +1,62 @@
+/*
+ * 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.generator;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyNested;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class EventPropertyNestedTestBuilder
+    extends EventPropertyTestBuilder<EventPropertyNested, EventPropertyNestedTestBuilder> {
+
+  private List<EventProperty> eventProperties;
+
+  protected EventPropertyNestedTestBuilder() {
+    super(new EventPropertyNested());
+    eventProperties = new ArrayList<>();
+  }
+
+  public static EventPropertyNestedTestBuilder create() {
+    return new EventPropertyNestedTestBuilder();
+  }
+
+  public EventPropertyNestedTestBuilder withEventProperties(List<EventProperty> eventProperties) {
+    this.eventProperties.addAll(eventProperties);
+    return this;
+  }
+
+  public EventPropertyNestedTestBuilder withEventProperty(EventProperty eventProperty) {
+    this.eventProperties.add(eventProperty);
+    return this;
+  }
+
+  @Override
+  protected EventPropertyNestedTestBuilder me() {
+    return this;
+  }
+
+  @Override
+  public EventPropertyNested build() {
+    eventProperty.setEventProperties(eventProperties);
+    return eventProperty;
+  }
+
+}
diff --git a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaGenerator.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
similarity index 51%
rename from streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaGenerator.java
rename to streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
index 90e96e5849..8fd69330df 100644
--- a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaGenerator.java
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
@@ -17,6 +17,33 @@
  */
 package org.apache.streampipes.test.generator;
 
-public class EventSchemaGenerator {
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 
+import java.net.URI;
+import java.util.List;
+
+public class EventPropertyPrimitiveTestBuilder
+    extends EventPropertyTestBuilder<EventPropertyPrimitive, EventPropertyPrimitiveTestBuilder> {
+
+  protected EventPropertyPrimitiveTestBuilder() {
+    super(new EventPropertyPrimitive());
+  }
+
+  public EventPropertyPrimitiveTestBuilder withSemanticType(String semanticType) {
+    this.eventProperty.setDomainProperties(List.of(URI.create(semanticType)));
+    return this;
+  }
+
+  public static EventPropertyPrimitiveTestBuilder create() {
+    return new EventPropertyPrimitiveTestBuilder();
+  }
+
+  @Override
+  public EventPropertyPrimitive build() {
+    return eventProperty;
+  }
+
+  protected EventPropertyPrimitiveTestBuilder me() {
+    return this;
+  }
 }
diff --git a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyGenerator.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyTestBuilder.java
similarity index 63%
rename from streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyGenerator.java
rename to streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyTestBuilder.java
index c3c4d33a76..6bdb52aab1 100644
--- a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyGenerator.java
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyTestBuilder.java
@@ -15,8 +15,26 @@
  * limitations under the License.
  *
  */
+
 package org.apache.streampipes.test.generator;
 
-public class EventPropertyGenerator {
+import org.apache.streampipes.model.schema.EventProperty;
+
+public abstract  class EventPropertyTestBuilder<T extends EventProperty, K extends EventPropertyTestBuilder> {
+
+  protected T eventProperty;
+
+  protected abstract K me();
+
+  protected EventPropertyTestBuilder(T eventProperty) {
+    this.eventProperty = eventProperty;
+  }
+
+  public K withRuntimeName(String runtimeName) {
+    this.eventProperty.setRuntimeName(runtimeName);
+    return me();
+  }
+
+  public abstract T build();
 
 }
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/SimpleSchemaProvider.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaTestBuilder.java
similarity index 52%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/SimpleSchemaProvider.java
rename to streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaTestBuilder.java
index df060d59c0..234505ff88 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/SimpleSchemaProvider.java
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventSchemaTestBuilder.java
@@ -15,39 +15,40 @@
  * limitations under the License.
  *
  */
-
-package org.apache.streampipes.performance.dataprovider;
+package org.apache.streampipes.test.generator;
 
 import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.schema.EventSchema;
 
-import org.apache.commons.lang3.RandomStringUtils;
-
 import java.util.ArrayList;
 import java.util.List;
 
-public class SimpleSchemaProvider {
+public class EventSchemaTestBuilder {
 
-//  private static final List<String> runtimeTypes = Arrays.asList(XSD.INTEGER, XSD.LONG, XSD.STRING, XSD.FLOAT, XSD
-//          .DOUBLE);
+  private List<EventProperty> eventProperties;
 
-  public EventSchema getSchema() {
-    EventSchema schema = new EventSchema();
-    List<EventProperty> properties = new ArrayList<>();
-    for (int i = 0; i < 5; i++) {
-      properties.add(makeRandomProperty());
-    }
+  private EventSchemaTestBuilder() {
+    eventProperties = new ArrayList<>();
+  }
 
-    schema.setEventProperties(properties);
-    return schema;
+  public static EventSchemaTestBuilder create() {
+    return new EventSchemaTestBuilder();
   }
 
-  private EventProperty makeRandomProperty() {
-    EventPropertyPrimitive property = new EventPropertyPrimitive();
-    property.setRuntimeName(RandomStringUtils.randomAlphabetic(5));
-    //property.setRuntimeType(runtimeTypes.get(new Random().nextInt(5)));
+  public EventSchemaTestBuilder withEventProperties(List<EventProperty> eventProperties) {
+    this.eventProperties.addAll(eventProperties);
+    return this;
+  }
 
-    return property;
+  public EventSchemaTestBuilder withEventProperty(EventProperty eventProperty) {
+    eventProperties.add(eventProperty);
+    return this;
   }
+
+  public EventSchema build() {
+    EventSchema eventSchema = new EventSchema();
+    eventSchema.setEventProperties(eventProperties);
+    return eventSchema;
+  }
+
 }