Repository: apache/streampipes
Pull request #1097 (2023-01-15 21:08:10)

Improve structure of pipeline execution management (#1096)
<!--
  ~ 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.
-->

### 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..dda9b6095b 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,13 +151,13 @@
 	<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>
 	<errorprone-annotations.version>2.2.0</errorprone-annotations.version>
 	<google-maps-services.version>0.10.0</google-maps-services.version>
-	<graalvm.js.version>21.3.0</graalvm.js.version>
+	<graalvm.js.version>22.3.1</graalvm.js.version>
 	<iotdb.version>1.0.0</iotdb.version>
 	<java-websocket.version>1.5.0</java-websocket.version>
 	<jakarta-websocket-client-api.version>2.1.0</jakarta-websocket-client-api.version>
@@ -197,7 +197,6 @@
 	<commons-collections4.version>4.4</commons-collections4.version>
 	<commons-text.version>1.8</commons-text.version>
 	<guice.version>5.1.0</guice.version>
-	<jcl-over-slf4j.version>1.7.30</jcl-over-slf4j.version>
 	<eclipse.milo.version>0.6.3</eclipse.milo.version>
 	<netty.version>4.1.72.Final</netty.version>
 	<nimbus-jose-jwt.version>7.9</nimbus-jose-jwt.version>
@@ -1151,11 +1150,6 @@
                 <artifactId>lombok</artifactId>
                 <version>${lombok.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.slf4j</groupId>
-                <artifactId>jcl-over-slf4j</artifactId>
-                <version>${jcl-over-slf4j.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.netty</groupId>
                 <artifactId>netty-buffer</artifactId>
@@ -1436,7 +1430,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..90a3d84fd8 100644
--- a/streampipes-client-python/mkdocs.yml
+++ b/streampipes-client-python/mkdocs.yml
@@ -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..3dd81477de 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.20.0",
+    "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-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
index c66701e277..dea69c4c84 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
@@ -29,19 +29,18 @@ import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.adapter.AdapterStreamDescription;
 import org.apache.streampipes.model.grounding.EventGrounding;
+import org.apache.streampipes.model.util.ElementIdGenerator;
 import org.apache.streampipes.resource.management.AdapterResourceManager;
 import org.apache.streampipes.resource.management.DataStreamResourceManager;
 import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.api.IAdapterStorage;
 import org.apache.streampipes.storage.couchdb.impl.AdapterInstanceStorageImpl;
 
-import org.apache.commons.lang3.RandomStringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.URISyntaxException;
 import java.util.List;
-import java.util.UUID;
 
 /**
  * This class is responsible for managing all the adapter instances which are executed on worker nodes
@@ -74,10 +73,8 @@ public class AdapterMasterManagement {
       throws AdapterException {
 
     // Create elementId for adapter
-    // TODO centralized provisioning of element id
-    String dataStreamElementId = "urn:streampipes.apache.org:eventstream:" + RandomStringUtils.randomAlphabetic(6);
-    String uuid = UUID.randomUUID().toString();
-    ad.setElementId(ad.getElementId() + ":" + uuid);
+    String dataStreamElementId = ElementIdGenerator.makeElementId(SpDataStream.class);
+    ad.setElementId(ElementIdGenerator.makeElementId(ad));
     ad.setCreatedAt(System.currentTimeMillis());
     ad.setCorrespondingDataStreamElementId(dataStreamElementId);
 
@@ -87,18 +84,13 @@ public class AdapterMasterManagement {
 
     String elementId = this.adapterResourceManager.encryptAndCreate(ad);
 
-    // start when stream adapter
-    if (ad instanceof AdapterStreamDescription) {
-      startStreamAdapter(elementId);
-    }
-
     // Create stream
     SpDataStream storedDescription = new SourcesManagement().createAdapterDataStream(ad, dataStreamElementId);
     storedDescription.setCorrespondingAdapterId(elementId);
     installDataSource(storedDescription, principalSid, true);
     LOG.info("Install source (source URL: {} in backend", ad.getElementId());
 
-    return storedDescription.getElementId();
+    return ad.getElementId();
   }
 
   public void updateAdapter(AdapterDescription ad,
@@ -164,17 +156,6 @@ public class AdapterMasterManagement {
     return allAdapters;
   }
 
-  public List<AdapterDescription> getAllAdapterDescriptions() throws AdapterException {
-
-    List<AdapterDescription> allAdapters = adapterInstanceStorage.getAllAdapters();
-
-    if (allAdapters == null) {
-      throw new AdapterException("Could not get all adapters");
-    }
-
-    return allAdapters;
-  }
-
   public void stopStreamAdapter(String elementId) throws AdapterException {
     AdapterDescription ad = adapterInstanceStorage.getAdapter(elementId);
 
@@ -212,11 +193,11 @@ public class AdapterMasterManagement {
     }
   }
 
-  private void updateDataSource(AdapterDescription ad) throws AdapterException {
+  private void updateDataSource(AdapterDescription ad) {
     // get data source
     SpDataStream dataStream = this.dataStreamResourceManager.find(ad.getCorrespondingDataStreamElementId());
 
-    dataStream = SourcesManagement.updateDataStream(ad, dataStream);
+    SourcesManagement.updateDataStream(ad, dataStream);
 
     // Update data source in database
     this.dataStreamResourceManager.update(dataStream);
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..fef4869907 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,78 @@ 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());
+
+
+    var eventProcessor = new LocalEventProcessor(adapterPipeline, timestampKey);
+    executor = Executors.newScheduledThreadPool(1);
+
+    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) {
+              logger.info("File stream adapter was stopped, the current replay is interuppted", e);
+              return false;
             }
-          } 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 +182,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-extensions/streampipes-sinks-databases-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
index 64ad6cbc67..0cea72e1df 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
@@ -107,10 +107,6 @@
             <groupId>org.bouncycastle</groupId>
             <artifactId>bcpkix-jdk15on</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>jcl-over-slf4j</artifactId>
-        </dependency>
 
         <!-- remove these javax dependencies whenever possible -->
         <dependency>
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java b/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
index b0a9dff087..cfcbf77e3f 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java
@@ -17,10 +17,13 @@
  */
 package org.apache.streampipes.model;
 
+import org.apache.streampipes.model.api.EndpointSelectable;
 import org.apache.streampipes.model.grounding.EventGrounding;
 import org.apache.streampipes.model.schema.EventSchema;
 
-public class SpDataSet extends SpDataStream {
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+public class SpDataSet extends SpDataStream implements EndpointSelectable {
 
   private EventGrounding supportedGrounding;
 
@@ -90,18 +93,28 @@ public class SpDataSet extends SpDataStream {
     this.datasetInvocationId = datasetInvocationId;
   }
 
+  @Override
   public String getCorrespondingPipeline() {
     return correspondingPipeline;
   }
 
+  @Override
   public void setCorrespondingPipeline(String correspondingPipeline) {
     this.correspondingPipeline = correspondingPipeline;
   }
 
+  @Override
+  @JsonIgnore
+  public String getDetachPath() {
+    return "/" + getCorrespondingAdapterId() + "/" + getDatasetInvocationId();
+  }
+
+  @Override
   public String getSelectedEndpointUrl() {
     return selectedEndpointUrl;
   }
 
+  @Override
   public void setSelectedEndpointUrl(String selectedEndpointUrl) {
     this.selectedEndpointUrl = selectedEndpointUrl;
   }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/api/EndpointSelectable.java b/streampipes-model/src/main/java/org/apache/streampipes/model/api/EndpointSelectable.java
new file mode 100644
index 0000000000..b8bc01b2ad
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/api/EndpointSelectable.java
@@ -0,0 +1,69 @@
+/*
+ * 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.model.api;
+
+/**
+ * Interface for pipeline elements which are invoked upon pipeline start (processors, sinks, sets)
+ */
+public interface EndpointSelectable {
+
+  /**
+   * Gets the name of this selectable endpoint.
+   *
+   * @return The name of this object.
+   */
+  String getName();
+
+  /**
+   * Gets the URL of the currently selected endpoint.
+   *
+   * @return endpoint URL.
+   */
+  String getSelectedEndpointUrl();
+
+  /**
+   * Sets the URL of the currently selected endpoint.
+   *
+   * @param selectedEndpointUrl new URL.
+   */
+  void setSelectedEndpointUrl(String selectedEndpointUrl);
+
+  /**
+   * Gets the corresponding pipeline ID for the endpoint.
+   *
+   * @return corresponding pipeline ID.
+   */
+  String getCorrespondingPipeline();
+
+
+  /**
+   * Sets the ID of the corresponding pipeline.
+   *
+   * @param pipelineId The ID of the pipeline to set as corresponding to the currently selected endpoint.
+   */
+  void setCorrespondingPipeline(String pipelineId);
+
+  /**
+   * Gets the path detach that can be called to detatch the endpoint.
+   * When an endpoint is detached it usally is also sopped
+   *
+   * @return detach path.
+   */
+  String getDetachPath();
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java b/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
index 3a87cc5ba2..0d64297fc9 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java
@@ -18,19 +18,21 @@
 
 package org.apache.streampipes.model.base;
 
+import org.apache.streampipes.commons.constants.InstanceIdExtractor;
 import org.apache.streampipes.logging.LoggerFactory;
 import org.apache.streampipes.logging.api.Logger;
 import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.api.EndpointSelectable;
 import org.apache.streampipes.model.grounding.EventGrounding;
 import org.apache.streampipes.model.monitoring.ElementStatusInfoSettings;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.model.util.Cloner;
 
-import java.util.List;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 
-public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity {
+import java.util.List;
 
-  private static final long serialVersionUID = 2727573914765473470L;
+public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity implements EndpointSelectable {
 
   protected List<SpDataStream> inputStreams;
 
@@ -120,10 +122,12 @@ public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity
     this.supportedGrounding = supportedGrounding;
   }
 
+  @Override
   public String getCorrespondingPipeline() {
     return correspondingPipeline;
   }
 
+  @Override
   public void setCorrespondingPipeline(String correspondingPipeline) {
     this.correspondingPipeline = correspondingPipeline;
   }
@@ -168,14 +172,22 @@ public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity
     this.uncompleted = uncompleted;
   }
 
+  @Override
   public String getSelectedEndpointUrl() {
     return selectedEndpointUrl;
   }
 
+  @Override
   public void setSelectedEndpointUrl(String selectedEndpointUrl) {
     this.selectedEndpointUrl = selectedEndpointUrl;
   }
 
+  @Override
+  @JsonIgnore
+  public String getDetachPath() {
+    return "/" + InstanceIdExtractor.extractId(getElementId());
+  }
+
   //public Logger getLogger(Class clazz, PeConfig peConfig) {
   public Logger getLogger(Class clazz) {
     //return LoggerFactory.getPeLogger(clazz, getCorrespondingPipeline(), getUri(), peConfig);
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/message/PipelineStatusMessage.java b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineStatusMessage.java
index a688a101cb..6fa8f7816c 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineStatusMessage.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineStatusMessage.java
@@ -28,13 +28,14 @@ public class PipelineStatusMessage {
   private String messageType;
   private String message;
 
-  public PipelineStatusMessage(String pipelineId, long timestamp,
-                               String messageType, String message) {
+  public PipelineStatusMessage(String pipelineId,
+                               long timestamp,
+                               PipelineStatusMessageType message) {
     super();
     this.pipelineId = pipelineId;
     this.timestamp = timestamp;
-    this.messageType = messageType;
-    this.message = message;
+    this.messageType = message.title();
+    this.message = message.description();
   }
 
   public String getPipelineId() {
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineOperationStatus.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineOperationStatus.java
index 25b3c5b275..8e42372e22 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineOperationStatus.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineOperationStatus.java
@@ -33,7 +33,9 @@ public class PipelineOperationStatus {
 
   private List<PipelineElementStatus> elementStatus;
 
-  public PipelineOperationStatus(String pipelineId, String pipelineName, String title,
+  public PipelineOperationStatus(String pipelineId,
+                                 String pipelineName,
+                                 String title,
                                  List<PipelineElementStatus> elementStatus) {
     super();
     this.title = title;
@@ -46,6 +48,12 @@ public class PipelineOperationStatus {
     this.elementStatus = new ArrayList<>();
   }
 
+  public PipelineOperationStatus(String pipelineId, String pipelineName) {
+    this();
+    this.pipelineId = pipelineId;
+    this.pipelineName = pipelineName;
+  }
+
   public String getPipelineId() {
     return pipelineId;
   }
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-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionInfo.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionInfo.java
new file mode 100644
index 0000000000..ba13ab7b0f
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionInfo.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.execution;
+
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class PipelineExecutionInfo {
+
+  private final List<NamedStreamPipesEntity> failedServices;
+  private final List<InvocableStreamPipesEntity> processorsAndSinks;
+  private final List<SpDataSet> dataSets;
+
+  private PipelineOperationStatus pipelineOperationStatus;
+
+  private final String pipelineId;
+
+  public static PipelineExecutionInfo create(Pipeline pipeline) {
+    return new PipelineExecutionInfo(pipeline);
+  }
+
+  private PipelineExecutionInfo(Pipeline pipeline) {
+    this.failedServices = new ArrayList<>();
+    this.processorsAndSinks = findProcessorsAndSinks(pipeline);
+    this.dataSets = findDataSets(pipeline);
+    this.pipelineOperationStatus = new PipelineOperationStatus();
+    this.pipelineId = pipeline.getPipelineId();
+  }
+
+  private List<InvocableStreamPipesEntity> findProcessorsAndSinks(Pipeline pipeline) {
+    return Stream
+        .concat(
+            pipeline.getSepas().stream(),
+            pipeline.getActions().stream()
+        ).collect(Collectors.toList());
+  }
+
+  private List<SpDataSet> findDataSets(Pipeline pipeline) {
+    return pipeline
+        .getStreams()
+        .stream()
+        .filter(s -> s instanceof SpDataSet)
+        .map(s -> new SpDataSet((SpDataSet) s))
+        .toList();
+  }
+
+  public void addFailedPipelineElement(NamedStreamPipesEntity failedElement) {
+    this.failedServices.add(failedElement);
+  }
+
+  public void addDataSets(List<SpDataSet> dataSets) {
+    this.dataSets.addAll(dataSets);
+  }
+
+  public List<SpDataSet> getDataSets() {
+    return dataSets;
+  }
+
+  public List<NamedStreamPipesEntity> getFailedServices() {
+    return failedServices;
+  }
+
+  public List<InvocableStreamPipesEntity> getProcessorsAndSinks() {
+    return processorsAndSinks;
+  }
+
+  public void applyPipelineOperationStatus(PipelineOperationStatus status) {
+    this.pipelineOperationStatus = status;
+  }
+
+  public PipelineOperationStatus getPipelineOperationStatus() {
+    return this.pipelineOperationStatus;
+  }
+
+  public String getPipelineId() {
+    return pipelineId;
+  }
+
+  public boolean isOperationSuccessful() {
+    return failedServices.size() == 0 && pipelineOperationStatus.isSuccess();
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionTaskFactory.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionTaskFactory.java
new file mode 100644
index 0000000000..3fab1d07e6
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionTaskFactory.java
@@ -0,0 +1,60 @@
+/*
+ * 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.manager.execution;
+
+import org.apache.streampipes.manager.execution.http.DetachPipelineElementSubmitter;
+import org.apache.streampipes.manager.execution.http.InvokePipelineElementSubmitter;
+import org.apache.streampipes.manager.execution.provider.CurrentPipelineElementProvider;
+import org.apache.streampipes.manager.execution.provider.StoredPipelineElementProvider;
+import org.apache.streampipes.manager.execution.task.AfterInvocationTask;
+import org.apache.streampipes.manager.execution.task.DiscoverEndpointsTask;
+import org.apache.streampipes.manager.execution.task.PipelineExecutionTask;
+import org.apache.streampipes.manager.execution.task.SecretEncryptionTask;
+import org.apache.streampipes.manager.execution.task.StorePipelineStatusTask;
+import org.apache.streampipes.manager.execution.task.SubmitRequestTask;
+import org.apache.streampipes.manager.execution.task.UpdateGroupIdTask;
+import org.apache.streampipes.model.message.PipelineStatusMessageType;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.resource.management.secret.SecretProvider;
+
+import java.util.List;
+
+public class PipelineExecutionTaskFactory {
+
+  public static List<PipelineExecutionTask> makeStartPipelineTasks(Pipeline pipeline) {
+    return List.of(
+        new UpdateGroupIdTask(),
+        new SecretEncryptionTask(SecretProvider.getDecryptionService()),
+        new DiscoverEndpointsTask(),
+        new SubmitRequestTask(new InvokePipelineElementSubmitter(pipeline), new CurrentPipelineElementProvider()),
+        new SecretEncryptionTask(SecretProvider.getEncryptionService()),
+        new AfterInvocationTask(PipelineStatusMessageType.PIPELINE_STARTED),
+        new StorePipelineStatusTask(true, false)
+    );
+  }
+
+  public static List<PipelineExecutionTask> makeStopPipelineTasks(Pipeline pipeline,
+                                                                  boolean forceStop) {
+    return List.of(
+        new SubmitRequestTask(new DetachPipelineElementSubmitter(pipeline), new StoredPipelineElementProvider()),
+        new AfterInvocationTask(PipelineStatusMessageType.PIPELINE_STOPPED),
+        new StorePipelineStatusTask(false, forceStop)
+    );
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutor.java
new file mode 100644
index 0000000000..bc14e46f58
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutor.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.execution;
+
+import org.apache.streampipes.manager.execution.task.PipelineExecutionTask;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+
+public class PipelineExecutor {
+
+  private final Pipeline pipeline;
+  private final boolean forceStop;
+
+  public PipelineExecutor(Pipeline pipeline,
+                          boolean forceStop) {
+    this.pipeline = pipeline;
+    this.forceStop = forceStop;
+  }
+
+  public PipelineOperationStatus startPipeline() {
+    return executeOperation(PipelineExecutionTaskFactory.makeStartPipelineTasks(pipeline));
+  }
+
+  public PipelineOperationStatus stopPipeline() {
+    return executeOperation(PipelineExecutionTaskFactory.makeStopPipelineTasks(pipeline, forceStop));
+  }
+
+  private PipelineOperationStatus executeOperation(List<PipelineExecutionTask> executionTasks) {
+    var executionInfo = PipelineExecutionInfo.create(pipeline);
+    executionTasks
+        .forEach(task -> {
+          if (task.shouldExecute(executionInfo)) {
+            task.executeTask(pipeline, executionInfo);
+          }
+        });
+    return executionInfo.getPipelineOperationStatus();
+  }
+
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java
new file mode 100644
index 0000000000..51760b2529
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.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.manager.execution.endpoint;
+
+import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants;
+import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
+import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ExtensionsServiceEndpointProvider {
+
+  public String findSelectedEndpoint(InvocableStreamPipesEntity g) throws NoServiceEndpointsAvailableException {
+    return new ExtensionsServiceEndpointGenerator(
+        g.getAppId(),
+        ExtensionsServiceEndpointUtils.getPipelineElementType(g))
+        .getEndpointResourceUrl();
+  }
+
+  public String findSelectedEndpoint(SpDataSet ds) throws NoServiceEndpointsAvailableException {
+    String appId = ds.getAppId() != null ? ds.getAppId() : ds.getCorrespondingAdapterId();
+    if (ds.isInternallyManaged()) {
+      return getConnectMasterSourcesUrl();
+    } else {
+      return new ExtensionsServiceEndpointGenerator(appId, SpServiceUrlProvider.DATA_SET)
+          .getEndpointResourceUrl();
+    }
+  }
+
+  private String getConnectMasterSourcesUrl() throws NoServiceEndpointsAvailableException {
+    List<String> connectMasterEndpoints = SpServiceDiscovery.getServiceDiscovery()
+        .getServiceEndpoints(DefaultSpServiceGroups.CORE, true,
+            Collections.singletonList(DefaultSpServiceTags.CONNECT_MASTER.asString()));
+    if (connectMasterEndpoints.size() > 0) {
+      return connectMasterEndpoints.get(0) + GlobalStreamPipesConstants.CONNECT_MASTER_SOURCES_ENDPOINT;
+    } else {
+      throw new NoServiceEndpointsAvailableException("Could not find any available connect master service endpoint");
+    }
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachPipelineElementSubmitter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachPipelineElementSubmitter.java
new file mode 100644
index 0000000000..47651cda38
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachPipelineElementSubmitter.java
@@ -0,0 +1,53 @@
+/*
+ * 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.manager.execution.http;
+
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+
+import java.util.List;
+
+public class DetachPipelineElementSubmitter extends PipelineElementSubmitter {
+
+  public DetachPipelineElementSubmitter(Pipeline pipeline) {
+    super(pipeline);
+  }
+
+  @Override
+  protected PipelineElementStatus submitElement(EndpointSelectable pipelineElement) {
+    return performDetach(pipelineElement);
+  }
+
+  @Override
+  protected boolean shouldSubmitDataSets() {
+    return true;
+  }
+
+  @Override
+  protected void onSuccess() {
+    status.setTitle("Pipeline " + pipelineName + " successfully stopped");
+  }
+
+  @Override
+  protected void onFailure(List<InvocableStreamPipesEntity> processorsAndSinks) {
+    status.setTitle("Could not stop all pipeline elements of pipeline " + pipelineName + ".");
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
deleted file mode 100644
index 59598fa146..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
+++ /dev/null
@@ -1,132 +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.manager.execution.http;
-
-import org.apache.streampipes.commons.constants.InstanceIdExtractor;
-import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.pipeline.PipelineElementStatus;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
-public class GraphSubmitter {
-
-  private List<InvocableStreamPipesEntity> graphs;
-  private List<SpDataSet> dataSets;
-
-  private String pipelineId;
-  private String pipelineName;
-
-  private static final Logger LOG = LoggerFactory.getLogger(GraphSubmitter.class);
-
-  public GraphSubmitter(String pipelineId,
-                        String pipelineName,
-                        List<InvocableStreamPipesEntity> graphs,
-                        List<SpDataSet> dataSets) {
-    this.graphs = graphs != null ? graphs : new ArrayList<>();
-    this.pipelineId = pipelineId;
-    this.pipelineName = pipelineName;
-    this.dataSets = dataSets != null ? dataSets : new ArrayList<>();
-  }
-
-  public PipelineOperationStatus invokeGraphs() {
-    PipelineOperationStatus status = new PipelineOperationStatus();
-    status.setPipelineId(pipelineId);
-    status.setPipelineName(pipelineName);
-
-
-    graphs.forEach(g -> status.addPipelineElementStatus(performInvocation(g)));
-    if (status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess)) {
-      dataSets.forEach(dataSet ->
-          status.addPipelineElementStatus
-              (performInvocation(dataSet)));
-    }
-    status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
-
-    if (status.isSuccess()) {
-      status.setTitle("Pipeline " + pipelineName + " successfully started");
-    } else {
-      LOG.info("Could not start pipeline, initializing rollback...");
-      rollbackInvokedPipelineElements(status);
-      status.setTitle("Could not start pipeline " + pipelineName + ".");
-    }
-    return status;
-  }
-
-  private void rollbackInvokedPipelineElements(PipelineOperationStatus status) {
-    for (PipelineElementStatus s : status.getElementStatus()) {
-      if (s.isSuccess()) {
-        Optional<InvocableStreamPipesEntity> graph = findGraph(s.getElementId());
-        graph.ifPresent(g -> {
-          LOG.info("Rolling back element " + g.getElementId());
-          performDetach(g);
-        });
-      }
-    }
-  }
-
-  private Optional<InvocableStreamPipesEntity> findGraph(String elementId) {
-    return graphs.stream().filter(g -> g.getBelongsTo().equals(elementId)).findFirst();
-  }
-
-  public PipelineOperationStatus detachGraphs() {
-    PipelineOperationStatus status = new PipelineOperationStatus();
-    status.setPipelineId(pipelineId);
-    status.setPipelineName(pipelineName);
-
-    graphs.forEach(g -> status.addPipelineElementStatus(performDetach(g)));
-    dataSets.forEach(dataSet -> status.addPipelineElementStatus(performDetach(dataSet)));
-    status.setSuccess(status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess));
-
-    if (status.isSuccess()) {
-      status.setTitle("Pipeline " + pipelineName + " successfully stopped");
-    } else {
-      status.setTitle("Could not stop all pipeline elements of pipeline " + pipelineName + ".");
-    }
-
-    return status;
-  }
-
-  private PipelineElementStatus performInvocation(InvocableStreamPipesEntity entity) {
-    String endpointUrl = entity.getSelectedEndpointUrl();
-    return new HttpRequestBuilder(entity, endpointUrl, this.pipelineId).invoke();
-  }
-
-  private PipelineElementStatus performInvocation(SpDataSet dataset) {
-    String endpointUrl = dataset.getSelectedEndpointUrl();
-    return new HttpRequestBuilder(dataset, endpointUrl, this.pipelineId).invoke();
-  }
-
-  private PipelineElementStatus performDetach(InvocableStreamPipesEntity entity) {
-    String endpointUrl = entity.getSelectedEndpointUrl() + "/" + InstanceIdExtractor.extractId(entity.getElementId());
-    return new HttpRequestBuilder(entity, endpointUrl, this.pipelineId).detach();
-  }
-
-  private PipelineElementStatus performDetach(SpDataSet dataset) {
-    String endpointUrl = dataset.getSelectedEndpointUrl() + "/" + dataset.getCorrespondingAdapterId() + "/"
-        + dataset.getDatasetInvocationId();
-    return new HttpRequestBuilder(dataset, endpointUrl, this.pipelineId).detach();
-  }
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java
deleted file mode 100644
index d9dd5a3130..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/HttpRequestBuilder.java
+++ /dev/null
@@ -1,98 +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.manager.execution.http;
-
-import org.apache.streampipes.manager.util.AuthTokenUtils;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-import org.apache.streampipes.model.pipeline.PipelineElementStatus;
-import org.apache.streampipes.serializers.json.JacksonSerializer;
-
-import com.google.gson.JsonSyntaxException;
-import org.apache.http.client.fluent.Request;
-import org.apache.http.client.fluent.Response;
-import org.apache.http.entity.ContentType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-public class HttpRequestBuilder {
-
-  private final NamedStreamPipesEntity payload;
-  private final String endpointUrl;
-  private String pipelineId;
-
-  private static final Logger LOG = LoggerFactory.getLogger(HttpRequestBuilder.class);
-
-  public HttpRequestBuilder(NamedStreamPipesEntity payload,
-                            String endpointUrl,
-                            String pipelineId) {
-    this.payload = payload;
-    this.endpointUrl = endpointUrl;
-    this.pipelineId = pipelineId;
-  }
-
-  public PipelineElementStatus invoke() {
-    LOG.info("Invoking element: " + endpointUrl);
-    try {
-      String jsonDocument = toJson();
-      Response httpResp =
-          Request.Post(endpointUrl)
-              .addHeader("Authorization", AuthTokenUtils.getAuthToken(this.pipelineId))
-              .bodyString(jsonDocument, ContentType.APPLICATION_JSON)
-              .connectTimeout(10000)
-              .execute();
-      return handleResponse(httpResp);
-    } catch (Exception e) {
-      LOG.error("Could not perform invocation request", e);
-      return new PipelineElementStatus(endpointUrl, payload.getName(), false, e.getMessage());
-    }
-  }
-
-  public PipelineElementStatus detach() {
-    try {
-      Response httpResp = Request.Delete(endpointUrl)
-          .addHeader("Authorization", AuthTokenUtils.getAuthToken(this.pipelineId))
-          .connectTimeout(10000).execute();
-      return handleResponse(httpResp);
-    } catch (Exception e) {
-      LOG.error("Could not stop pipeline {}", endpointUrl, e);
-      return new PipelineElementStatus(endpointUrl, payload.getName(), false, e.getMessage());
-    }
-  }
-
-  private PipelineElementStatus handleResponse(Response httpResp) throws JsonSyntaxException, IOException {
-    String resp = httpResp.returnContent().asString();
-    org.apache.streampipes.model.Response streamPipesResp = JacksonSerializer
-        .getObjectMapper()
-        .readValue(resp, org.apache.streampipes.model.Response.class);
-    return convert(streamPipesResp);
-  }
-
-  private String toJson() throws Exception {
-    return JacksonSerializer.getObjectMapper().writeValueAsString(payload);
-  }
-
-  private PipelineElementStatus convert(org.apache.streampipes.model.Response response) {
-    return new PipelineElementStatus(endpointUrl, payload.getName(), response.isSuccess(),
-        response.getOptionalMessage());
-  }
-
-
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeHttpRequest.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeHttpRequest.java
new file mode 100644
index 0000000000..1656e677b0
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeHttpRequest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.manager.execution.http;
+
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.http.client.fluent.Request;
+import org.apache.http.entity.ContentType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InvokeHttpRequest extends PipelineElementHttpRequest {
+
+  private static final Logger LOG = LoggerFactory.getLogger(InvokeHttpRequest.class);
+
+  @Override
+  protected Request initRequest(EndpointSelectable pipelineElement,
+                             String endpointUrl) throws JsonProcessingException {
+    LOG.info("Invoking element: " + endpointUrl);
+    return Request
+        .Post(endpointUrl)
+        .bodyString(toJson(pipelineElement), ContentType.APPLICATION_JSON);
+  }
+
+  @Override
+  protected void logError(String endpointUrl,
+                       String pipelineElementName,
+                       String exceptionMessage) {
+    LOG.error("Could not perform invocation request at {} for pipeline element {}: {}",
+        endpointUrl, pipelineElementName, exceptionMessage);
+  }
+
+  private String toJson(EndpointSelectable pipelineElement) throws JsonProcessingException {
+    return JacksonSerializer.getObjectMapper().writeValueAsString(pipelineElement);
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
new file mode 100644
index 0000000000..d746d23196
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
@@ -0,0 +1,81 @@
+/*
+ * 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.manager.execution.http;
+
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Optional;
+
+public class InvokePipelineElementSubmitter extends PipelineElementSubmitter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(InvokePipelineElementSubmitter.class);
+
+  public InvokePipelineElementSubmitter(Pipeline pipeline) {
+    super(pipeline);
+  }
+
+  @Override
+  protected PipelineElementStatus submitElement(EndpointSelectable pipelineElement) {
+    String endpointUrl = pipelineElement.getSelectedEndpointUrl();
+    return new InvokeHttpRequest().execute(pipelineElement, endpointUrl, this.pipelineId);
+  }
+
+  @Override
+  protected boolean shouldSubmitDataSets() {
+    return isSuccess();
+  }
+
+  @Override
+  protected void onSuccess() {
+    status.setTitle("Pipeline " + pipelineName + " successfully started");
+  }
+
+  @Override
+  protected void onFailure(List<InvocableStreamPipesEntity> processorsAndSinks) {
+    LOG.info("Could not start pipeline, initializing rollback...");
+    rollbackInvokedPipelineElements(status, processorsAndSinks);
+    status.setTitle("Could not start pipeline " + pipelineName + ".");
+  }
+
+  private void rollbackInvokedPipelineElements(PipelineOperationStatus status,
+                                               List<InvocableStreamPipesEntity> pe) {
+    for (PipelineElementStatus s : status.getElementStatus()) {
+      if (s.isSuccess()) {
+        Optional<InvocableStreamPipesEntity> graph = findPipelineElements(s.getElementId(), pe);
+        graph.ifPresent(this::performDetach);
+      }
+    }
+  }
+
+  private Optional<InvocableStreamPipesEntity> findPipelineElements(String elementId,
+                                                                    List<InvocableStreamPipesEntity> pe) {
+    return pe
+        .stream()
+        .filter(g -> g.getBelongsTo().equals(elementId))
+        .findFirst();
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementHttpRequest.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementHttpRequest.java
new file mode 100644
index 0000000000..400626e54d
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementHttpRequest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.manager.execution.http;
+
+import org.apache.streampipes.manager.util.AuthTokenUtils;
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.gson.JsonSyntaxException;
+import org.apache.http.client.fluent.Request;
+import org.apache.http.client.fluent.Response;
+
+import java.io.IOException;
+
+public abstract class PipelineElementHttpRequest {
+
+  public PipelineElementStatus execute(EndpointSelectable pipelineElement,
+                                       String endpointUrl,
+                                       String pipelineId) {
+    try {
+      Response httpResp = initRequest(pipelineElement, endpointUrl)
+              .addHeader("Authorization", AuthTokenUtils.getAuthToken(pipelineId))
+              .connectTimeout(10000)
+              .execute();
+      return handleResponse(httpResp, pipelineElement, endpointUrl);
+    } catch (Exception e) {
+      logError(endpointUrl, pipelineElement.getName(), e.getMessage());
+      return new PipelineElementStatus(endpointUrl, pipelineElement.getName(), false, e.getMessage());
+    }
+  }
+
+  protected abstract Request initRequest(EndpointSelectable pipelineElement,
+                                      String endpointUrl) throws JsonProcessingException;
+
+  protected abstract void logError(String endpointUrl,
+                                String pipelineElementName,
+                                String exceptionMessage);
+
+  protected PipelineElementStatus handleResponse(Response httpResp,
+                                                 EndpointSelectable pipelineElement,
+                                                 String endpointUrl) throws JsonSyntaxException, IOException {
+    String resp = httpResp.returnContent().asString();
+    org.apache.streampipes.model.Response streamPipesResp = JacksonSerializer
+        .getObjectMapper()
+        .readValue(resp, org.apache.streampipes.model.Response.class);
+    return convert(streamPipesResp, endpointUrl, pipelineElement.getName());
+  }
+
+  private PipelineElementStatus convert(org.apache.streampipes.model.Response response,
+                                        String endpointUrl,
+                                        String pipelineElementName) {
+    return new PipelineElementStatus(endpointUrl, pipelineElementName, response.isSuccess(),
+        response.getOptionalMessage());
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementSubmitter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementSubmitter.java
new file mode 100644
index 0000000000..8be496b138
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementSubmitter.java
@@ -0,0 +1,82 @@
+/*
+ * 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.manager.execution.http;
+
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+
+public abstract class PipelineElementSubmitter {
+
+  protected final String pipelineId;
+  protected final String pipelineName;
+
+  protected final PipelineOperationStatus status;
+
+  public PipelineElementSubmitter(Pipeline pipeline) {
+    this.pipelineId = pipeline.getPipelineId();
+    this.pipelineName = pipeline.getName();
+    this.status = new PipelineOperationStatus(pipelineId, pipelineName);
+  }
+
+  public PipelineOperationStatus submit(List<InvocableStreamPipesEntity> processorsAndSinks,
+                                        List<SpDataSet> dataSets) {
+    // First, try handling all data processors and sinks
+    processorsAndSinks.forEach(g -> status.addPipelineElementStatus(submitElement(g)));
+
+    // Then,submit data sets always for detach operation and otherwise only in case of success
+    if (shouldSubmitDataSets()) {
+      dataSets.forEach(dataSet -> status.addPipelineElementStatus(submitElement(dataSet)));
+    }
+
+    applySuccess(processorsAndSinks);
+    return status;
+  }
+
+  protected boolean isSuccess() {
+    return status.getElementStatus().stream().allMatch(PipelineElementStatus::isSuccess);
+  }
+
+  protected void applySuccess(List<InvocableStreamPipesEntity> processorsAndSinks) {
+    status.setSuccess(isSuccess());
+    if (status.isSuccess()) {
+      this.onSuccess();
+    } else {
+      this.onFailure(processorsAndSinks);
+    }
+  }
+
+  protected PipelineElementStatus performDetach(EndpointSelectable pipelineElement) {
+    String endpointUrl = pipelineElement.getSelectedEndpointUrl() + pipelineElement.getDetachPath();
+    return new DetachHttpRequest().execute(pipelineElement, endpointUrl, this.pipelineId);
+  }
+
+  protected abstract PipelineElementStatus submitElement(EndpointSelectable pipelineElement);
+
+  protected abstract boolean shouldSubmitDataSets();
+
+  protected abstract void onSuccess();
+
+  protected abstract void onFailure(List<InvocableStreamPipesEntity> processorsAndSinks);
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java
deleted file mode 100644
index 9ee3b8467c..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java
+++ /dev/null
@@ -1,243 +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.manager.execution.http;
-
-import org.apache.streampipes.commons.MD5;
-import org.apache.streampipes.commons.Utils;
-import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants;
-import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
-import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
-import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
-import org.apache.streampipes.manager.util.TemporaryGraphStorage;
-import org.apache.streampipes.model.SpDataSet;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.message.PipelineStatusMessage;
-import org.apache.streampipes.model.message.PipelineStatusMessageType;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.model.pipeline.PipelineElementStatus;
-import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
-import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import org.apache.streampipes.resource.management.secret.SecretProvider;
-import org.apache.streampipes.storage.api.IPipelineStorage;
-import org.apache.streampipes.storage.management.StorageDispatcher;
-import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
-import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceGroups;
-import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTags;
-import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
-
-import org.lightcouch.DocumentConflictException;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.stream.Collectors;
-
-public class PipelineExecutor {
-
-  private final Pipeline pipeline;
-  private final boolean storeStatus;
-  private final boolean forceStop;
-
-  public PipelineExecutor(Pipeline pipeline,
-                          boolean storeStatus,
-                          boolean forceStop) {
-    this.pipeline = pipeline;
-    this.storeStatus = storeStatus;
-    this.forceStop = forceStop;
-  }
-
-  public PipelineOperationStatus startPipeline() {
-
-    pipeline.getSepas().forEach(this::updateGroupIds);
-    pipeline.getActions().forEach(this::updateGroupIds);
-
-    List<DataProcessorInvocation> sepas = pipeline.getSepas();
-    List<DataSinkInvocation> secs = pipeline.getActions();
-
-    List<SpDataSet> dataSets = pipeline
-        .getStreams()
-        .stream()
-        .filter(s -> s instanceof SpDataSet)
-        .map(s -> new SpDataSet((SpDataSet) s))
-        .collect(Collectors.toList());
-
-    List<NamedStreamPipesEntity> failedServices = new ArrayList<>();
-
-    dataSets.forEach(ds -> {
-      ds.setCorrespondingPipeline(pipeline.getPipelineId());
-      try {
-        ds.setSelectedEndpointUrl(findSelectedEndpoint(ds));
-      } catch (NoServiceEndpointsAvailableException e) {
-        failedServices.add(ds);
-      }
-    });
-
-    List<InvocableStreamPipesEntity> graphs = new ArrayList<>();
-    graphs.addAll(sepas);
-    graphs.addAll(secs);
-
-    decryptSecrets(graphs);
-
-    graphs.forEach(g -> {
-      try {
-        g.setSelectedEndpointUrl(findSelectedEndpoint(g));
-        g.setCorrespondingPipeline(pipeline.getPipelineId());
-      } catch (NoServiceEndpointsAvailableException e) {
-        failedServices.add(g);
-      }
-    });
-
-    PipelineOperationStatus status;
-    if (failedServices.size() == 0) {
-
-      status = new GraphSubmitter(pipeline.getPipelineId(),
-          pipeline.getName(), graphs, dataSets)
-          .invokeGraphs();
-
-      encryptSecrets(graphs);
-
-      if (status.isSuccess()) {
-        storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets);
-
-        PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(),
-            new PipelineStatusMessage(pipeline.getPipelineId(), System.currentTimeMillis(),
-                PipelineStatusMessageType.PIPELINE_STARTED.title(),
-                PipelineStatusMessageType.PIPELINE_STARTED.description()));
-
-        if (storeStatus) {
-          pipeline.setHealthStatus(PipelineHealthStatus.OK);
-          setPipelineStarted(pipeline);
-        }
-      }
-    } else {
-      List<PipelineElementStatus> pe = failedServices.stream().map(fs ->
-          new PipelineElementStatus(fs.getElementId(),
-              fs.getName(),
-              false,
-              "No active supporting service found")).collect(Collectors.toList());
-      status = new PipelineOperationStatus(pipeline.getPipelineId(),
-          pipeline.getName(),
-          "Could not start pipeline " + pipeline.getName() + ".",
-          pe);
-    }
-    return status;
-  }
-
-  private String findSelectedEndpoint(InvocableStreamPipesEntity g) throws NoServiceEndpointsAvailableException {
-    return new ExtensionsServiceEndpointGenerator(
-        g.getAppId(),
-        ExtensionsServiceEndpointUtils.getPipelineElementType(g))
-        .getEndpointResourceUrl();
-  }
-
-  private String findSelectedEndpoint(SpDataSet ds) throws NoServiceEndpointsAvailableException {
-    String appId = ds.getAppId() != null ? ds.getAppId() : ds.getCorrespondingAdapterId();
-    if (ds.isInternallyManaged()) {
-      return getConnectMasterSourcesUrl();
-    } else {
-      return new ExtensionsServiceEndpointGenerator(appId, SpServiceUrlProvider.DATA_SET)
-          .getEndpointResourceUrl();
-    }
-  }
-
-  private String getConnectMasterSourcesUrl() throws NoServiceEndpointsAvailableException {
-    List<String> connectMasterEndpoints = SpServiceDiscovery.getServiceDiscovery()
-        .getServiceEndpoints(DefaultSpServiceGroups.CORE, true,
-            Collections.singletonList(DefaultSpServiceTags.CONNECT_MASTER.asString()));
-    if (connectMasterEndpoints.size() > 0) {
-      return connectMasterEndpoints.get(0) + GlobalStreamPipesConstants.CONNECT_MASTER_SOURCES_ENDPOINT;
-    } else {
-      throw new NoServiceEndpointsAvailableException("Could not find any available connect master service endpoint");
-    }
-  }
-
-  private void updateGroupIds(InvocableStreamPipesEntity entity) {
-    entity.getInputStreams()
-        .stream()
-        .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)
-        .map(is -> is.getEventGrounding().getTransportProtocol())
-        .map(KafkaTransportProtocol.class::cast)
-        .forEach(tp -> tp.setGroupId(Utils.filterSpecialChar(pipeline.getName()) + MD5.crypt(tp.getElementId())));
-  }
-
-  private void decryptSecrets(List<InvocableStreamPipesEntity> graphs) {
-    SecretProvider.getDecryptionService().apply(graphs);
-  }
-
-  private void encryptSecrets(List<InvocableStreamPipesEntity> graphs) {
-    SecretProvider.getEncryptionService().apply(graphs);
-  }
-
-  public PipelineOperationStatus stopPipeline() {
-    List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
-    List<SpDataSet> dataSets = TemporaryGraphStorage.datasetStorage.get(pipeline.getPipelineId());
-
-    PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(),
-        pipeline.getName(), graphs, dataSets)
-        .detachGraphs();
-
-    if (status.isSuccess()) {
-      PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(),
-          new PipelineStatusMessage(pipeline.getPipelineId(),
-              System.currentTimeMillis(),
-              PipelineStatusMessageType.PIPELINE_STOPPED.title(),
-              PipelineStatusMessageType.PIPELINE_STOPPED.description()));
-
-    }
-
-    if (status.isSuccess() || forceStop) {
-      if (storeStatus) {
-        setPipelineStopped(pipeline);
-      }
-    }
-    return status;
-  }
-
-  private void setPipelineStarted(Pipeline pipeline) {
-    pipeline.setRunning(true);
-    pipeline.setStartedAt(new Date().getTime());
-    try {
-      getPipelineStorageApi().updatePipeline(pipeline);
-    } catch (DocumentConflictException dce) {
-      //dce.printStackTrace();
-    }
-  }
-
-  private void setPipelineStopped(Pipeline pipeline) {
-    pipeline.setRunning(false);
-    getPipelineStorageApi().updatePipeline(pipeline);
-  }
-
-  private void storeInvocationGraphs(String pipelineId, List<InvocableStreamPipesEntity> graphs,
-                                     List<SpDataSet> dataSets) {
-    TemporaryGraphStorage.graphStorage.put(pipelineId, graphs);
-    TemporaryGraphStorage.datasetStorage.put(pipelineId, dataSets);
-  }
-
-  private IPipelineStorage getPipelineStorageApi() {
-    return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
-  }
-
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/CurrentPipelineElementProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/CurrentPipelineElementProvider.java
new file mode 100644
index 0000000000..aa1aeaeccc
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/CurrentPipelineElementProvider.java
@@ -0,0 +1,41 @@
+/*
+ * 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.manager.execution.provider;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+
+import java.util.List;
+
+/**
+ * Provides pipeline elements from the pipeline of interest (for start actions)
+ */
+
+public class CurrentPipelineElementProvider implements PipelineElementProvider {
+  @Override
+  public List<InvocableStreamPipesEntity> getProcessorsAndSinks(PipelineExecutionInfo executionInfo) {
+    return executionInfo.getProcessorsAndSinks();
+  }
+
+  @Override
+  public List<SpDataSet> getDataSets(PipelineExecutionInfo executionInfo) {
+    return executionInfo.getDataSets();
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/StoredPipelineElementProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/StoredPipelineElementProvider.java
new file mode 100644
index 0000000000..ad5860711e
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/StoredPipelineElementProvider.java
@@ -0,0 +1,42 @@
+/*
+ * 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.manager.execution.provider;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.manager.storage.RunningPipelineElementStorage;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+
+import java.util.List;
+
+/**
+ * Provides pipeline elements from the cache (for stop actions)
+ */
+
+public class StoredPipelineElementProvider implements PipelineElementProvider {
+  @Override
+  public List<InvocableStreamPipesEntity> getProcessorsAndSinks(PipelineExecutionInfo executionInfo) {
+    return RunningPipelineElementStorage.runningProcessorsAndSinks.get(executionInfo.getPipelineId());
+  }
+
+  @Override
+  public List<SpDataSet> getDataSets(PipelineExecutionInfo executionInfo) {
+    return RunningPipelineElementStorage.runningDataSets.get(executionInfo.getPipelineId());
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/AfterInvocationTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/AfterInvocationTask.java
new file mode 100644
index 0000000000..a7d967d2a8
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/AfterInvocationTask.java
@@ -0,0 +1,65 @@
+/*
+ * 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.manager.execution.task;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
+import org.apache.streampipes.manager.storage.RunningPipelineElementStorage;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.message.PipelineStatusMessage;
+import org.apache.streampipes.model.message.PipelineStatusMessageType;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+import java.util.List;
+
+public class AfterInvocationTask implements PipelineExecutionTask {
+
+  private final PipelineStatusMessageType statusMessageType;
+
+  public AfterInvocationTask(PipelineStatusMessageType statusMessageType) {
+    this.statusMessageType = statusMessageType;
+  }
+
+  @Override
+  public boolean shouldExecute(PipelineExecutionInfo executionInfo) {
+    return executionInfo.isOperationSuccessful();
+  }
+
+  @Override
+  public void executeTask(Pipeline pipeline,
+                          PipelineExecutionInfo executionInfo) {
+    var graphs = executionInfo.getProcessorsAndSinks();
+    var dataSets = executionInfo.getDataSets();
+    storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets);
+    addPipelineStatus(pipeline);
+  }
+
+  private void storeInvocationGraphs(String pipelineId,
+                                     List<InvocableStreamPipesEntity> graphs,
+                                     List<SpDataSet> dataSets) {
+    RunningPipelineElementStorage.runningProcessorsAndSinks.put(pipelineId, graphs);
+    RunningPipelineElementStorage.runningDataSets.put(pipelineId, dataSets);
+  }
+
+  private void addPipelineStatus(Pipeline pipeline) {
+    PipelineStatusManager.addPipelineStatus(pipeline.getPipelineId(),
+        new PipelineStatusMessage(pipeline.getPipelineId(), System.currentTimeMillis(), statusMessageType));
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java
new file mode 100644
index 0000000000..713644650f
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java
@@ -0,0 +1,88 @@
+/*
+ * 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.manager.execution.task;
+
+import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointProvider;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.api.EndpointSelectable;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DiscoverEndpointsTask implements PipelineExecutionTask {
+  @Override
+  public void executeTask(Pipeline pipeline,
+                          PipelineExecutionInfo executionInfo) {
+    var processorsAndSinks = executionInfo.getProcessorsAndSinks();
+    var dataSets = executionInfo.getDataSets();
+
+    processorsAndSinks.forEach(el -> {
+      try {
+        var endpointUrl = findSelectedEndpoint(el);
+        applyEndpointAndPipeline(pipeline.getPipelineId(), el, endpointUrl);
+      } catch (NoServiceEndpointsAvailableException e) {
+        executionInfo.addFailedPipelineElement(el);
+      }
+    });
+    dataSets.forEach(ds -> {
+      try {
+        var endpointUrl = findSelectedDsEndpoint(ds);
+        applyEndpointAndPipeline(pipeline.getPipelineId(), ds, endpointUrl);
+      } catch (NoServiceEndpointsAvailableException e) {
+        executionInfo.addFailedPipelineElement(ds);
+      }
+    });
+
+    var failedServices = executionInfo.getFailedServices();
+    if (executionInfo.getFailedServices().size() > 0) {
+      List<PipelineElementStatus> pe = failedServices
+          .stream()
+          .map(fs -> new PipelineElementStatus(fs.getElementId(), fs.getName(), false,
+              "No active extensions service found which provides this pipeline element"))
+          .collect(Collectors.toList());
+      var status = new PipelineOperationStatus(pipeline.getPipelineId(),
+          pipeline.getName(),
+          "Could not start pipeline " + pipeline.getName() + ".",
+          pe);
+      executionInfo.applyPipelineOperationStatus(status);
+    }
+  }
+
+  private void applyEndpointAndPipeline(String pipelineId,
+                                        EndpointSelectable pipelineElement,
+                                        String endpointUrl) {
+    pipelineElement.setSelectedEndpointUrl(endpointUrl);
+    pipelineElement.setCorrespondingPipeline(pipelineId);
+  }
+
+  private String findSelectedEndpoint(InvocableStreamPipesEntity pipelineElement)
+      throws NoServiceEndpointsAvailableException {
+    return new ExtensionsServiceEndpointProvider().findSelectedEndpoint(pipelineElement);
+  }
+
+  private String findSelectedDsEndpoint(SpDataSet dataSet) throws NoServiceEndpointsAvailableException {
+    return new ExtensionsServiceEndpointProvider().findSelectedEndpoint(dataSet);
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java
new file mode 100644
index 0000000000..df3c09f954
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.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.manager.execution.task;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
+import org.apache.streampipes.storage.api.IPipelineStorage;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import org.lightcouch.DocumentConflictException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+
+public class StorePipelineStatusTask implements PipelineExecutionTask {
+
+  private static final Logger LOG = LoggerFactory.getLogger(StorePipelineStatusTask.class);
+
+  private final boolean start;
+  private final boolean forceStop;
+
+  public StorePipelineStatusTask(boolean start,
+                                 boolean forceStop) {
+    this.start = start;
+    this.forceStop = forceStop;
+  }
+
+  @Override
+  public boolean shouldExecute(PipelineExecutionInfo executionInfo) {
+    return executionInfo.isOperationSuccessful() || forceStop;
+  }
+
+  @Override
+  public void executeTask(Pipeline pipeline,
+                          PipelineExecutionInfo executionInfo) {
+    if (this.start) {
+      pipeline.setHealthStatus(PipelineHealthStatus.OK);
+      setPipelineStarted(pipeline);
+    } else {
+      setPipelineStopped(pipeline);
+    }
+  }
+
+  private void setPipelineStarted(Pipeline pipeline) {
+    pipeline.setRunning(true);
+    pipeline.setStartedAt(new Date().getTime());
+    try {
+      getPipelineStorageApi().updatePipeline(pipeline);
+    } catch (DocumentConflictException dce) {
+      LOG.error("Could not update pipeline {}", pipeline.getPipelineId(), dce);
+    }
+  }
+
+  private void setPipelineStopped(Pipeline pipeline) {
+    pipeline.setRunning(false);
+    getPipelineStorageApi().updatePipeline(pipeline);
+  }
+
+  private IPipelineStorage getPipelineStorageApi() {
+    return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SubmitRequestTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SubmitRequestTask.java
new file mode 100644
index 0000000000..c29d749091
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SubmitRequestTask.java
@@ -0,0 +1,51 @@
+/*
+ * 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.manager.execution.task;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.manager.execution.http.PipelineElementSubmitter;
+import org.apache.streampipes.manager.execution.provider.PipelineElementProvider;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+public class SubmitRequestTask implements PipelineExecutionTask {
+
+  private final PipelineElementProvider elementProvider;
+  private final PipelineElementSubmitter submitter;
+
+  public SubmitRequestTask(PipelineElementSubmitter submitter,
+                           PipelineElementProvider elementProvider) {
+    this.elementProvider = elementProvider;
+    this.submitter = submitter;
+  }
+
+  @Override
+  public boolean shouldExecute(PipelineExecutionInfo executionInfo) {
+    return executionInfo.getFailedServices().size() == 0;
+  }
+
+  @Override
+  public void executeTask(Pipeline pipeline, PipelineExecutionInfo executionInfo) {
+    var processorsAndSinks = elementProvider.getProcessorsAndSinks(executionInfo);
+    var dataSets = elementProvider.getDataSets(executionInfo);
+
+    var status = submitter.submit(processorsAndSinks, dataSets);
+
+    executionInfo.applyPipelineOperationStatus(status);
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/UpdateGroupIdTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/UpdateGroupIdTask.java
new file mode 100644
index 0000000000..19fd879107
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/UpdateGroupIdTask.java
@@ -0,0 +1,46 @@
+/*
+ * 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.manager.execution.task;
+
+import org.apache.streampipes.commons.MD5;
+import org.apache.streampipes.commons.Utils;
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+public class UpdateGroupIdTask implements PipelineExecutionTask {
+  @Override
+  public void executeTask(Pipeline pipeline,
+                          PipelineExecutionInfo executionInfo) {
+    var sanitizedPipelineName = Utils.filterSpecialChar(pipeline.getName());
+    pipeline.getSepas().forEach(processor -> updateGroupIds(processor, sanitizedPipelineName));
+    pipeline.getActions().forEach(sink -> updateGroupIds(sink, sanitizedPipelineName));
+  }
+
+  private void updateGroupIds(InvocableStreamPipesEntity entity,
+                              String sanitizedPipelineName) {
+    entity.getInputStreams()
+        .stream()
+        .filter(is -> is.getEventGrounding().getTransportProtocol() instanceof KafkaTransportProtocol)
+        .map(is -> is.getEventGrounding().getTransportProtocol())
+        .map(KafkaTransportProtocol.class::cast)
+        .forEach(tp -> tp.setGroupId(sanitizedPipelineName + MD5.crypt(tp.getElementId())));
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
index 335d76c46a..5f0abc0672 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
@@ -22,8 +22,8 @@ import org.apache.streampipes.commons.constants.InstanceIdExtractor;
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
 import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
-import org.apache.streampipes.manager.execution.http.HttpRequestBuilder;
-import org.apache.streampipes.manager.util.TemporaryGraphStorage;
+import org.apache.streampipes.manager.execution.http.InvokeHttpRequest;
+import org.apache.streampipes.manager.storage.RunningPipelineElementStorage;
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
@@ -67,7 +67,10 @@ public class PipelineHealthCheck implements Runnable {
         List<String> failedInstances = new ArrayList<>();
         List<String> recoveredInstances = new ArrayList<>();
         List<String> pipelineNotifications = new ArrayList<>();
-        List<InvocableStreamPipesEntity> graphs = TemporaryGraphStorage.graphStorage.get(pipeline.getPipelineId());
+        List<InvocableStreamPipesEntity> graphs = RunningPipelineElementStorage
+            .runningProcessorsAndSinks
+            .get(pipeline.getPipelineId());
+
         graphs.forEach(graph -> {
           String instanceId = extractInstanceId(graph);
           if (allRunningInstances.stream().noneMatch(runningInstanceId -> runningInstanceId.equals(instanceId))) {
@@ -77,7 +80,7 @@ public class PipelineHealthCheck implements Runnable {
               boolean success;
               try {
                 endpointUrl = findEndpointUrl(graph);
-                success = new HttpRequestBuilder(graph, endpointUrl, pipeline.getPipelineId()).invoke().isSuccess();
+                success = new InvokeHttpRequest().execute(graph, endpointUrl, pipeline.getPipelineId()).isSuccess();
               } catch (NoServiceEndpointsAvailableException e) {
                 success = false;
               }
@@ -182,7 +185,7 @@ public class PipelineHealthCheck implements Runnable {
 
   private Map<String, List<InvocableStreamPipesEntity>> generateEndpointMap() {
     Map<String, List<InvocableStreamPipesEntity>> endpointMap = new HashMap<>();
-    TemporaryGraphStorage.graphStorage.forEach((pipelineId, graphs) ->
+    RunningPipelineElementStorage.runningProcessorsAndSinks.forEach((pipelineId, graphs) ->
         graphs.forEach(graph -> addEndpoint(endpointMap, graph)));
 
     return endpointMap;
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
index 6e45801ca3..15f5e80ba1 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java
@@ -22,13 +22,13 @@ import org.apache.streampipes.commons.exceptions.NoSuitableSepasAvailableExcepti
 import org.apache.streampipes.commons.exceptions.SepaParseException;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.manager.endpoint.EndpointItemFetcher;
-import org.apache.streampipes.manager.execution.http.PipelineExecutor;
-import org.apache.streampipes.manager.execution.http.PipelineStorageService;
+import org.apache.streampipes.manager.execution.PipelineExecutor;
 import org.apache.streampipes.manager.matching.DataSetGroundingSelector;
 import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
 import org.apache.streampipes.manager.recommender.ElementRecommender;
 import org.apache.streampipes.manager.remote.ContainerProvidedOptionsHandler;
 import org.apache.streampipes.manager.runtime.PipelineElementRuntimeInfoFetcher;
+import org.apache.streampipes.manager.storage.PipelineStorageService;
 import org.apache.streampipes.manager.template.PipelineTemplateGenerator;
 import org.apache.streampipes.manager.template.PipelineTemplateInvocationGenerator;
 import org.apache.streampipes.manager.template.PipelineTemplateInvocationHandler;
@@ -96,19 +96,8 @@ public class Operations {
     new PipelineStorageService(pipeline).updatePipeline();
   }
 
-  public static PipelineOperationStatus startPipeline(
-      Pipeline pipeline) {
-    return startPipeline(pipeline, true);
-  }
-
-  public static PipelineOperationStatus startPipeline(
-      Pipeline pipeline, boolean storeStatus) {
-    return new PipelineExecutor(pipeline, storeStatus, false).startPipeline();
-  }
-
-  public static PipelineOperationStatus stopPipeline(
-      Pipeline pipeline, boolean forceStop) {
-    return stopPipeline(pipeline, true, forceStop);
+  public static PipelineOperationStatus startPipeline(Pipeline pipeline) {
+    return new PipelineExecutor(pipeline, false).startPipeline();
   }
 
   public static List<PipelineOperationStatus> stopAllPipelines(boolean forceStop) {
@@ -125,9 +114,8 @@ public class Operations {
   }
 
   public static PipelineOperationStatus stopPipeline(Pipeline pipeline,
-                                                     boolean storeStatus,
                                                      boolean forceStop) {
-    return new PipelineExecutor(pipeline, storeStatus, forceStop).stopPipeline();
+    return new PipelineExecutor(pipeline, forceStop).stopPipeline();
   }
 
   public static List<ExtensionsServiceEndpointItem> getEndpointUriContents(List<ExtensionsServiceEndpoint> endpoints) {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
index bb37f1e733..7b5e8e67dd 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
@@ -17,11 +17,11 @@
  */
 package org.apache.streampipes.manager.preview;
 
-import org.apache.streampipes.commons.constants.InstanceIdExtractor;
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
 import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
-import org.apache.streampipes.manager.execution.http.HttpRequestBuilder;
+import org.apache.streampipes.manager.execution.http.DetachHttpRequest;
+import org.apache.streampipes.manager.execution.http.InvokeHttpRequest;
 import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
 import org.apache.streampipes.manager.operations.Operations;
 import org.apache.streampipes.model.SpDataSet;
@@ -92,7 +92,7 @@ public class PipelinePreview {
     graphs.forEach(g -> {
       try {
         g.setSelectedEndpointUrl(findSelectedEndpoint(g));
-        new HttpRequestBuilder(g, g.getSelectedEndpointUrl(), null).invoke();
+        new InvokeHttpRequest().execute(g, g.getSelectedEndpointUrl(), null);
       } catch (NoServiceEndpointsAvailableException e) {
         e.printStackTrace();
       }
@@ -101,8 +101,8 @@ public class PipelinePreview {
 
   private void detachGraphs(List<InvocableStreamPipesEntity> graphs) {
     graphs.forEach(g -> {
-      String endpointUrl = g.getSelectedEndpointUrl() + "/" + InstanceIdExtractor.extractId(g.getElementId());
-      new HttpRequestBuilder(g, endpointUrl, null).detach();
+      String endpointUrl = g.getSelectedEndpointUrl() + g.getDetachPath();
+      new DetachHttpRequest().execute(g, endpointUrl, null);
     });
   }
 
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
similarity index 98%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
index 5c42b5eb06..9d0f2b5824 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.manager.execution.http;
+package org.apache.streampipes.manager.storage;
 
 import org.apache.streampipes.manager.data.PipelineGraph;
 import org.apache.streampipes.manager.data.PipelineGraphBuilder;
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/util/TemporaryGraphStorage.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/RunningPipelineElementStorage.java
similarity index 78%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/util/TemporaryGraphStorage.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/RunningPipelineElementStorage.java
index 085d17e44b..360e6db90a 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/util/TemporaryGraphStorage.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/RunningPipelineElementStorage.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.manager.util;
+package org.apache.streampipes.manager.storage;
 
 import org.apache.streampipes.model.SpDataSet;
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
@@ -25,10 +25,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-public class TemporaryGraphStorage {
+public class RunningPipelineElementStorage {
 
-  public static Map<String, List<InvocableStreamPipesEntity>> graphStorage = new HashMap<>();
+  public static Map<String, List<InvocableStreamPipesEntity>> runningProcessorsAndSinks = new HashMap<>();
 
-  public static Map<String, List<SpDataSet>> datasetStorage = new HashMap<>();
+  public static Map<String, List<SpDataSet>> runningDataSets = new HashMap<>();
 
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/UserService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/UserService.java
deleted file mode 100644
index 0523158ff6..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/UserService.java
+++ /dev/null
@@ -1,101 +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.manager.storage;
-
-import org.apache.streampipes.model.client.user.Principal;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.storage.api.INoSqlStorage;
-import org.apache.streampipes.storage.api.IUserStorage;
-import org.apache.streampipes.storage.management.StorageDispatcher;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.stream.Collectors;
-
-public class UserService {
-
-  private IUserStorage userStorage;
-
-  public UserService(IUserStorage userStorage) {
-    this.userStorage = userStorage;
-  }
-
-  public List<Pipeline> getOwnPipelines(String email) {
-    return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getAllPipelines().stream().filter(p -> p
-            .getCreatedByUser()
-            .equals(email))
-        .collect(Collectors.toList());
-  }
-
-  public void deleteOwnSource(String username, String sourceId) {
-    if (checkUser(username)) {
-      Principal user = getPrincipal(username);
-      //user.getOwnSources().removeIf(a -> a.getElementId().equals(sourceId));
-      userStorage.updateUser(user);
-    }
-  }
-
-  /**
-   * Get actions/sepas/sources
-   */
-
-  public List<String> getOwnActionUris(String username) {
-    // TODO permissions
-    return new ArrayList<>();
-    //return userStorage.getUser(username)
-    // .getOwnActions().stream().map(r -> r.getElementId()).collect(Collectors.toList());
-  }
-
-  public List<String> getOwnSepaUris(String username) {
-    // TODO Permissions
-    return new ArrayList<>();
-    //return userStorage.getUser(username)
-    // .getOwnSepas().stream().map(r -> r.getElementId()).collect(Collectors.toList());
-  }
-
-
-  public List<String> getOwnSourceUris(String email) {
-    // TODO permissions
-    return new ArrayList<>();
-//    return userStorage
-//            .getUser(email)
-//            .getOwnSources()
-//            .stream()
-//            .map(r -> r.getElementId())
-//            .collect(Collectors.toList());
-  }
-
-  private Principal getPrincipal(String username) {
-    return userStorage.getUser(username);
-  }
-
-
-  /**
-   * @param username
-   * @return True if user exists exactly once, false otherwise
-   */
-  public boolean checkUser(String username) {
-    return userStorage.checkUser(username);
-  }
-
-  private INoSqlStorage getStorageManager() {
-    return StorageDispatcher.INSTANCE.getNoSqlStore();
-  }
-
-}
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-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java
index 33b2607b94..10c74f6192 100644
--- a/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java
+++ b/streampipes-rest-core-base/src/main/java/org/apache/streampipes/rest/core/base/impl/AbstractRestResource.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.rest.core.base.impl;
 
 import org.apache.streampipes.manager.endpoint.HttpJsonParser;
-import org.apache.streampipes.manager.storage.UserService;
 import org.apache.streampipes.model.message.ErrorMessage;
 import org.apache.streampipes.model.message.Message;
 import org.apache.streampipes.model.message.Notification;
@@ -68,10 +67,6 @@ public abstract class AbstractRestResource extends AbstractSharedRestInterface {
     return getNoSqlStorage().getUserStorageAPI();
   }
 
-  protected UserService getUserService() {
-    return new UserService(getUserStorage());
-  }
-
   protected IVisualizationStorage getVisualizationStorage() {
     return getNoSqlStorage().getVisualizationStorageApi();
   }
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 df54ff075c..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
diff --git a/ui/.prettierignore b/ui/.prettierignore
index 4b0f32f39f..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
diff --git a/ui/cypress/support/utils/connect/ConnectBtns.ts b/ui/cypress/support/utils/connect/ConnectBtns.ts
index 17887ff2a4..f5fbf14510 100644
--- a/ui/cypress/support/utils/connect/ConnectBtns.ts
+++ b/ui/cypress/support/utils/connect/ConnectBtns.ts
@@ -47,8 +47,7 @@ export class ConnectBtns {
             .parent();
     }
 
-    // =================================================================
-    // Format button options
+    // =====================  Format button options  ==========================
     public static json() {
         return cy.dataCy('connect-select-json-formats');
     }
@@ -82,4 +81,15 @@ export class ConnectBtns {
     }
 
     // =================================================================
+
+    // =====================  Adapter settings btns  ==========================
+    public static adapterSettingsStartAdapter() {
+        return cy.dataCy('adapter-settings-start-adapter-btn');
+    }
+
+    public static startAdapterNowCheckbox() {
+        return cy.dataCy('start-adapter-now-checkbox');
+    }
+
+    // ========================================================================
 }
diff --git a/ui/cypress/support/utils/connect/ConnectUtils.ts b/ui/cypress/support/utils/connect/ConnectUtils.ts
index feca0e0044..88ae97032b 100644
--- a/ui/cypress/support/utils/connect/ConnectUtils.ts
+++ b/ui/cypress/support/utils/connect/ConnectUtils.ts
@@ -30,6 +30,8 @@ import { ConnectBtns } from './ConnectBtns';
 export class ConnectUtils {
     public static testSpecificStreamAdapter(
         adapterConfiguration: SpecificAdapterInput,
+        starting = true,
+        successElement = 'sp-connect-adapter-live-preview',
     ) {
         ConnectUtils.goToConnect();
 
@@ -53,7 +55,11 @@ export class ConnectUtils {
 
         ConnectEventSchemaUtils.finishEventSchemaConfiguration();
 
-        ConnectUtils.startStreamAdapter(adapterConfiguration);
+        ConnectUtils.startStreamAdapter(
+            adapterConfiguration,
+            starting,
+            successElement,
+        );
     }
 
     public static testGenericStreamAdapter(
@@ -188,21 +194,27 @@ export class ConnectUtils {
         cy.get('#event-schema-next-button').click();
     }
 
-    public static startStreamAdapter(adapterInput: AdapterInput) {
-        ConnectUtils.startAdapter(
-            adapterInput,
-            'sp-connect-adapter-live-preview',
-        );
+    public static startStreamAdapter(
+        adapterInput: AdapterInput,
+        starting = true,
+        successElement = 'sp-connect-adapter-live-preview',
+    ) {
+        ConnectUtils.startAdapter(adapterInput, successElement, starting);
     }
 
     public static startSetAdapter(adapterInput: AdapterInput) {
         ConnectUtils.startAdapter(
             adapterInput,
             'sp-connect-adapter-set-success',
+            true,
         );
     }
 
-    public static startAdapter(adapterInput: AdapterInput, successElement) {
+    public static startAdapter(
+        adapterInput: AdapterInput,
+        successElement: string,
+        starting: boolean,
+    ) {
         // Set adapter name
         cy.dataCy('sp-adapter-name').type(adapterInput.adapterName);
 
@@ -215,8 +227,13 @@ export class ConnectUtils {
                 .click();
         }
 
-        // Start adapter
-        cy.get('#button-startAdapter').click();
+        // Deselect auto start of adapter
+        if (!starting) {
+            ConnectBtns.startAdapterNowCheckbox().parent().click();
+        }
+
+        ConnectBtns.adapterSettingsStartAdapter().click();
+
         cy.dataCy(successElement, { timeout: 60000 }).should('be.visible');
 
         this.closeAdapterPreview();
@@ -269,6 +286,26 @@ export class ConnectUtils {
         return adapterConfiguration;
     }
 
+    public static startAndValidateAdapter(amountOfProperties: number) {
+        ConnectBtns.startAdapter().should('not.be.disabled');
+
+        ConnectBtns.startAdapter().click();
+
+        // View data
+        ConnectBtns.infoAdapter().click();
+        cy.get('div').contains('Values').parent().click();
+
+        // Validate resulting event
+        cy.dataCy('sp-connect-adapter-live-preview', { timeout: 10000 }).should(
+            'be.visible',
+        );
+
+        // validate that three event properties
+        cy.get('.preview-row', { timeout: 10000 })
+            .its('length')
+            .should('eq', amountOfProperties);
+    }
+
     public static tearDownPreprocessingRuleTest(
         adapterConfiguration: AdapterInput,
         expectedFile: string,
diff --git a/ui/projects/streampipes/platform-services/src/lib/apis/adapter.service.ts b/ui/projects/streampipes/platform-services/src/lib/apis/adapter.service.ts
index 217e7c06c3..d24c1dcfaf 100644
--- a/ui/projects/streampipes/platform-services/src/lib/apis/adapter.service.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/apis/adapter.service.ts
@@ -82,8 +82,12 @@ export class AdapterService {
     }
 
     startAdapter(adapter: AdapterDescriptionUnion): Observable<Message> {
+        return this.startAdapterByElementId(adapter.elementId);
+    }
+
+    startAdapterByElementId(elementId: string): Observable<Message> {
         return this.http
-            .post(this.adapterMasterUrl + adapter.elementId + '/start', {})
+            .post(this.adapterMasterUrl + elementId + '/start', {})
             .pipe(map(response => Message.fromData(response as any)));
     }
 
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/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.html b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.html
index fe4fc3030e..c219c181fd 100644
--- a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.html
+++ b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.html
@@ -26,6 +26,7 @@
                 <div fxLayoutAlign="start start" class="mr-15">
                     <mat-checkbox
                         (change)="optionSelectedEmitter.emit($event.checked)"
+                        [checked]="isChecked"
                         [attr.data-cy]="dataCy"
                         class="large-checkbox"
                     >
diff --git a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.ts b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.ts
index e18e4e380a..c3599e0947 100644
--- a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.ts
+++ b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/adapter-options-panel/adapter-options-panel.component.ts
@@ -16,7 +16,7 @@
  *
  */
 
-import { Component, EventEmitter, Input, OnInit, Output } from '@angular/core';
+import { Component, EventEmitter, Input, Output } from '@angular/core';
 
 @Component({
     selector: 'sp-adapter-options-panel',
@@ -36,6 +36,9 @@ export class SpAdapterOptionsPanelComponent {
     @Input()
     dataCy: string;
 
+    @Input()
+    isChecked = false;
+
     @Output()
     optionSelectedEmitter: EventEmitter<boolean> = new EventEmitter<boolean>();
 }
diff --git a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.html b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.html
index 6c85414ebb..579285f219 100644
--- a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.html
+++ b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.html
@@ -47,6 +47,16 @@
     </sp-basic-inner-panel>
 
     <div fxFlex="100" fxLayout="column">
+        <sp-adapter-options-panel
+            optionTitle="Run adapter"
+            optionDescription="Start adapter now"
+            optionIcon="play_arrow"
+            dataCy="start-adapter-now-checkbox"
+            [isChecked]="startAdapterNow"
+            (optionSelectedEmitter)="startAdapterNow = $event"
+        >
+        </sp-adapter-options-panel>
+
         <sp-adapter-options-panel
             optionTitle="Remove Duplicates"
             optionDescription="Avoid duplicated events within a certain time interval"
@@ -150,7 +160,7 @@
             *ngIf="!isEditMode"
             [disabled]="!startAdapterSettingsFormValid"
             mat-raised-button
-            id="button-startAdapter"
+            data-cy="adapter-settings-start-adapter-btn"
             color="accent"
             (click)="startAdapter()"
             mat-button
diff --git a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.ts b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.ts
index fbff3bcb0c..8f9a6f46f6 100644
--- a/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.ts
+++ b/ui/src/app/connect/components/adapter-configuration/start-adapter-configuration/start-adapter-configuration.component.ts
@@ -88,6 +88,8 @@ export class StartAdapterConfigurationComponent implements OnInit {
     saveInDataLake = false;
     dataLakeTimestampField: string;
 
+    startAdapterNow = true;
+
     constructor(
         private dialogService: DialogService,
         private shepherdService: ShepherdService,
@@ -172,6 +174,7 @@ export class StartAdapterConfigurationComponent implements OnInit {
                 saveInDataLake: this.saveInDataLake,
                 dataLakeTimestampField: this.dataLakeTimestampField,
                 editMode: false,
+                startAdapterNow: this.startAdapterNow,
             },
         });
 
diff --git a/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.html b/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.html
index 4d969d9ab6..bccae12951 100644
--- a/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.html
+++ b/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.html
@@ -77,6 +77,7 @@
                         >
                             <i class="material-icons">done</i>
                             <span
+                                data-cy="sp-connect-adapter-not-started-success"
                                 >&nbsp;Your new data stream is now available in
                                 the pipeline editor.</span
                             >
@@ -98,7 +99,7 @@
                         </div>
                     </div>
 
-                    <div *ngIf="!isSetAdapter" fxFlex="100">
+                    <div *ngIf="!isSetAdapter && startAdapterNow" fxFlex="100">
                         <sp-pipeline-element-runtime-info
                             [streamDescription]="streamDescription"
                             [pollingActive]="pollingActive"
diff --git a/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.ts b/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.ts
index aaa4941cf5..9960fd0f45 100644
--- a/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.ts
+++ b/ui/src/app/connect/dialog/adapter-started/adapter-started-dialog.component.ts
@@ -26,6 +26,8 @@ import {
     PipelineOperationStatus,
     PipelineTemplateService,
     SpDataStream,
+    GenericAdapterStreamDescription,
+    SpecificAdapterStreamDescription,
     SpecificAdapterSetDescription,
 } from '@streampipes/platform-services';
 import { DialogRef } from '@streampipes/shared-ui';
@@ -67,6 +69,11 @@ export class AdapterStartedDialog implements OnInit {
      */
     @Input() editMode = false;
 
+    /**
+     * This option will immediately start the adapter, when false it the adapter is only created and not started
+     */
+    @Input() startAdapterNow = true;
+
     constructor(
         public dialogRef: DialogRef<AdapterStartedDialog>,
         private adapterService: AdapterService,
@@ -91,73 +98,100 @@ export class AdapterStartedDialog implements OnInit {
     }
 
     startAdapter() {
-        // const newAdapter = this.adapter;
         this.adapterService.addAdapter(this.adapter).subscribe(status => {
             this.adapterStatus = status;
+            const isStreamAdapter =
+                this.adapter instanceof GenericAdapterStreamDescription ||
+                this.adapter instanceof SpecificAdapterStreamDescription;
             if (status.success) {
-                // Start preview on streams and message for sets
-                if (
-                    this.adapter instanceof GenericAdapterSetDescription ||
-                    this.adapter instanceof SpecificAdapterSetDescription
-                ) {
-                    this.isSetAdapter = true;
+                const adapterElementId = status.notifications[0].title;
+                if (this.startAdapterNow && isStreamAdapter) {
+                    this.adapterService
+                        .startAdapterByElementId(adapterElementId)
+                        .subscribe(startStatus => {
+                            this.showAdapterPreview(
+                                startStatus,
+                                adapterElementId,
+                            );
+                        });
                 } else {
-                    this.getLiveViewPreview(status);
-                }
-
-                if (this.saveInDataLake) {
-                    this.startSaveInDataLakePipeline(status);
-                } else {
-                    this.adapterInstalled = true;
+                    this.showAdapterPreview(status, adapterElementId);
                 }
             }
         });
     }
 
+    showAdapterPreview(status: Message, adapterElementId: string) {
+        // Start preview on streams and message for sets
+        if (status.success) {
+            if (
+                this.adapter instanceof GenericAdapterSetDescription ||
+                this.adapter instanceof SpecificAdapterSetDescription
+            ) {
+                this.isSetAdapter = true;
+            } else {
+                this.getLiveViewPreview(adapterElementId);
+            }
+
+            if (this.saveInDataLake) {
+                this.startSaveInDataLakePipeline(adapterElementId);
+            } else {
+                this.adapterInstalled = true;
+            }
+        }
+    }
+
     onCloseConfirm() {
         this.pollingActive = false;
         this.dialogRef.close('Confirm');
         this.shepherdService.trigger('confirm_adapter_started_button');
     }
 
-    private getLiveViewPreview(status: Message) {
-        this.restService
-            .getSourceDetails(status.notifications[0].title)
-            .subscribe(st => {
-                this.streamDescription = st;
-                this.pollingActive = true;
-            });
+    private getLiveViewPreview(adapterElementId: string) {
+        this.adapterService.getAdapter(adapterElementId).subscribe(adapter => {
+            this.restService
+                .getSourceDetails(adapter.correspondingDataStreamElementId)
+                .subscribe(st => {
+                    this.streamDescription = st;
+                    this.pollingActive = true;
+                });
+        });
     }
 
-    private startSaveInDataLakePipeline(status: Message) {
-        const pipelineId =
-            'org.apache.streampipes.manager.template.instances.DataLakePipelineTemplate';
-        this.pipelineTemplateService
-            .getPipelineTemplateInvocation(
-                status.notifications[0].title,
-                pipelineId,
-            )
-            .subscribe(res => {
-                const pipelineName = 'Persist ' + this.adapter.name;
-
-                const indexName = this.adapter.name;
-
-                const pipelineInvocation = PipelineInvocationBuilder.create(res)
-                    .setName(pipelineName)
-                    .setTemplateId(pipelineId)
-                    .setFreeTextStaticProperty('db_measurement', indexName)
-                    .setMappingPropertyUnary(
-                        'timestamp_mapping',
-                        's0::' + this.dataLakeTimestampField,
+    private startSaveInDataLakePipeline(adapterElementId: string) {
+        this.adapterService.getAdapter(adapterElementId).subscribe(adapter => {
+            const pipelineId =
+                'org.apache.streampipes.manager.template.instances.DataLakePipelineTemplate';
+            this.pipelineTemplateService
+                .getPipelineTemplateInvocation(
+                    adapter.correspondingDataStreamElementId,
+                    pipelineId,
+                )
+                .subscribe(res => {
+                    const pipelineName = 'Persist ' + this.adapter.name;
+
+                    const indexName = this.adapter.name;
+
+                    const pipelineInvocation = PipelineInvocationBuilder.create(
+                        res,
                     )
-                    .build();
-
-                this.pipelineTemplateService
-                    .createPipelineTemplateInvocation(pipelineInvocation)
-                    .subscribe(pipelineOperationStatus => {
-                        this.pipelineOperationStatus = pipelineOperationStatus;
-                        this.adapterInstalled = true;
-                    });
-            });
+                        .setName(pipelineName)
+                        .setTemplateId(pipelineId)
+                        .setFreeTextStaticProperty('db_measurement', indexName)
+                        .setMappingPropertyUnary(
+                            'timestamp_mapping',
+                            's0::' + this.dataLakeTimestampField,
+                        )
+                        .build();
+
+                    this.pipelineTemplateService
+                        .createPipelineTemplateInvocation(pipelineInvocation)
+                        .subscribe(pipelineOperationStatus => {
+                            this.pipelineOperationStatus =
+                                pipelineOperationStatus;
+                            this.adapterInstalled = true;
+                        });
+                });
+        });
     }
 }
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;
+    }
 }
-


Test patch:
diff --git a/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java
new file mode 100644
index 0000000000..f525319ad9
--- /dev/null
+++ b/streampipes-extensions-management/src/test/java/org/apache/streampipes/extensions/management/connect/adapter/format/csv/CsvParserTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.format.csv;
+
+
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+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());
+  }
+
+}
\ 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/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/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/dataprovider/StreamProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachHttpRequest.java
similarity index 50%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/StreamProvider.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachHttpRequest.java
index 2a9fc50880..ac9b73705a 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/StreamProvider.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/DetachHttpRequest.java
@@ -16,11 +16,27 @@
  *
  */
 
-package org.apache.streampipes.performance.dataprovider;
+package org.apache.streampipes.manager.execution.http;
 
-import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.api.EndpointSelectable;
 
-public interface StreamProvider {
+import org.apache.http.client.fluent.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-  SpDataStream getStreamDescription();
+
+public class DetachHttpRequest extends PipelineElementHttpRequest {
+
+  private static final Logger LOG = LoggerFactory.getLogger(DetachHttpRequest.class);
+
+  @Override
+  protected Request initRequest(EndpointSelectable pipelineElement, String endpointUrl) {
+    LOG.info("Detaching element: " + endpointUrl);
+    return Request.Delete(endpointUrl);
+  }
+
+  @Override
+  protected void logError(String endpointUrl, String pipelineElementName, String exceptionMessage) {
+    LOG.error("Could not stop pipeline element {} at {}: {}", endpointUrl, pipelineElementName, exceptionMessage);
+  }
 }
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/PipelineElementProvider.java
similarity index 64%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/PipelineElementProvider.java
index 52a3ce6126..d8f7248017 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/dataprovider/DataProvider.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/provider/PipelineElementProvider.java
@@ -16,11 +16,17 @@
  *
  */
 
-package org.apache.streampipes.performance.dataprovider;
+package org.apache.streampipes.manager.execution.provider;
+
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 
 import java.util.List;
 
-public interface DataProvider<T> {
+public interface PipelineElementProvider {
+
+  List<InvocableStreamPipesEntity> getProcessorsAndSinks(PipelineExecutionInfo executionInfo);
 
-  List<T> getPreparedItems();
+  List<SpDataSet> getDataSets(PipelineExecutionInfo executionInfo);
 }
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/PipelineExecutionTask.java
similarity index 66%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/PipelineExecutionTask.java
index 93626cee3b..0964560610 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/PerformanceTest.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/PipelineExecutionTask.java
@@ -16,11 +16,17 @@
  *
  */
 
-package org.apache.streampipes.performance;
+package org.apache.streampipes.manager.execution.task;
 
-public class PerformanceTest {
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.pipeline.Pipeline;
 
-  public static void main(String[] args) {
+public interface PipelineExecutionTask {
 
+  default boolean shouldExecute(PipelineExecutionInfo executionInfo) {
+    return true;
   }
+
+  void executeTask(Pipeline pipeline,
+                   PipelineExecutionInfo executionInfo);
 }
diff --git a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SecretEncryptionTask.java
similarity index 55%
rename from streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java
rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SecretEncryptionTask.java
index bca098b63f..b530f93c8a 100644
--- a/streampipes-performance-tests/src/main/java/org/apache/streampipes/performance/util/ParameterTool.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/SecretEncryptionTask.java
@@ -15,22 +15,24 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.performance.util;
 
-import org.apache.streampipes.performance.model.PerformanceTestSettings;
+package org.apache.streampipes.manager.execution.task;
 
-public class ParameterTool {
+import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.resource.management.secret.SecretService;
 
-  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]);
-  }
+public class SecretEncryptionTask implements PipelineExecutionTask {
+
+  private final SecretService secretService;
 
-  private static Long toLong(String arg) {
-    return Long.parseLong(arg);
+  public SecretEncryptionTask(SecretService secretService) {
+    this.secretService = secretService;
   }
 
-  private static Integer toInt(String arg) {
-    return Integer.parseInt(arg);
+  @Override
+  public void executeTask(Pipeline pipeline,
+                          PipelineExecutionInfo executionInfo) {
+    this.secretService.apply(executionInfo.getProcessorsAndSinks());
   }
 }
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/EventPropertyGenerator.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/EventPropertyGenerator.java
rename to streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
index c3c4d33a76..8fd69330df 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/EventPropertyPrimitiveTestBuilder.java
@@ -17,6 +17,33 @@
  */
 package org.apache.streampipes.test.generator;
 
-public class EventPropertyGenerator {
+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/EventSchemaGenerator.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/EventSchemaGenerator.java
rename to streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyTestBuilder.java
index 90e96e5849..6bdb52aab1 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/EventPropertyTestBuilder.java
@@ -15,8 +15,26 @@
  * limitations under the License.
  *
  */
+
 package org.apache.streampipes.test.generator;
 
-public class EventSchemaGenerator {
+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;
+  }
+
 }
diff --git a/ui/cypress/tests/adapter/createAdapterWithoutStarting.spec.ts b/ui/cypress/tests/adapter/createAdapterWithoutStarting.spec.ts
new file mode 100644
index 0000000000..12ba2cced9
--- /dev/null
+++ b/ui/cypress/tests/adapter/createAdapterWithoutStarting.spec.ts
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ *
+ */
+
+import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
+import { SpecificAdapterBuilder } from '../../support/builder/SpecificAdapterBuilder';
+
+describe('Creates a new adapter without starting it', () => {
+    beforeEach('Setup Test', () => {
+        cy.initStreamPipesTest();
+    });
+
+    it('Perform Test', () => {
+        const adapterInput = SpecificAdapterBuilder.create(
+            'Machine_Data_Simulator',
+        )
+            .setName('Machine Data Simulator Test')
+            .addInput('input', 'wait-time-ms', '1000')
+            .build();
+
+        ConnectUtils.testSpecificStreamAdapter(
+            adapterInput,
+            false,
+            'sp-connect-adapter-not-started-success',
+        );
+
+        ConnectUtils.startAndValidateAdapter(7);
+
+        ConnectUtils.closeAdapterPreview();
+
+        ConnectUtils.deleteAdapter();
+    });
+});
diff --git a/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts b/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
index 336464f762..a6db9b08b8 100644
--- a/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
+++ b/ui/cypress/tests/adapter/editAdapter.smoke.spec.ts
@@ -68,23 +68,7 @@ describe('Test Edit Adapter', () => {
 
         ConnectUtils.closeAdapterPreview();
 
-        // Start Adapter
-        ConnectBtns.startAdapter().should('not.be.disabled');
-        ConnectBtns.startAdapter().click();
-
-        // View data
-        ConnectBtns.infoAdapter().click();
-        cy.get('div').contains('Values').parent().click();
-
-        // Validate resulting event
-        cy.dataCy('sp-connect-adapter-live-preview', { timeout: 10000 }).should(
-            'be.visible',
-        );
-
-        // validate that three event properties
-        cy.get('.preview-row', { timeout: 10000 })
-            .its('length')
-            .should('eq', 3);
+        ConnectUtils.startAndValidateAdapter(3);
 
         // Validate that name of adapter and data source
         cy.dataCy('adapter-name').contains(newAdapterName);