Repository: apache/streampipes
Pull request #3312 (2024-10-29 09:16:06)

Support output strategies in pipeline code
<!--
  ~ 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/community/get-involved/
    - 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.
-->

Support custom and user-defined output strategies in the pipeline-as-code model.

Example: 

```
id: test
name: Test
description: ""
pipelineElements:
  - type: stream
    ref: 0_6fSh
    id: sp:spdatastream:Kmneup
  - type: processor
    ref: 1_ScpA
    id: org.apache.streampipes.processors.filters.jvm.project
    connectedTo:
      - 0_6fSh
    configuration: []
    output:
      keep:
        - s0::timestamp
        - s0::volume_flow
  - type: processor
    ref: 2_QGAv
    id: org.apache.streampipes.processors.enricher.jvm.jseval
    connectedTo:
      - 1_ScpA
    configuration:
      - jsFunction: >-
          function process(event) {
              // do processing here.
              // return processed event.
             // Type 'event' and press Ctrl+Space to see available fields.
             // Example: 
              return {timestamp: event.timestamp, tempInCelsius: (event.tempInKelvin - 273.15)};
          }
    output:
      userDefined:
        - fieldName: timestamp
          runtimeType: http://www.w3.org/2001/XMLSchema#long
          semanticType: http://schema.org/DateTime
        - fieldName: volume_flow
          runtimeType: http://www.w3.org/2001/XMLSchema#float
          semanticType: http://schema.org/Number
        - fieldName: volume_flow2
          runtimeType: http://www.w3.org/2001/XMLSchema#float
          semanticType: http://schema.org/Number  
  - type: sink
    ref: 3_9Wl9
    id: org.apache.streampipes.sinks.internal.jvm.datalake
    connectedTo:
      - 2_QGAv
    configuration:
      - timestamp_mapping: s0::timestamp
      - db_measurement: asddd2
      - schema_update: Update schema
      - dimensions_selection: []
      - ignore_duplicates: false
createOptions:
  start: false
```
### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/streampipes-client-go/docs/content/en/_index.md b/streampipes-client-go/docs/content/en/_index.md
index f852feaeb6..2d70e278f3 100644
--- a/streampipes-client-go/docs/content/en/_index.md
+++ b/streampipes-client-go/docs/content/en/_index.md
@@ -2,21 +2,21 @@
 title: Apache StreamPipesocs
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{< blocks/cover title="Welcome to StreamPipes Go" image_anchor="top" height="full" >}}
diff --git a/streampipes-client-go/docs/content/en/docs/_index.md b/streampipes-client-go/docs/content/en/docs/_index.md
index 2f6e16f46c..7952ce6127 100644
--- a/streampipes-client-go/docs/content/en/docs/_index.md
+++ b/streampipes-client-go/docs/content/en/docs/_index.md
@@ -4,21 +4,21 @@ linkTitle: Docs
 menu: {main: {weight: 20}}
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/contribution-guidelines.md b/streampipes-client-go/docs/content/en/docs/contribution-guidelines.md
index c9f0ca0326..a5836d8ebe 100644
--- a/streampipes-client-go/docs/content/en/docs/contribution-guidelines.md
+++ b/streampipes-client-go/docs/content/en/docs/contribution-guidelines.md
@@ -4,21 +4,21 @@ description: How to contribute Apache StreamPipes.
 weight: 10
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/getting-started/_index.md b/streampipes-client-go/docs/content/en/docs/getting-started/_index.md
index 1eba1d701e..c930741806 100644
--- a/streampipes-client-go/docs/content/en/docs/getting-started/_index.md
+++ b/streampipes-client-go/docs/content/en/docs/getting-started/_index.md
@@ -4,21 +4,21 @@ description: Getting Started to use StreamPipes Go.
 weight: 2
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/getting-started/first-steps.md b/streampipes-client-go/docs/content/en/docs/getting-started/first-steps.md
index 8b2135cdfb..2d28e4391f 100644
--- a/streampipes-client-go/docs/content/en/docs/getting-started/first-steps.md
+++ b/streampipes-client-go/docs/content/en/docs/getting-started/first-steps.md
@@ -5,21 +5,21 @@ description: >
   First Steps to use StreamPipes Go.
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
@@ -47,3 +47,14 @@ Access to a running instance of Apache StreamPipes is required.
 If you do not have it set up yet, please refer to the [Apache StreamPipes installation guide](https://streampipes.apache.org/docs/try-installation/) for detailed instructions on how to install and configure it on your system.
 
 Once these requirements are met, you are ready to proceed with the next step Quickstart in the setup process!
+
+## 3. Installation StreamPipes Go library
+
+The StreamPipes Go library is meant to work with Go 1.21 and above. You can install the latest development version from GitHub, as so:
+
+```bash
+go get https://github.com/apache/streampipes
+// if you want to have the current development state you can also execute
+go get https://github.com/apache/streampipes/streampipes-client-go
+// the corresponding documentation can be found here: https://streampipes.apache.org/docs/docs/python/dev/
+```
\ No newline at end of file
diff --git a/streampipes-client-go/docs/content/en/docs/getting-started/quickstart.md b/streampipes-client-go/docs/content/en/docs/getting-started/quickstart.md
index e2498e20f9..0be36623a3 100644
--- a/streampipes-client-go/docs/content/en/docs/getting-started/quickstart.md
+++ b/streampipes-client-go/docs/content/en/docs/getting-started/quickstart.md
@@ -4,21 +4,21 @@ description: >
   Quickstart to use StreamPipes Go.
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/overview.md b/streampipes-client-go/docs/content/en/docs/overview.md
index 8d31e43e33..41215a9edf 100644
--- a/streampipes-client-go/docs/content/en/docs/overview.md
+++ b/streampipes-client-go/docs/content/en/docs/overview.md
@@ -4,21 +4,21 @@ description: Apache StreamPipesocs Go Client Overview.
 weight: 1
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/reference/_index.md b/streampipes-client-go/docs/content/en/docs/reference/_index.md
index 97f44a4b1f..99b48048b3 100644
--- a/streampipes-client-go/docs/content/en/docs/reference/_index.md
+++ b/streampipes-client-go/docs/content/en/docs/reference/_index.md
@@ -4,21 +4,21 @@ description: Reference for StreamPipes GO.
 weight: 9
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/reference/parameter-reference.md b/streampipes-client-go/docs/content/en/docs/reference/parameter-reference.md
index 7f10d3847f..dd1e697077 100644
--- a/streampipes-client-go/docs/content/en/docs/reference/parameter-reference.md
+++ b/streampipes-client-go/docs/content/en/docs/reference/parameter-reference.md
@@ -5,21 +5,21 @@ description: >
 date: 2024-06-26
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/docs/tutorials/Introduction-to-streampipes-go.md b/streampipes-client-go/docs/content/en/docs/tutorials/Introduction-to-streampipes-go.md
index d5530b56bc..f2bf7f0437 100644
--- a/streampipes-client-go/docs/content/en/docs/tutorials/Introduction-to-streampipes-go.md
+++ b/streampipes-client-go/docs/content/en/docs/tutorials/Introduction-to-streampipes-go.md
@@ -6,21 +6,21 @@ date: 2024-06-26
 weight: 4
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
@@ -28,3 +28,5 @@ Introduction to StreamPipes Go.
 {{% /pageinfo %}}
 
 Apache StreamPipes aims to enable non-technical users to connect and analyze IoT data streams. To this end, it provides an easy-to-use and convenient user interface that allows one to connect to an IoT data source and create some visual graphs within a few minutes.
+
+--- StreamPipes Go-Client
\ No newline at end of file
diff --git a/streampipes-client-go/docs/content/en/docs/tutorials/ManageAdapter.md b/streampipes-client-go/docs/content/en/docs/tutorials/ManageAdapter.md
new file mode 100644
index 0000000000..059700c2c6
--- /dev/null
+++ b/streampipes-client-go/docs/content/en/docs/tutorials/ManageAdapter.md
@@ -0,0 +1,72 @@
+<!--
+  // 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.
+  //
+  -->
+
+###  Manage StreamPipes Adapter
+---
+## Initialize Go-Client
+```bash
+import (
+	"log"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+)
+func main() {
+      clientConfig := config.StreamPipesClientConfig{
+            Url: "http://localhost:8030",
+            Credential: config.StreamPipesApiKeyCredentials{
+                UserName: "<Your-User-Name>",
+                ApiKey:   "<Your-API-Key>",
+            },
+        }
+    
+      streamPipesClient, err := streampipes.NewStreamPipesClient(clientConfig)
+      if err != nil {
+          log.Fatal(err)
+      }
+
+}
+```
+
+### Create a StreamPipes Adapter
+
+```bash
+      // Populating go-model as a data source is not supported at this time,This is because the inheritance of the model has not been resolved for the time being.
+      adapterData = `json`
+      adapter = []byte(adapterData) 
+	  err := streamPipesClient.Adapter().CreateAdapter(adapterData) 
+```
+
+### Get a StreamPipes Adapter
+
+```bash
+	streampipesAdapter, err := streamPipesClient.Adapter().GetSingleAdapter("adapterId")
+```
+
+### Start a StreamPipes Adapter
+
+```bash
+	err := streamPipesClient.Adapter().StartSingleAdapter("adapterId")
+```
+
+### Stop a StreamPipes Adapter
+
+```bash
+	err := streamPipesClient.Adapter().StopSingleAdapter("adapterId")
+```
+
+### ... 
\ No newline at end of file
diff --git a/streampipes-client-go/docs/content/en/docs/tutorials/ManagePipeline.md b/streampipes-client-go/docs/content/en/docs/tutorials/ManagePipeline.md
new file mode 100644
index 0000000000..39bf323ecd
--- /dev/null
+++ b/streampipes-client-go/docs/content/en/docs/tutorials/ManagePipeline.md
@@ -0,0 +1,72 @@
+<!--
+  // 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.
+  //
+  -->
+
+###  Manage StreamPipes Pipeline
+---
+## Initialize Go-Client
+```bash
+import (
+	"log"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+)
+func main() {
+      clientConfig := config.StreamPipesClientConfig{
+            Url: "http://localhost:8030",
+            Credential: config.StreamPipesApiKeyCredentials{
+                UserName: "<Your-User-Name>",
+                ApiKey:   "<Your-API-Key>",
+            },
+        }
+    
+      streamPipesClient, err := streampipes.NewStreamPipesClient(clientConfig)
+      if err != nil {
+          log.Fatal(err)
+      }
+
+}
+```
+
+### Create a StreamPipes Pipeline
+
+```bash
+      // Populating go-model as a data source is not supported at this time,This is because the inheritance of the model has not been resolved for the time being.
+      pipelineData = `json`
+      pipeline = []byte(pipelineData) 
+	  responseMsg,err := streamPipesClient.Pipeline().CreatePipeline(pipelineData)
+```
+
+### Get a StreamPipes Pipeline
+
+```bash
+	streampipesPipeline, err := streamPipesClient.Pipeline().GetSinglePipelineStatus("pipelineId")
+```
+
+### Start a StreamPipes Pipeline
+
+```bash
+	operationStatus, err := streamPipesClient.Pipeline().StartSinglePipeline("pipelineId")
+```
+
+### Stop a StreamPipes Pipeline
+
+```bash
+	operationStatus, err := streamPipesClient.Pipeline().StopSinglePipeline("pipelineId")
+```
+
+### ... 
\ No newline at end of file
diff --git a/streampipes-client-go/docs/content/en/docs/tutorials/_index.md b/streampipes-client-go/docs/content/en/docs/tutorials/_index.md
index 9714093a1e..8a3674ee2c 100644
--- a/streampipes-client-go/docs/content/en/docs/tutorials/_index.md
+++ b/streampipes-client-go/docs/content/en/docs/tutorials/_index.md
@@ -5,21 +5,21 @@ date: 2024-06-26
 weight: 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.
-  ~
+  // 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.
+  //
   -->
 
 {{% pageinfo %}}
diff --git a/streampipes-client-go/docs/content/en/search.md b/streampipes-client-go/docs/content/en/search.md
index acd8ec8396..802db9006c 100644
--- a/streampipes-client-go/docs/content/en/search.md
+++ b/streampipes-client-go/docs/content/en/search.md
@@ -3,19 +3,19 @@ title: Search Results
 layout: search
 ---
 <!--
-  ~ 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.
-  ~
+  // 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.
+  //
   -->
diff --git a/streampipes-client-go/streampipes/adapter_api.go b/streampipes-client-go/streampipes/adapter_api.go
index 06fb42d789..d2f7eb7637 100644
--- a/streampipes-client-go/streampipes/adapter_api.go
+++ b/streampipes-client-go/streampipes/adapter_api.go
@@ -120,13 +120,9 @@ func (a *Adapter) DeleteSingleAdapter(adapterId string) error {
 	return nil
 }
 
-func (a *Adapter) CreateAdapter(adapters adapter.AdapterDescription) error {
+func (a *Adapter) CreateAdapter(adapters []byte) error {
 	endPointUrl := util.NewStreamPipesApiPath(a.config.Url, "streampipes-backend/api/v2/connect/master/adapters", nil)
-	body, err := serializer.NewAdapterSerializer().Marshal(adapters)
-	if err != nil {
-		return err
-	}
-	response, err := a.executeRequest("PUT", endPointUrl, body)
+	response, err := a.executeRequest("POST", endPointUrl, adapters)
 	if err != nil {
 		return err
 	}
@@ -158,8 +154,7 @@ func (a *Adapter) StopSingleAdapter(adapterId string) error {
 }
 
 func (a *Adapter) StartSingleAdapter(adapterId string) error {
-	endPointUrl := util.NewStreamPipesApiPath(a.config.Url, "streampipes-backend/api/v2/pipelines", []string{adapterId, "start"})
-
+	endPointUrl := util.NewStreamPipesApiPath(a.config.Url, "streampipes-backend/api/v2/connect/master/adapters", []string{adapterId, "start"})
 	response, err := a.executeRequest("POST", endPointUrl, nil)
 	if err != nil {
 		return err
diff --git a/streampipes-client-go/streampipes/internal/serializer/deserializer.go b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
index 9c47fd3383..27283c85b4 100644
--- a/streampipes-client-go/streampipes/internal/serializer/deserializer.go
+++ b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
@@ -20,6 +20,8 @@ package serializer
 import (
 	"encoding/json"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/adapter"
+	"log"
+	"strings"
 
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/data_lake"
@@ -45,8 +47,10 @@ func NewDataLakeMeasuresDeserializer() *DataLakeMeasuresDeserializer {
 
 func (d DataLakeMeasuresDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var dataLakeMeasures []data_lake.DataLakeMeasure
-	err := json.Unmarshal(data, &dataLakeMeasures)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&dataLakeMeasures); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return dataLakeMeasures, nil
@@ -60,8 +64,10 @@ func NewDataLakeMeasureDeserializer() *DataLakeMeasureDeserializer {
 
 func (d DataLakeMeasureDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var dataLakeMeasure data_lake.DataLakeMeasure
-	err := json.Unmarshal(data, &dataLakeMeasure)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&dataLakeMeasure); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return dataLakeMeasure, nil
@@ -75,8 +81,10 @@ func NewDataSeriesDeserializer() *DataSeriesDeserializer {
 
 func (d DataSeriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var dataSeries data_lake.DataSeries
-	err := json.Unmarshal(data, &dataSeries)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&dataSeries); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return dataSeries, nil
@@ -89,12 +97,14 @@ func NewStreamPipesVersionDeserializer() *StreamPipesVersionDeserializer {
 }
 
 func (d StreamPipesVersionDeserializer) Unmarshal(data []byte) (interface{}, error) {
-	var dataSeries streampipes_version.Versions
-	err := json.Unmarshal(data, &dataSeries)
-	if err != nil {
+	var version streampipes_version.Versions
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&version); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
-	return dataSeries, nil
+	return version, nil
 }
 
 type ResponseMessageDeserializer struct{}
@@ -105,8 +115,10 @@ func NewResponseMessageDeserializer() *ResponseMessageDeserializer {
 
 func (r ResponseMessageDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var responseMessage model.ResponseMessage
-	err := json.Unmarshal(data, &responseMessage)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&responseMessage); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return responseMessage, nil
@@ -120,10 +132,15 @@ func NewPipelineCategoriesDeserializer() *PipelineCategoriesDeserializer {
 
 func (p PipelineCategoriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var pipelineCategory []pipeline.PipelineCategory
-	err := json.Unmarshal(data, &pipelineCategory)
-	if err != nil {
+
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+
+	if err := dec.Decode(&pipelineCategory); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Fatal(err)
 		return nil, err
 	}
+
 	return pipelineCategory, nil
 }
 
@@ -135,8 +152,10 @@ func NewDataLakeDashboardDeserializer() *DataLakeDashboardDeserializer {
 
 func (d DataLakeDashboardDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var dashborad data_lake.Dashboard
-	err := json.Unmarshal(data, &dashborad)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&dashborad); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return dashborad, nil
@@ -149,13 +168,15 @@ func NewDataLakeDashboardsDeserializer() *DataLakeDashboardsDeserializer {
 }
 
 func (d DataLakeDashboardsDeserializer) Unmarshal(data []byte) (interface{}, error) {
-	var dashborad []data_lake.Dashboard
-	err := json.Unmarshal(data, &dashborad)
-	if err != nil {
+	var dashborads []data_lake.Dashboard
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&dashborads); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 
-	return dashborad, nil
+	return dashborads, nil
 }
 
 type DataLakeWidgetDeserializer struct{}
@@ -166,8 +187,10 @@ func NewDataLakeWidgetDeserializer() *DataLakeWidgetDeserializer {
 
 func (d DataLakeWidgetDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var widget data_lake.DataExplorerWidgetModel
-	err := json.Unmarshal(data, &widget)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&widget); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return widget, nil
@@ -180,12 +203,14 @@ func NewDataLakeWidgetsDeserializer() *DataLakeWidgetsDeserializer {
 }
 
 func (d DataLakeWidgetsDeserializer) Unmarshal(data []byte) (interface{}, error) {
-	var widget []data_lake.DataExplorerWidgetModel
-	err := json.Unmarshal(data, &widget)
-	if err != nil {
+	var widgets []data_lake.DataExplorerWidgetModel
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&widgets); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
-	return widget, nil
+	return widgets, nil
 }
 
 type SpLogEntriesDeserializer struct{}
@@ -194,13 +219,15 @@ func NewSpLogEntriesDeserializer() *SpLogEntriesDeserializer {
 	return &SpLogEntriesDeserializer{}
 }
 
-func (p SpLogEntriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
-	var userAccount []functions.SpLogEntry
-	err := json.Unmarshal(data, &userAccount)
-	if err != nil {
+func (s SpLogEntriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var spLogEntry []functions.SpLogEntry
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&spLogEntry); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
-	return userAccount, nil
+	return spLogEntry, nil
 }
 
 type SpMetricsEntryDeserializer struct{}
@@ -209,10 +236,12 @@ func NewSpMetricsEntryDeserializer() *SpMetricsEntryDeserializer {
 	return &SpMetricsEntryDeserializer{}
 }
 
-func (p SpMetricsEntryDeserializer) Unmarshal(data []byte) (interface{}, error) {
+func (s SpMetricsEntryDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var spMetricsEntry functions.SpMetricsEntry
-	err := json.Unmarshal(data, &spMetricsEntry)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&spMetricsEntry); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return spMetricsEntry, nil
@@ -224,10 +253,12 @@ func NewFunctionDefinitionsDeserializer() *FunctionDefinitionsDeserializer {
 	return &FunctionDefinitionsDeserializer{}
 }
 
-func (p FunctionDefinitionsDeserializer) Unmarshal(data []byte) (interface{}, error) {
+func (f FunctionDefinitionsDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var functionDefinitions []functions.FunctionDefinition
-	err := json.Unmarshal(data, &functionDefinitions)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&functionDefinitions); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return functionDefinitions, nil
@@ -242,8 +273,10 @@ func NewShortUserInfosDeserializer() *ShortUserInfosDeserializer {
 
 func (s ShortUserInfosDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var shortUserInfo []streampipes_user.ShortUserInfo
-	err := json.Unmarshal(data, &shortUserInfo)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&shortUserInfo); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return shortUserInfo, nil
@@ -257,8 +290,10 @@ func NewUserAccountDeserializer() *UserAccountDeserializer {
 
 func (p UserAccountDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var userAccount streampipes_user.UserAccount
-	err := json.Unmarshal(data, &userAccount)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&userAccount); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return userAccount, nil
@@ -273,10 +308,13 @@ func NewPipelineDeserializer() *PipelineDeserializer {
 
 func (p PipelineDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var pipeLine pipeline.Pipeline
-	err := json.Unmarshal(data, &pipeLine)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&pipeLine); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
+
 	return pipeLine, nil
 
 }
@@ -289,8 +327,11 @@ func NewPipelinesDeserializer() *PipelinesDeserializer {
 
 func (p PipelinesDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var pipelines []pipeline.Pipeline
-	err := json.Unmarshal(data, &pipelines)
-	if err != nil {
+
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&pipelines); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return pipelines, nil
@@ -305,8 +346,10 @@ func NewPipelineStatusMessagesDeserializer() *PipelineStatusMessagesDeserializer
 
 func (p PipelineStatusMessagesDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var pipelineStatusMessage []pipeline.PipelineStatusMessage
-	err := json.Unmarshal(data, &pipelineStatusMessage)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&pipelineStatusMessage); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return pipelineStatusMessage, nil
@@ -320,8 +363,10 @@ func NewPipelineOperationStatusDeserializer() *PipelineOperationStatusDeserializ
 
 func (p PipelineOperationStatusDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var pipelineOperationStatus pipeline.PipelineOperationStatus
-	err := json.Unmarshal(data, &pipelineOperationStatus)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&pipelineOperationStatus); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Println(err)
 		return nil, err
 	}
 	return pipelineOperationStatus, nil
@@ -335,8 +380,11 @@ func NewAdapterDeserializer() *AdapterDeserializer {
 
 func (a AdapterDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var adapterDescription adapter.AdapterDescription
-	err := json.Unmarshal(data, &adapterDescription)
-	if err != nil {
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+
+	if err := dec.Decode(&adapterDescription); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Fatal(err)
 		return nil, err
 	}
 	return adapterDescription, nil
@@ -351,8 +399,12 @@ func NewAdaptersDeserializer() *AdaptersDeserializer {
 
 func (a AdaptersDeserializer) Unmarshal(data []byte) (interface{}, error) {
 	var adapters []adapter.AdapterDescription
-	err := json.Unmarshal(data, &adapters)
-	if err != nil {
+
+	dec := json.NewDecoder(strings.NewReader(string(data)))
+	dec.DisallowUnknownFields()
+
+	if err := dec.Decode(&adapters); err != nil && !strings.Contains(err.Error(), "unknown field") {
+		log.Fatal(err)
 		return nil, err
 	}
 	return adapters, nil
diff --git a/streampipes-client-go/streampipes/model/adapter/adapter_description.go b/streampipes-client-go/streampipes/model/adapter/adapter_description.go
index 4b856d32fa..7180442721 100644
--- a/streampipes-client-go/streampipes/model/adapter/adapter_description.go
+++ b/streampipes-client-go/streampipes/model/adapter/adapter_description.go
@@ -20,6 +20,7 @@ package adapter
 import "github.com/apache/streampipes/streampipes-client-go/streampipes/model"
 
 type AdapterDescription struct {
+	Class                            string                                 `json:"@class"`
 	ElementID                        string                                 `json:"elementId"`
 	Rev                              string                                 `json:"rev"`
 	DOM                              string                                 `json:"dom"`
diff --git a/streampipes-client-go/streampipes/model/common.go b/streampipes-client-go/streampipes/model/common.go
index d5a81d684b..b353308556 100644
--- a/streampipes-client-go/streampipes/model/common.go
+++ b/streampipes-client-go/streampipes/model/common.go
@@ -86,6 +86,51 @@ type StaticProperty struct {
 	Class              string `json:"@class"`
 }
 
+type MappingProperty struct {
+	StaticProperty
+	RequirementSelector string
+	MapsFromOptions     []string
+	PropertyScope       string
+}
+
+type MappingPropertyUnary struct {
+	MappingProperty
+	SelectedProperty string
+}
+
+type FreeTextStaticProperty struct {
+	StaticProperty
+	value                 string
+	MapsTo                string
+	MultiLine             bool
+	HtmlAllowed           bool
+	HtmlFontFormat        bool
+	PlaceholdersSupported bool
+}
+
+type SelectionStaticProperty struct {
+	HorizontalRendering bool
+}
+
+type OneOfStaticProperty struct {
+	SelectionStaticProperty
+}
+
+type AnyStaticProperty struct {
+	SelectionStaticProperty
+}
+
+type RuntimeResolvableAnyStaticProperty struct {
+	AnyStaticProperty
+	DependsOn []string
+}
+
+type SlideToggleStaticProperty struct {
+	StaticProperty
+	Selected     bool
+	DefaultValue bool
+}
+
 type SpDataStream struct {
 	ElementId              string         `json:"elementId"`
 	Dom                    string         `json:"dom"`
diff --git a/streampipes-client-go/streampipes/model/pipeline/pipeline.go b/streampipes-client-go/streampipes/model/pipeline/pipeline.go
index 0586be659d..65273cd227 100644
--- a/streampipes-client-go/streampipes/model/pipeline/pipeline.go
+++ b/streampipes-client-go/streampipes/model/pipeline/pipeline.go
@@ -30,23 +30,23 @@ const (
 )
 
 type Pipeline struct {
-	Sepas                 []DataProcessorInvocation //`json:"sepas"`
-	Streams               []model.SpDataStream      //`json:"streams"`
-	Name                  string                    //`json:"name"`
-	Description           string                    //`json:"description,omitempty"`
-	Actions               []DataSinkInvocation      //`json:"actions"`
-	Running               bool                      //`json:"running"`
-	RestartOnSystemReboot bool                      //`json:"restartOnSystemReboot"`
-	Valid                 bool                      //`json:"valid"`
-	StartedAt             int64                     //`json:"startedAt,omitempty"`
-	CreatedAt             int64                     //`json:"createdAt"`
-	PublicElement         bool                      //`json:"publicElement"`
-	CreatedByUser         string                    //`json:"createdByUser"`
-	PipelineCategories    []string                  //`json:"pipelineCategories"`
-	PipelineNotifications []string                  //`json:"pipelineNotifications"`
-	HealthStatus          PipelineHealthStatus      //`json:"healthStatus"`
-	ID                    string                    //`json:"_id,omitempty"`
-	Rev                   string                    //`json:"_rev,omitempty"`
+	Sepas                 []DataProcessorInvocation `json:"sepas"`
+	Streams               []model.SpDataStream      `json:"streams"`
+	Name                  string                    `json:"name"`
+	Description           string                    `json:"description,omitempty"`
+	Actions               []DataSinkInvocation      `json:"actions"`
+	Running               bool                      `json:"running"`
+	RestartOnSystemReboot bool                      `json:"restartOnSystemReboot"`
+	Valid                 bool                      `json:"valid"`
+	StartedAt             int64                     `json:"startedAt,omitempty"`
+	CreatedAt             int64                     `json:"createdAt"`
+	PublicElement         bool                      `json:"publicElement"`
+	CreatedByUser         string                    `json:"createdByUser"`
+	PipelineCategories    []string                  `json:"pipelineCategories"`
+	PipelineNotifications []string                  `json:"pipelineNotifications"`
+	HealthStatus          PipelineHealthStatus      `json:"healthStatus"`
+	PipelineId            string                    `json:"_id"`
+	Rev                   string                    `json:"_rev"`
 }
 
 type DataProcessorInvocation struct {
diff --git a/streampipes-client-go/streampipes/pipeline_api.go b/streampipes-client-go/streampipes/pipeline_api.go
index e58ccc7096..f507134685 100644
--- a/streampipes-client-go/streampipes/pipeline_api.go
+++ b/streampipes-client-go/streampipes/pipeline_api.go
@@ -18,15 +18,14 @@
 package streampipes
 
 import (
-	"fmt"
-	"io"
-	"log"
-	"net/http"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/serializer"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/util"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/pipeline"
+	"io"
+	"log"
+	"net/http"
 )
 
 type Pipeline struct {
@@ -156,13 +155,9 @@ func (p *Pipeline) GetAllPipeline() ([]pipeline.Pipeline, error) {
 }
 
 // CreatePipeline store a new pipeline
-func (p *Pipeline) CreatePipeline(pp pipeline.Pipeline) (model.ResponseMessage, error) {
+func (p *Pipeline) CreatePipeline(body []byte) (model.ResponseMessage, error) {
 	endPointUrl := util.NewStreamPipesApiPath(p.config.Url, "streampipes-backend/api/v2/pipelines", nil)
 
-	body, err := serializer.NewPipelineSerializer().Marshal(pp)
-	if err != nil {
-		return model.ResponseMessage{}, err
-	}
 	response, err := p.executeRequest("POST", endPointUrl, body)
 	if err != nil {
 		return model.ResponseMessage{}, err
@@ -180,7 +175,6 @@ func (p *Pipeline) CreatePipeline(pp pipeline.Pipeline) (model.ResponseMessage,
 
 	unmarshalData, err := serializer.NewResponseMessageDeserializer().Unmarshal(data)
 	if err != nil {
-		fmt.Println(err, 11)
 		return model.ResponseMessage{}, err
 	}
 	message := unmarshalData.(model.ResponseMessage)
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/compact/PersistPipelineHandler.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/compact/PersistPipelineHandler.java
index dd0afb7ef7..8b53e019f6 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/compact/PersistPipelineHandler.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/compact/PersistPipelineHandler.java
@@ -94,6 +94,7 @@ public class PersistPipelineHandler {
         DATA_LAKE_CONNECTOR_ID,
         adapterDescription.getCorrespondingDataStreamElementId(),
         null,
+        null,
         null
     ));
     return pipelineElements;
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 62c514b1e6..5c42478431 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
@@ -53,10 +53,11 @@ public class AdapterMasterManagement {
 
   private final DataStreamResourceManager dataStreamResourceManager;
 
-  public AdapterMasterManagement(IAdapterStorage adapterInstanceStorage,
-                                 AdapterResourceManager adapterResourceManager,
-                                 DataStreamResourceManager dataStreamResourceManager,
-                                 AdapterMetrics adapterMetrics
+  public AdapterMasterManagement(
+      IAdapterStorage adapterInstanceStorage,
+      AdapterResourceManager adapterResourceManager,
+      DataStreamResourceManager dataStreamResourceManager,
+      AdapterMetrics adapterMetrics
   ) {
     this.adapterInstanceStorage = adapterInstanceStorage;
     this.adapterMetrics = adapterMetrics;
@@ -64,28 +65,40 @@ public class AdapterMasterManagement {
     this.dataStreamResourceManager = dataStreamResourceManager;
   }
 
-  public void addAdapter(AdapterDescription ad,
-                           String adapterElementId,
-                           String principalSid)
+  public void addAdapter(
+      AdapterDescription adapterDescription,
+      String adapterId,
+      String principalSid
+  )
       throws AdapterException {
 
-    // Create elementId for adapter
+    // Create elementId for datastream
     var dataStreamElementId = ElementIdGenerator.makeElementId(SpDataStream.class);
-    ad.setElementId(adapterElementId);
-    ad.setCreatedAt(System.currentTimeMillis());
-    ad.setCorrespondingDataStreamElementId(dataStreamElementId);
+    adapterDescription.setElementId(adapterId);
+    adapterDescription.setCreatedAt(System.currentTimeMillis());
+    adapterDescription.setCorrespondingDataStreamElementId(dataStreamElementId);
 
     // Add EventGrounding to AdapterDescription
     var eventGrounding = GroundingUtils.createEventGrounding();
-    ad.setEventGrounding(eventGrounding);
+    adapterDescription.setEventGrounding(eventGrounding);
 
-    this.adapterResourceManager.encryptAndCreate(ad);
+    this.adapterResourceManager.encryptAndCreate(adapterDescription);
 
-    // Create stream
-    var storedDescription = new SourcesManagement().createAdapterDataStream(ad, dataStreamElementId);
-    storedDescription.setCorrespondingAdapterId(adapterElementId);
+    // Stream is only created if the adpater is successfully stored
+    createDataStreamForAdapter(adapterDescription, adapterId, dataStreamElementId, principalSid);
+  }
+
+  private void createDataStreamForAdapter(
+      AdapterDescription adapterDescription,
+      String adapterId,
+      String streamId,
+      String principalSid
+  ) throws AdapterException {
+    var storedDescription = new SourcesManagement()
+        .createAdapterDataStream(adapterDescription, streamId);
+    storedDescription.setCorrespondingAdapterId(adapterId);
     installDataSource(storedDescription, principalSid, true);
-    LOG.info("Install source (source URL: {} in backend", ad.getElementId());
+    LOG.info("Install source (source URL: {} in backend", adapterDescription.getElementId());
   }
 
   public AdapterDescription getAdapter(String elementId) throws AdapterException {
@@ -163,7 +176,8 @@ public class AdapterMasterManagement {
       var baseUrl = new ExtensionsServiceEndpointGenerator().getEndpointBaseUrl(
           ad.getAppId(),
           SpServiceUrlProvider.ADAPTER,
-          ad.getDeploymentConfiguration().getDesiredServiceTags()
+          ad.getDeploymentConfiguration()
+            .getDesiredServiceTags()
       );
 
       // Update selected endpoint URL of adapter
@@ -182,9 +196,11 @@ public class AdapterMasterManagement {
     }
   }
 
-  private void installDataSource(SpDataStream stream,
-                                 String principalSid,
-                                 boolean publicElement) throws AdapterException {
+  private void installDataSource(
+      SpDataStream stream,
+      String principalSid,
+      boolean publicElement
+  ) throws AdapterException {
     try {
       new DataStreamVerifier(stream).verifyAndAdd(principalSid, publicElement);
     } catch (SepaParseException e) {
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/message/Notifications.java b/streampipes-model/src/main/java/org/apache/streampipes/model/message/Notifications.java
index 4e14af9262..965a84257d 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/message/Notifications.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/message/Notifications.java
@@ -44,8 +44,12 @@ public class Notifications {
     return new ErrorMessage(new Notification(type.name(), type.description()));
   }
 
+  public static ErrorMessage error(String message, String description) {
+    return new ErrorMessage(new Notification(message, description));
+  }
+
   public static ErrorMessage error(String message) {
-    return new ErrorMessage(new Notification(message, ""));
+    return error(message, "");
   }
 
   public static ErrorMessage error(NotificationType type, String info) {
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/AppendOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/AppendOutputStrategy.java
index 9d22313bc4..fa7ccca3c4 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/AppendOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/AppendOutputStrategy.java
@@ -26,8 +26,6 @@ import java.util.List;
 
 public class AppendOutputStrategy extends OutputStrategy {
 
-  private static final long serialVersionUID = 7202888911899551012L;
-
   private List<EventProperty> eventProperties;
 
   public AppendOutputStrategy() {
@@ -35,6 +33,11 @@ public class AppendOutputStrategy extends OutputStrategy {
     eventProperties = new ArrayList<>();
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
+
   public AppendOutputStrategy(AppendOutputStrategy other) {
     super(other);
     this.setEventProperties(new Cloner().properties(other.getEventProperties()));
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomOutputStrategy.java
index 4468ba14f2..fa859aeef4 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomOutputStrategy.java
@@ -23,8 +23,6 @@ import java.util.List;
 
 public class CustomOutputStrategy extends OutputStrategy {
 
-  private static final long serialVersionUID = -5858193127308435472L;
-
   private List<String> selectedPropertyKeys;
 
   private boolean outputRight;
@@ -77,4 +75,9 @@ public class CustomOutputStrategy extends OutputStrategy {
   public void setAvailablePropertyKeys(List<String> availablePropertyKeys) {
     this.availablePropertyKeys = availablePropertyKeys;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomTransformOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomTransformOutputStrategy.java
index 91d1a6ae3d..426a8d29eb 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomTransformOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/CustomTransformOutputStrategy.java
@@ -32,7 +32,6 @@ public class CustomTransformOutputStrategy extends OutputStrategy {
     this.eventProperties = new ArrayList<>();
   }
 
-
   public CustomTransformOutputStrategy(CustomTransformOutputStrategy other) {
     super(other);
     this.eventProperties = new Cloner().properties(other.getEventProperties());
@@ -45,4 +44,9 @@ public class CustomTransformOutputStrategy extends OutputStrategy {
   public void setEventProperties(List<EventProperty> eventProperties) {
     this.eventProperties = eventProperties;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/FixedOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/FixedOutputStrategy.java
index e698348efd..1558aada49 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/FixedOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/FixedOutputStrategy.java
@@ -50,5 +50,9 @@ public class FixedOutputStrategy extends OutputStrategy {
     this.eventProperties = eventProperties;
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/KeepOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/KeepOutputStrategy.java
index d455e7b4c2..0c8b9b0e00 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/KeepOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/KeepOutputStrategy.java
@@ -63,4 +63,9 @@ public class KeepOutputStrategy extends OutputStrategy {
   public void setKeepBoth(boolean keepBoth) {
     this.keepBoth = keepBoth;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/ListOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/ListOutputStrategy.java
index 4040d9387c..b2e1f4ba51 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/ListOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/ListOutputStrategy.java
@@ -46,5 +46,9 @@ public class ListOutputStrategy extends OutputStrategy {
     this.propertyName = propertyName;
   }
 
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategy.java
index 05484f1679..ab19a8d89b 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategy.java
@@ -75,4 +75,6 @@ public abstract class OutputStrategy {
   public void setRenameRules(List<PropertyRenameRule> renameRules) {
     this.renameRules = renameRules;
   }
+
+  public abstract void accept(OutputStrategyVisitor visitor);
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategyVisitor.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategyVisitor.java
new file mode 100644
index 0000000000..024e22b994
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/OutputStrategyVisitor.java
@@ -0,0 +1,38 @@
+/*
+ * 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.output;
+
+public interface OutputStrategyVisitor {
+
+  void visit(AppendOutputStrategy appendOutputStrategy);
+
+  void visit(CustomOutputStrategy customOutputStrategy);
+
+  void visit(CustomTransformOutputStrategy customTransformOutputStrategy);
+
+  void visit(FixedOutputStrategy fixedOutputStrategy);
+
+  void visit(KeepOutputStrategy keepOutputStrategy);
+
+  void visit(ListOutputStrategy listOutputStrategy);
+
+  void visit(TransformOutputStrategy transformOutputStrategy);
+
+  void visit(UserDefinedOutputStrategy userDefinedOutputStrategy);
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/TransformOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/TransformOutputStrategy.java
index 1ca951654b..7e9de992ad 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/TransformOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/TransformOutputStrategy.java
@@ -43,4 +43,9 @@ public class TransformOutputStrategy extends OutputStrategy {
   public void setTransformOperations(List<TransformOperation> transformOperations) {
     this.transformOperations = transformOperations;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/output/UserDefinedOutputStrategy.java b/streampipes-model/src/main/java/org/apache/streampipes/model/output/UserDefinedOutputStrategy.java
index 7f8f725eef..622d1e25ec 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/output/UserDefinedOutputStrategy.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/output/UserDefinedOutputStrategy.java
@@ -42,4 +42,9 @@ public class UserDefinedOutputStrategy extends OutputStrategy {
   public void setEventProperties(List<EventProperty> eventProperties) {
     this.eventProperties = eventProperties;
   }
+
+  @Override
+  public void accept(OutputStrategyVisitor visitor) {
+    visitor.visit(this);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/CompactPipelineElement.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/CompactPipelineElement.java
index b145c2aad3..f027e60e05 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/CompactPipelineElement.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/CompactPipelineElement.java
@@ -28,5 +28,6 @@ public record CompactPipelineElement(String type,
                                      String ref,
                                      String id,
                                      List<String> connectedTo,
-                                     List<Map<String, Object>> configuration) {
+                                     List<Map<String, Object>> configuration,
+                                     OutputConfiguration output) {
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/OutputConfiguration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/OutputConfiguration.java
new file mode 100644
index 0000000000..f9d88fa97d
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/OutputConfiguration.java
@@ -0,0 +1,28 @@
+/*
+ * 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.pipeline.compact;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import java.util.List;
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public record OutputConfiguration(List<String> keep,
+                                  List<UserDefinedOutput> userDefined) {
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/UserDefinedOutput.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/UserDefinedOutput.java
new file mode 100644
index 0000000000..22b9b924a1
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/compact/UserDefinedOutput.java
@@ -0,0 +1,24 @@
+/*
+ * 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.pipeline.compact;
+
+public record UserDefinedOutput(String fieldName,
+                                String runtimeType,
+                                String semanticType) {
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/CompactPipelineConverter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/CompactPipelineConverter.java
index bebc3b9286..c3230822b3 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/CompactPipelineConverter.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/CompactPipelineConverter.java
@@ -19,8 +19,15 @@
 package org.apache.streampipes.manager.pipeline.compact.generation;
 
 import org.apache.streampipes.manager.template.CompactConfigGenerator;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.output.OutputStrategy;
+import org.apache.streampipes.model.output.UserDefinedOutputStrategy;
 import org.apache.streampipes.model.pipeline.Pipeline;
 import org.apache.streampipes.model.pipeline.compact.CompactPipelineElement;
+import org.apache.streampipes.model.pipeline.compact.OutputConfiguration;
+import org.apache.streampipes.model.pipeline.compact.UserDefinedOutput;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 
 import java.util.ArrayList;
@@ -38,6 +45,7 @@ public class CompactPipelineConverter {
             stream.getDom(),
             stream.getElementId(),
             null,
+            null,
             null))
         .forEach(pipelineElements::add);
 
@@ -47,7 +55,8 @@ public class CompactPipelineConverter {
             processor.getDom(),
             processor.getAppId(),
             processor.getConnectedTo(),
-            getConfig(processor.getStaticProperties())))
+            getConfig(processor.getStaticProperties()),
+            getOutput(processor.getOutputStrategies().get(0))))
         .forEach(pipelineElements::add);
 
     pipeline.getActions().stream()
@@ -56,7 +65,8 @@ public class CompactPipelineConverter {
             sink.getDom(),
             sink.getAppId(),
             sink.getConnectedTo(),
-            getConfig(sink.getStaticProperties())))
+            getConfig(sink.getStaticProperties()),
+            null))
         .forEach(pipelineElements::add);
 
     return pipelineElements;
@@ -66,11 +76,12 @@ public class CompactPipelineConverter {
                                                String ref,
                                                String elementId,
                                                List<String> connectedTo,
-                                               List<Map<String, Object>> config) {
+                                               List<Map<String, Object>> config,
+                                               OutputConfiguration outputConfiguration) {
     var connections = connectedTo != null ? connectedTo.stream()
         .map(this::replaceId)
         .toList() : null;
-    return new CompactPipelineElement(type, replaceId(ref), elementId, connections, config);
+    return new CompactPipelineElement(type, replaceId(ref), elementId, connections, config, outputConfiguration);
   }
 
   public List<Map<String, Object>> getConfig(List<StaticProperty> staticProperties) {
@@ -79,6 +90,26 @@ public class CompactPipelineConverter {
     return configs;
   }
 
+  public OutputConfiguration getOutput(OutputStrategy outputStrategy) {
+    if (outputStrategy instanceof CustomOutputStrategy) {
+      return new OutputConfiguration(((CustomOutputStrategy) outputStrategy).getSelectedPropertyKeys(), null);
+    } else if (outputStrategy instanceof UserDefinedOutputStrategy) {
+      return new OutputConfiguration(
+          null,
+          toCustomConfig(((UserDefinedOutputStrategy) outputStrategy).getEventProperties())
+      );
+    } else {
+      return null;
+    }
+  }
+
+  private List<UserDefinedOutput> toCustomConfig(List<EventProperty> eventProperties) {
+    return eventProperties.stream().map(ep -> new UserDefinedOutput(
+        ep.getRuntimeName(),
+        ((EventPropertyPrimitive) ep).getRuntimeType(),
+        ep.getSemanticType())).toList();
+  }
+
   private String replaceId(String id) {
     return id.replaceAll(InvocablePipelineElementGenerator.ID_PREFIX, "");
   }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/DataProcessorPipelineElementGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/DataProcessorPipelineElementGenerator.java
index c736293c20..89f7fd10dd 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/DataProcessorPipelineElementGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/DataProcessorPipelineElementGenerator.java
@@ -35,7 +35,13 @@ public class DataProcessorPipelineElementGenerator {
                                           CompactPipelineElement pipelineElement) {
     basicGenerator.apply(processor, pipelineElement);
     var template = basicGenerator.makeTemplate(processor, pipelineElement);
-    return new DataProcessorTemplateHandler(template, processor, false)
+    var element = new DataProcessorTemplateHandler(template, processor, false)
         .applyTemplateOnPipelineElement();
+
+    if (pipelineElement.output() != null) {
+      var outputStrategyGenerator = new OutputStrategyGenerator(pipelineElement.output());
+      element.getOutputStrategies().forEach(o -> o.accept(outputStrategyGenerator));
+    }
+    return element;
   }
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/OutputStrategyGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/OutputStrategyGenerator.java
new file mode 100644
index 0000000000..2cb4cd6067
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/compact/generation/OutputStrategyGenerator.java
@@ -0,0 +1,108 @@
+/*
+ * 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.pipeline.compact.generation;
+
+import org.apache.streampipes.model.output.AppendOutputStrategy;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+import org.apache.streampipes.model.output.CustomTransformOutputStrategy;
+import org.apache.streampipes.model.output.FixedOutputStrategy;
+import org.apache.streampipes.model.output.KeepOutputStrategy;
+import org.apache.streampipes.model.output.ListOutputStrategy;
+import org.apache.streampipes.model.output.OutputStrategyVisitor;
+import org.apache.streampipes.model.output.TransformOutputStrategy;
+import org.apache.streampipes.model.output.UserDefinedOutputStrategy;
+import org.apache.streampipes.model.pipeline.compact.OutputConfiguration;
+import org.apache.streampipes.model.pipeline.compact.UserDefinedOutput;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+
+import java.util.List;
+
+public class OutputStrategyGenerator implements OutputStrategyVisitor {
+
+  private final OutputConfiguration config;
+
+  public OutputStrategyGenerator(OutputConfiguration outputConfiguration) {
+    this.config = outputConfiguration;
+  }
+
+  @Override
+  public void visit(AppendOutputStrategy appendOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(CustomOutputStrategy customOutputStrategy) {
+    var keepConfig = config.keep();
+    if (keepConfig != null && !keepConfig.isEmpty()) {
+      customOutputStrategy.setSelectedPropertyKeys(keepConfig);
+    }
+  }
+
+  @Override
+  public void visit(CustomTransformOutputStrategy customTransformOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(FixedOutputStrategy fixedOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(KeepOutputStrategy keepOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(ListOutputStrategy listOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(TransformOutputStrategy transformOutputStrategy) {
+
+  }
+
+  @Override
+  public void visit(UserDefinedOutputStrategy userDefinedOutputStrategy) {
+    var userDefinedConfig = config.userDefined();
+    if (userDefinedConfig != null && !userDefinedConfig.isEmpty()) {
+      userDefinedOutputStrategy.setEventProperties(
+          toEp(userDefinedConfig)
+      );
+    }
+  }
+
+  private List<EventProperty> toEp(List<UserDefinedOutput> userDefinedOutput) {
+    return userDefinedOutput
+        .stream()
+        .map(this::toPrimitive)
+        .toList();
+  }
+
+  private EventProperty toPrimitive(UserDefinedOutput u) {
+    var ep = new EventPropertyPrimitive();
+    ep.setRuntimeName(u.fieldName());
+    ep.setSemanticType(u.semanticType());
+    ep.setRuntimeType(u.runtimeType());
+
+    return ep;
+  }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/compact/CompactPipelineTemplateManagement.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/compact/CompactPipelineTemplateManagement.java
index 0bf308cfc4..1238a0aeda 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/compact/CompactPipelineTemplateManagement.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/compact/CompactPipelineTemplateManagement.java
@@ -57,7 +57,8 @@ public class CompactPipelineTemplateManagement {
             key,
             stream.getElementId(),
             List.of(),
-            List.of()
+            List.of(),
+            null
         ));
       });
     }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/instances/PersistDataLakePipelineTemplate.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/instances/PersistDataLakePipelineTemplate.java
index caa6e7bf94..00c768f4a7 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/instances/PersistDataLakePipelineTemplate.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/instances/PersistDataLakePipelineTemplate.java
@@ -52,7 +52,8 @@ public class PersistDataLakePipelineTemplate implements DefaultPipelineTemplateP
                 List.of(
                     Map.of("schema_update", "Update schema"),
                     Map.of("ignore_duplicates", false)
-                )
+                ),
+                null
             )
         )
     );
diff --git a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java
index 72c77121aa..8e3d4d4a85 100644
--- a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java
+++ b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java
@@ -17,6 +17,7 @@
  */
 package org.apache.streampipes.resource.management;
 
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.util.Cloner;
 import org.apache.streampipes.resource.management.secret.SecretProvider;
@@ -30,7 +31,8 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor
   }
 
   public AdapterResourceManager() {
-    super(StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage());
+    super(StorageDispatcher.INSTANCE.getNoSqlStore()
+                                    .getAdapterInstanceStorage());
   }
 
   /**
@@ -39,10 +41,15 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor
    * @param adapterDescription input adapter description
    * @return the id of the created adapter
    */
-  public String encryptAndCreate(AdapterDescription adapterDescription) {
-    AdapterDescription encryptedAdapterDescription = cloneAndEncrypt(adapterDescription);
+  public String encryptAndCreate(AdapterDescription adapterDescription) throws AdapterException {
+    var encryptedAdapterDescription = cloneAndEncrypt(adapterDescription);
     encryptedAdapterDescription.setRev(null);
-    return db.persist(encryptedAdapterDescription).v;
+
+    try {
+      return db.persist(encryptedAdapterDescription).v;
+    } catch (org.lightcouch.DocumentConflictException e) {
+      throw new AdapterException("Conflict occurred while creating the adapter", e);
+    }
   }
 
   /**
@@ -50,8 +57,15 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor
    *
    * @param adapterDescription input adapter description
    */
-  public void encryptAndUpdate(AdapterDescription adapterDescription) {
-    db.updateElement(cloneAndEncrypt(adapterDescription));
+  public void encryptAndUpdate(AdapterDescription adapterDescription) throws AdapterException {
+    try {
+      db.updateElement(cloneAndEncrypt(adapterDescription));
+    } catch (org.lightcouch.DocumentConflictException e) {
+      throw new AdapterException(
+          "Conflict occurred while editing the adapter with id: %s".formatted(adapterDescription.getElementId()),
+          e
+      );
+    }
   }
 
   public void delete(String elementId) {
@@ -63,7 +77,8 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor
    */
   private AdapterDescription cloneAndEncrypt(AdapterDescription adapterDescription) {
     AdapterDescription encryptedAdapterDescription = new Cloner().adapterDescription(adapterDescription);
-    SecretProvider.getEncryptionService().apply(encryptedAdapterDescription);
+    SecretProvider.getEncryptionService()
+                  .apply(encryptedAdapterDescription);
     return encryptedAdapterDescription;
   }
 
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
index b972aaa7eb..372769a792 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
@@ -40,6 +40,7 @@ import org.apache.streampipes.rest.shared.constants.SpMediaType;
 import org.apache.streampipes.storage.api.IPipelineStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.http.HttpStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -115,7 +116,7 @@ public class AdapterResource extends AbstractAdapterResource<AdapterMasterManage
       updateManager.updateAdapter(adapterDescription);
     } catch (AdapterException e) {
       LOG.error("Error while updating adapter with id {}", adapterDescription.getElementId(), e);
-      return ok(Notifications.error(e.getMessage()));
+      return ok(Notifications.error(e.getMessage(), ExceptionUtils.getStackTrace(e)));
     }
 
     return ok(Notifications.success(adapterDescription.getElementId()));
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
index 88424c2256..0e51f00927 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
@@ -37,6 +37,7 @@ import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -58,7 +59,7 @@ public class CompactAdapterResource extends AbstractAdapterResource<AdapterMaste
   public CompactAdapterResource() {
     super(() -> new AdapterMasterManagement(
         StorageDispatcher.INSTANCE.getNoSqlStore()
-            .getAdapterInstanceStorage(),
+                                  .getAdapterInstanceStorage(),
         new SpResourceManager().manageAdapters(),
         new SpResourceManager().manageDataStreams(),
         AdapterMetricsManager.INSTANCE.getAdapterMetrics()
@@ -82,11 +83,23 @@ public class CompactAdapterResource extends AbstractAdapterResource<AdapterMaste
     var adapterDescription = getGeneratedAdapterDescription(compactAdapter);
     var principalSid = getAuthenticatedUserSid();
 
+    var adapterId = adapterDescription.getElementId();
+
     try {
-      var adapterId = adapterDescription.getElementId();
       managementService.addAdapter(adapterDescription, adapterId, principalSid);
+    } catch (AdapterException e) {
+      LOG.error(
+          "Error while storing the adapterDescription with appId {}. An adapter with the given id already exists.",
+          adapterDescription.getAppId(), e
+      );
+      return ResponseEntity.status(HttpStatus.CONFLICT)
+                           .body(Notifications.error(e.getMessage()));
+    }
+
+    try {
       if (compactAdapter.createOptions() != null) {
-        if (compactAdapter.createOptions().persist()) {
+        if (compactAdapter.createOptions()
+                          .persist()) {
           var storedAdapter = managementService.getAdapter(adapterId);
           var status = new PersistPipelineHandler(
               getNoSqlStorage().getPipelineTemplateStorage(),
@@ -96,7 +109,8 @@ public class CompactAdapterResource extends AbstractAdapterResource<AdapterMaste
               getAuthenticatedUserSid()
           ).createAndStartPersistPipeline(storedAdapter);
         }
-        if (compactAdapter.createOptions().start()) {
+        if (compactAdapter.createOptions()
+                          .start()) {
           managementService.startStreamAdapter(adapterId);
         }
       }
@@ -143,8 +157,10 @@ public class CompactAdapterResource extends AbstractAdapterResource<AdapterMaste
     return new CompactAdapterManagement(generators).convertToAdapterDescription(compactAdapter);
   }
 
-  private AdapterDescription getGeneratedAdapterDescription(CompactAdapter compactAdapter,
-                                                            AdapterDescription existingAdapter) throws Exception {
+  private AdapterDescription getGeneratedAdapterDescription(
+      CompactAdapter compactAdapter,
+      AdapterDescription existingAdapter
+  ) throws Exception {
     var generators = adapterGenerationSteps.getGenerators();
     return new CompactAdapterManagement(generators).convertToAdapterDescription(compactAdapter, existingAdapter);
   }
diff --git a/ui/cypress/fixtures/connect/compact/machineDataSimulator.yml b/ui/cypress/fixtures/connect/compact/machineDataSimulator.yml
new file mode 100644
index 0000000000..128b249ca6
--- /dev/null
+++ b/ui/cypress/fixtures/connect/compact/machineDataSimulator.yml
@@ -0,0 +1,24 @@
+# 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.
+
+id: sp:adapterdescription:4pxl2w
+name: Test
+appId: org.apache.streampipes.connect.iiot.adapters.simulator.machine
+configuration:
+    - wait-time-ms: '1000'
+    - selected-simulator-option: flowrate
+createOptions:
+    persist: false
+    start: false
diff --git a/ui/cypress/support/builder/CompactAdapterBuilder.ts b/ui/cypress/support/builder/CompactAdapterBuilder.ts
new file mode 100644
index 0000000000..2325985032
--- /dev/null
+++ b/ui/cypress/support/builder/CompactAdapterBuilder.ts
@@ -0,0 +1,93 @@
+/*
+ *  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 { CompactAdapter } from '../../../projects/streampipes/platform-services/src/lib/model/gen/streampipes-model';
+
+export class CompactAdapterBuilder {
+    private compactAdapter: CompactAdapter;
+
+    constructor() {
+        this.compactAdapter = new CompactAdapter();
+        this.compactAdapter.configuration = [];
+        this.compactAdapter.createOptions = {
+            persist: false,
+            start: false,
+        };
+        this.compactAdapter.transform = {
+            rename: {},
+            measurementUnit: {},
+        };
+    }
+
+    public static create(appId: string) {
+        const builder = new CompactAdapterBuilder();
+        builder.compactAdapter.appId = appId;
+        return builder;
+    }
+
+    // Optional parameter, when not set a random id will be generated
+    public setId(id: string) {
+        this.compactAdapter.id = id;
+        return this;
+    }
+
+    public setName(name: string) {
+        this.compactAdapter.name = name;
+        return this;
+    }
+
+    public setDescription(description: string) {
+        this.compactAdapter.description = description;
+        return this;
+    }
+
+    public withRename(from: string, to: string) {
+        this.compactAdapter.transform.rename[from] = to;
+        return this;
+    }
+
+    public withMeasurementUnit(property: string, unit: string) {
+        this.compactAdapter.transform.measurementUnit[property] = unit;
+        return this;
+    }
+
+    public addConfiguration(key: string, value: string) {
+        const configuration = { [key]: value };
+        this.compactAdapter.configuration.push(configuration);
+        return this;
+    }
+
+    public setPersist() {
+        this.compactAdapter.createOptions.persist = true;
+        return this;
+    }
+
+    public setStart() {
+        this.compactAdapter.createOptions.start = true;
+        return this;
+    }
+
+    public build() {
+        if (!this.compactAdapter.id) {
+            this.compactAdapter.id =
+                'sp:adapterdescription:' +
+                Math.random().toString(36).substring(7);
+        }
+        return this.compactAdapter;
+    }
+}
diff --git a/ui/cypress/support/utils/CompactUtils.ts b/ui/cypress/support/utils/CompactUtils.ts
new file mode 100644
index 0000000000..38df176d86
--- /dev/null
+++ b/ui/cypress/support/utils/CompactUtils.ts
@@ -0,0 +1,27 @@
+/*
+ *  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.
+ *
+ */
+
+export class CompactUtils {
+    public static ymlConfiguration() {
+        return cy.dataCy('yaml-configuration', { timeout: 5000 });
+    }
+
+    public static jsonConfiguration() {
+        return cy.dataCy('json-configuration', { timeout: 5000 });
+    }
+}
diff --git a/ui/cypress/support/utils/GeneralUtils.ts b/ui/cypress/support/utils/GeneralUtils.ts
new file mode 100644
index 0000000000..8db505e32b
--- /dev/null
+++ b/ui/cypress/support/utils/GeneralUtils.ts
@@ -0,0 +1,23 @@
+/*
+ *  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.
+ *
+ */
+
+export class GeneralUtils {
+    public static tab(identifier: string) {
+        return cy.dataCy(`tab-${identifier}`).click();
+    }
+}
diff --git a/ui/cypress/support/utils/ThirdPartyIntegrationUtils.ts b/ui/cypress/support/utils/ThirdPartyIntegrationUtils.ts
index a9229a9a05..09a871c6fb 100644
--- a/ui/cypress/support/utils/ThirdPartyIntegrationUtils.ts
+++ b/ui/cypress/support/utils/ThirdPartyIntegrationUtils.ts
@@ -18,7 +18,7 @@
 
 import { ConnectUtils } from './connect/ConnectUtils';
 import { PipelineBuilder } from '../builder/PipelineBuilder';
-import { PipelineUtils } from './PipelineUtils';
+import { PipelineUtils } from './pipeline/PipelineUtils';
 import { PipelineElementInput } from '../model/PipelineElementInput';
 import { AdapterInput } from '../model/AdapterInput';
 import { AdapterBuilder } from '../builder/AdapterBuilder';
diff --git a/ui/cypress/support/utils/connect/CompactAdapterUtils.ts b/ui/cypress/support/utils/connect/CompactAdapterUtils.ts
new file mode 100644
index 0000000000..43757f580c
--- /dev/null
+++ b/ui/cypress/support/utils/connect/CompactAdapterUtils.ts
@@ -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.
+ *
+ */
+
+import { CompactAdapter } from '../../../../projects/streampipes/platform-services/src/lib/model/gen/streampipes-model';
+import { CompactAdapterBuilder } from '../../builder/CompactAdapterBuilder';
+
+export class CompactAdapterUtils {
+    /**
+     * Stores a compact adapter by sending a POST request to the backend.
+     *
+     * @param {CompactAdapter} compactAdapter - The compact adapter to be stored.
+     * @param {boolean} [failOnStatusCode=true] - Whether to fail the request on a non-2xx status code.
+     * @returns {Cypress.Chainable} - The Cypress chainable object representing the request.
+     */
+    public static storeCompactAdapter(
+        compactAdapter: CompactAdapter,
+        failOnStatusCode: boolean = true,
+    ): Cypress.Chainable {
+        return this.postCompactAdapterRequest(
+            'application/json',
+            compactAdapter,
+            failOnStatusCode,
+        );
+    }
+
+    /**
+     * Stores a compact YAML adapter by sending a POST request to the backend.
+     *
+     * @param {string} body - The YAML string representing the compact adapter.
+     * @returns {Cypress.Chainable} - The Cypress chainable object representing the request.
+     */
+    public static storeCompactYmlAdapter(body: string) {
+        return this.postCompactAdapterRequest('application/yml', body);
+    }
+
+    private static postCompactAdapterRequest(
+        contentType: string,
+        body: any,
+        failOnStatusCode = true,
+    ) {
+        const token = window.localStorage.getItem('auth-token');
+        return cy.request({
+            method: 'POST',
+            url: '/streampipes-backend/api/v2/connect/compact-adapters',
+            body: body,
+            failOnStatusCode: failOnStatusCode,
+            headers: {
+                'Accept': contentType,
+                'Content-Type': contentType,
+                'Authorization': `Bearer ${token}`,
+            },
+        });
+    }
+
+    /**
+     * Creates a CompactAdapterBuilder instance configured for a machine data simulator.
+     */
+    public static getMachineDataSimulator(): CompactAdapterBuilder {
+        return CompactAdapterBuilder.create(
+            'org.apache.streampipes.connect.iiot.adapters.simulator.machine',
+        )
+            .setName('Test')
+            .addConfiguration('wait-time-ms', '1000')
+            .addConfiguration('selected-simulator-option', 'flowrate');
+    }
+}
diff --git a/ui/cypress/support/utils/connect/ConnectBtns.ts b/ui/cypress/support/utils/connect/ConnectBtns.ts
index 20bfd696c5..ea8b2d82c3 100644
--- a/ui/cypress/support/utils/connect/ConnectBtns.ts
+++ b/ui/cypress/support/utils/connect/ConnectBtns.ts
@@ -17,7 +17,11 @@
  */
 export class ConnectBtns {
     public static detailsAdapter() {
-        return cy.dataCy('details-adapter');
+        return cy.dataCy('details-adapter', { timeout: 10000 });
+    }
+
+    public static deleteAdapter() {
+        return cy.dataCy('delete-adapter', { timeout: 10000 });
     }
 
     public static editAdapter() {
@@ -69,6 +73,10 @@ export class ConnectBtns {
         return cy.dataCy('stop-all-adapters-btn');
     }
 
+    public static showCodeCheckbox() {
+        return cy.dataCy('show-code-checkbox');
+    }
+
     // ========================================================================
 
     // =====================  Event Schema buttons  ==========================
diff --git a/ui/cypress/support/utils/connect/ConnectUtils.ts b/ui/cypress/support/utils/connect/ConnectUtils.ts
index aaeccff82c..0a0eae22c1 100644
--- a/ui/cypress/support/utils/connect/ConnectUtils.ts
+++ b/ui/cypress/support/utils/connect/ConnectUtils.ts
@@ -23,7 +23,7 @@ import { DataLakeUtils } from '../datalake/DataLakeUtils';
 import { ConnectBtns } from './ConnectBtns';
 import { AdapterBuilder } from '../../builder/AdapterBuilder';
 import { UserUtils } from '../UserUtils';
-import { PipelineUtils } from '../PipelineUtils';
+import { PipelineUtils } from '../pipeline/PipelineUtils';
 
 export class ConnectUtils {
     public static testAdapter(
@@ -76,9 +76,11 @@ export class ConnectUtils {
             );
         }
 
-        ConnectEventSchemaUtils.markPropertyAsTimestamp(
-            adapterConfiguration.timestampProperty,
-        );
+        if (adapterConfiguration.timestampProperty) {
+            ConnectEventSchemaUtils.markPropertyAsTimestamp(
+                adapterConfiguration.timestampProperty,
+            );
+        }
 
         ConnectEventSchemaUtils.finishEventSchemaConfiguration();
     }
@@ -356,6 +358,12 @@ export class ConnectUtils {
         ConnectUtils.validateEventsInPreview(amountOfProperties);
     }
 
+    public static getLivePreviewValue(runtimeName: string) {
+        return cy.dataCy(`live-preview-value-${runtimeName}`, {
+            timeout: 10000,
+        });
+    }
+
     public static validateEventsInPreview(amountOfProperties: number) {
         // View data
         ConnectBtns.detailsAdapter().click();
@@ -370,11 +378,25 @@ export class ConnectUtils {
             amountOfProperties,
         );
 
-        cy.wait(1000);
+        cy.dataCy('live-preview-table-no-data', { timout: 1000 }).should(
+            'not.exist',
+        );
+    }
+
+    /**
+     * Validates the event schema for an adapter by checking the amount of properties
+     * and the runtime names of the event properties
+     * @param runtimeNames runtime names of the event properties
+     */
+    public static validateEventSchema(runtimeNames: string[]) {
+        ConnectUtils.goToConnect();
+        ConnectBtns.detailsAdapter().click();
+
+        cy.get('tr.mat-mdc-row').should('have.length', runtimeNames.length);
 
-        cy.dataCy('live-preview-table-value')
-            .invoke('text')
-            .then(text => expect(text).not.to.include('no data'));
+        runtimeNames.forEach(name => {
+            cy.get('td.mat-column-runtimeName').contains(name).should('exist');
+        });
     }
 
     public static tearDownPreprocessingRuleTest(
@@ -403,4 +425,21 @@ export class ConnectUtils {
         // Close dialog
         cy.get('button').contains('Close').parent().click();
     }
+
+    public static validateAdapterIsRunning() {
+        ConnectUtils.goToConnect();
+        ConnectBtns.startAdapter().should('have.length', 0);
+        ConnectBtns.stopAdapter().should('have.length', 1);
+    }
+
+    public static validateAdapterIsStopped() {
+        ConnectUtils.goToConnect();
+        ConnectBtns.startAdapter().should('have.length', 1);
+        ConnectBtns.stopAdapter().should('have.length', 0);
+    }
+
+    public static checkAmountOfAdapters(amount: number) {
+        ConnectUtils.goToConnect();
+        ConnectBtns.deleteAdapter().should('have.length', amount);
+    }
 }
diff --git a/ui/cypress/support/utils/pipeline/PipelineBtns.ts b/ui/cypress/support/utils/pipeline/PipelineBtns.ts
new file mode 100644
index 0000000000..6ace82220f
--- /dev/null
+++ b/ui/cypress/support/utils/pipeline/PipelineBtns.ts
@@ -0,0 +1,23 @@
+/*
+ *  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.
+ *
+ */
+
+export class PipelineBtns {
+    public static deletePipeline() {
+        return cy.dataCy('delete-pipeline', { timeout: 10000 });
+    }
+}
diff --git a/ui/cypress/support/utils/PipelineUtils.ts b/ui/cypress/support/utils/pipeline/PipelineUtils.ts
similarity index 72%
rename from ui/cypress/support/utils/PipelineUtils.ts
rename to ui/cypress/support/utils/pipeline/PipelineUtils.ts
index b901e225de..0fa9ce4443 100644
--- a/ui/cypress/support/utils/PipelineUtils.ts
+++ b/ui/cypress/support/utils/pipeline/PipelineUtils.ts
@@ -1,25 +1,26 @@
 /*
- * 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
+ *  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
+ *     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.
+ *  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 { PipelineInput } from '../model/PipelineInput';
-import { StaticPropertyUtils } from './userInput/StaticPropertyUtils';
-import { OutputStrategyUtils } from './OutputStrategyUtils';
-import { PipelineElementInput } from '../model/PipelineElementInput';
+import { PipelineInput } from '../../model/PipelineInput';
+import { StaticPropertyUtils } from '../userInput/StaticPropertyUtils';
+import { OutputStrategyUtils } from '../OutputStrategyUtils';
+import { PipelineElementInput } from '../../model/PipelineElementInput';
+import { PipelineBtns } from './PipelineBtns';
 
 export class PipelineUtils {
     public static addPipeline(pipelineInput: PipelineInput) {
@@ -131,21 +132,18 @@ export class PipelineUtils {
     }
 
     public static checkAmountOfPipelinesPipeline(amount: number) {
-        cy.visit('#/pipelines');
-        cy.dataCy('delete-pipeline').should('have.length', amount);
+        PipelineUtils.goToPipelines();
+        PipelineBtns.deletePipeline().should('have.length', amount);
     }
 
     public static deletePipeline() {
         // Delete pipeline
-        cy.visit('#/pipelines');
-        cy.dataCy('delete-pipeline').should('have.length', 1);
-        cy.dataCy('delete-pipeline').click({ force: true });
+        PipelineUtils.goToPipelines();
+        PipelineBtns.deletePipeline().should('have.length', 1);
+        PipelineBtns.deletePipeline().click({ force: true });
 
         cy.dataCy('sp-pipeline-stop-and-delete').click();
 
-        cy.dataCy('delete-pipeline', { timeout: 10000 }).should(
-            'have.length',
-            0,
-        );
+        PipelineBtns.deletePipeline().should('have.length', 0);
     }
 }
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index cf07248af8..3aa23a65bf 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -20,7 +20,7 @@
 /* tslint:disable */
 /* eslint-disable */
 // @ts-nocheck
-// Generated using typescript-generator version 3.2.1263 on 2024-10-11 10:41:46.
+// Generated using typescript-generator version 3.2.1263 on 2024-10-29 10:04:46.
 
 export class NamedStreamPipesEntity implements Storable {
     '@class':
@@ -844,6 +844,7 @@ export class CompactPipelineElement {
     configuration: { [index: string]: any }[];
     connectedTo: string[];
     id: string;
+    output: OutputConfiguration;
     ref: string;
     type: string;
 
@@ -862,6 +863,7 @@ export class CompactPipelineElement {
             data.connectedTo,
         );
         instance.id = data.id;
+        instance.output = OutputConfiguration.fromData(data.output);
         instance.ref = data.ref;
         instance.type = data.type;
         return instance;
@@ -2605,6 +2607,26 @@ export class Option {
     }
 }
 
+export class OutputConfiguration {
+    keep: string[];
+    userDefined: UserDefinedOutput[];
+
+    static fromData(
+        data: OutputConfiguration,
+        target?: OutputConfiguration,
+    ): OutputConfiguration {
+        if (!data) {
+            return data;
+        }
+        const instance = target || new OutputConfiguration();
+        instance.keep = __getCopyArrayFn(__identity<string>())(data.keep);
+        instance.userDefined = __getCopyArrayFn(UserDefinedOutput.fromData)(
+            data.userDefined,
+        );
+        return instance;
+    }
+}
+
 export class Pipeline implements Storable {
     _id: string;
     _rev: string;
@@ -4056,6 +4078,26 @@ export class UnitTransformRuleDescription extends ValueTransformationRuleDescrip
     }
 }
 
+export class UserDefinedOutput {
+    fieldName: string;
+    runtimeType: string;
+    semanticType: string;
+
+    static fromData(
+        data: UserDefinedOutput,
+        target?: UserDefinedOutput,
+    ): UserDefinedOutput {
+        if (!data) {
+            return data;
+        }
+        const instance = target || new UserDefinedOutput();
+        instance.fieldName = data.fieldName;
+        instance.runtimeType = data.runtimeType;
+        instance.semanticType = data.semanticType;
+        return instance;
+    }
+}
+
 export class UserDefinedOutputStrategy extends OutputStrategy {
     '@class': 'org.apache.streampipes.model.output.UserDefinedOutputStrategy';
     'eventProperties': EventPropertyUnion[];
diff --git a/ui/projects/streampipes/shared-ui/src/lib/components/basic-nav-tabs/basic-nav-tabs.component.html b/ui/projects/streampipes/shared-ui/src/lib/components/basic-nav-tabs/basic-nav-tabs.component.html
index 68f9377e8f..c613d3cda3 100644
--- a/ui/projects/streampipes/shared-ui/src/lib/components/basic-nav-tabs/basic-nav-tabs.component.html
+++ b/ui/projects/streampipes/shared-ui/src/lib/components/basic-nav-tabs/basic-nav-tabs.component.html
@@ -42,6 +42,7 @@
                     *ngFor="let item of spNavigationItems"
                     (click)="navigateTo(item)"
                     [active]="activeLink === item.itemId"
+                    [attr.data-cy]="'tab-' + item.itemTitle"
                 >
                     <span>{{ item.itemTitle }}</span>
                 </a>
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 f7ff5b8c7b..d4eecdd6a7 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
@@ -150,6 +150,7 @@
             optionTitle="Show code"
             optionDescription="Show code to programmatically deploy this adapter over the API"
             optionIcon="code"
+            dataCy="show-code-checkbox"
             (optionSelectedEmitter)="showCode = $event"
         >
             @if (showCode) {
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 598dae37ae..17cc830821 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
@@ -24,6 +24,7 @@ import {
     CompactPipeline,
     CompactPipelineElement,
     ErrorMessage,
+    Message,
     PipelineOperationStatus,
     PipelineTemplateService,
     PipelineUpdateInfo,
@@ -112,16 +113,22 @@ export class AdapterStartedDialog implements OnInit {
     updateAdapter(): void {
         this.loadingText = `Updating adapter ${this.adapter.name}`;
         this.loading = true;
-        this.adapterService.updateAdapter(this.adapter).subscribe(
-            res => {
-                this.onAdapterReady(
-                    `Adapter ${this.adapter.name} was successfully updated and is available in the pipeline editor.`,
-                );
+        this.adapterService.updateAdapter(this.adapter).subscribe({
+            next: status => {
+                if (status.success) {
+                    this.onAdapterReady(
+                        `Adapter ${this.adapter.name} was successfully updated and is available in the pipeline editor.`,
+                    );
+                } else {
+                    const errorLogMessage = this.getErrorLogMessage(status);
+
+                    this.onAdapterFailure(errorLogMessage);
+                }
             },
-            error => {
+            error: error => {
                 this.onAdapterFailure(error.error);
             },
-        );
+        });
     }
 
     addAdapter() {
@@ -137,16 +144,8 @@ export class AdapterStartedDialog implements OnInit {
                         this.startAdapter(adapterElementId, true);
                     }
                 } else {
-                    const errorMsg: SpLogMessage = {
-                        cause:
-                            status.notifications.length > 0
-                                ? status.notifications[0].title
-                                : 'Unknown Error',
-                        detail: '',
-                        fullStackTrace: '',
-                        level: 'ERROR',
-                        title: 'Unknown Error',
-                    };
+                    const errorMsg: SpLogMessage =
+                        this.getErrorLogMessage(status);
 
                     this.onAdapterFailure(errorMsg);
                 }
@@ -157,6 +156,20 @@ export class AdapterStartedDialog implements OnInit {
         );
     }
 
+    private getErrorLogMessage(status: Message): SpLogMessage {
+        const notification = status.notifications[0] || {
+            title: 'Unknown Error',
+            description: '',
+        };
+        return {
+            cause: notification.title,
+            detail: '',
+            fullStackTrace: notification.description,
+            level: 'ERROR',
+            title: 'Unknown Error',
+        };
+    }
+
     startAdapter(adapterElementId: string, showPreview = false) {
         const successMessage =
             'Your new data stream is now available in the pipeline editor.';
@@ -265,6 +278,7 @@ export class AdapterStartedDialog implements OnInit {
             configuration: undefined,
             id: adapter.correspondingDataStreamElementId,
             connectedTo: undefined,
+            output: undefined,
         });
         return template;
     }
diff --git a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
index 083d9291c7..a6590ce049 100644
--- a/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
+++ b/ui/src/app/core-ui/configuration-code-panel/configuration-code-panel.component.html
@@ -22,6 +22,7 @@
             [innerHTML]="configuration | yamlpretty"
             class="preview-text"
             [ngStyle]="{ maxHeight: maxHeight }"
+            data-cy="yaml-configuration"
         ></pre>
     </mat-tab>
     <mat-tab label="JSON">
@@ -29,6 +30,7 @@
             [innerHTML]="configuration | jsonpretty"
             class="preview-text"
             [ngStyle]="{ maxHeight: maxHeight }"
+            data-cy="json-configuration"
         ></pre>
     </mat-tab>
 </mat-tab-group>
diff --git a/ui/src/app/core-ui/help/help.component.html b/ui/src/app/core-ui/help/help.component.html
index ecc82fc6a8..fcdfbe7332 100644
--- a/ui/src/app/core-ui/help/help.component.html
+++ b/ui/src/app/core-ui/help/help.component.html
@@ -17,18 +17,28 @@
   -->
 
 <div class="sp-dialog-container">
-    <div class="sp-dialog-content p-15">
-        <h4>{{ pipelineElement.name }}</h4>
-        <p>
-            {{ pipelineElement.description }}
-        </p>
-
+    <div class="sp-dialog-content p-15" fxLayout="column">
+        <div fxLayout="row" fxFlex="100">
+            <div fxLayout="column" fxFlex>
+                <h4>{{ pipelineElement.name }}</h4>
+                <small>
+                    {{ pipelineElement.description }}
+                </small>
+            </div>
+            <div class="element-id" fxLayoutAlign="end start">
+                <span>ID</span>&nbsp;<b>{{
+                    isDataStream
+                        ? pipelineElement.elementId
+                        : pipelineElement.appId
+                }}</b>
+            </div>
+        </div>
         <mat-tab-group
             color="accent"
             [selectedIndex]="selectedTabIndex"
             (selectedIndexChange)="selectedTabIndex = $event"
         >
-            <mat-tab *ngFor="let tab of tabs" label="{{ tab }}"> </mat-tab>
+            <mat-tab *ngFor="let tab of tabs" label="{{ tab }}"></mat-tab>
         </mat-tab-group>
 
         <sp-pipeline-element-runtime-info
diff --git a/ui/src/app/core-ui/help/help.component.scss b/ui/src/app/core-ui/help/help.component.scss
index ecf2724830..c077f68f35 100644
--- a/ui/src/app/core-ui/help/help.component.scss
+++ b/ui/src/app/core-ui/help/help.component.scss
@@ -17,3 +17,13 @@
  */
 
 @import '../../../scss/sp/sp-dialog.scss';
+
+.element-id {
+    border-radius: 5px;
+    margin-right: 10px;
+    margin-top: 5px;
+    margin-bottom: 5px;
+    font-size: small;
+    display: inline-block;
+    padding: 5px;
+}
diff --git a/ui/src/app/core-ui/help/help.component.ts b/ui/src/app/core-ui/help/help.component.ts
index 517a2c927d..fcfc5b8d2d 100644
--- a/ui/src/app/core-ui/help/help.component.ts
+++ b/ui/src/app/core-ui/help/help.component.ts
@@ -34,12 +34,14 @@ export class HelpComponent implements OnInit {
 
     @Input()
     pipelineElement: PipelineElementUnion;
+    isDataStream: boolean;
 
     constructor(private dialogRef: DialogRef<HelpComponent>) {}
 
     ngOnInit() {
         if (this.pipelineElement instanceof SpDataStream) {
             this.tabs = this.availableTabs;
+            this.isDataStream = true;
         } else {
             this.tabs.push(this.availableTabs[1]);
             this.selectedTabIndex = 1;
@@ -51,4 +53,6 @@ export class HelpComponent implements OnInit {
             this.dialogRef.close();
         });
     }
+
+    protected readonly SpDataStream = SpDataStream;
 }
diff --git a/ui/src/app/core-ui/pipeline-element-runtime-info/live-preview-table/live-preview-table.component.html b/ui/src/app/core-ui/pipeline-element-runtime-info/live-preview-table/live-preview-table.component.html
index ffaf3bd5b2..ceb6851ab4 100644
--- a/ui/src/app/core-ui/pipeline-element-runtime-info/live-preview-table/live-preview-table.component.html
+++ b/ui/src/app/core-ui/pipeline-element-runtime-info/live-preview-table/live-preview-table.component.html
@@ -57,7 +57,12 @@
             <th mat-header-cell *matHeaderCellDef><strong>Value</strong></th>
             <td mat-cell *matCellDef="let element" style="width: 200px">
                 @if (element.value === undefined) {
-                    <div class="value no-data">no data</div>
+                    <div
+                        data-cy="live-preview-table-no-data"
+                        class="value no-data"
+                    >
+                        no data
+                    </div>
                 } @else {
                     <ng-container *ngIf="element.isImage">
                         <img
@@ -75,7 +80,9 @@
                     </ng-container>
                     <ng-container *ngIf="element.hasNoDomainProperty">
                         <div
-                            data-cy="live-preview-table-value"
+                            [attr.data-cy]="
+                                'live-preview-value-' + element.runtimeName
+                            "
                             class="value"
                             [class.value-changed]="element.valueChanged"
                         >


Test patch:
diff --git a/.github/workflows/go-client-e2e-test.yml b/.github/workflows/go-client-e2e-test.yml
new file mode 100644
index 0000000000..4adec93e16
--- /dev/null
+++ b/.github/workflows/go-client-e2e-test.yml
@@ -0,0 +1,66 @@
+# 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.
+
+name: run-e2e-tests
+
+on:
+  pull_request:
+
+jobs:
+  build-and-push-to-docker:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v1
+
+      - name: Cache Docker layers
+        uses: actions/cache@v2
+        with:
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-
+
+      - name: Set up Go 1.21
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.21'
+
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          distribution: 'temurin'
+          java-version: '17'
+          cache: 'maven'
+
+      - name: Build with Maven
+        run: mvn clean install
+
+      - name: Build Docker image
+        run: |
+          docker build -t streampipes_pipeline-elements-all-jvm:release-validation ./streampipes-extensions/streampipes-extensions-all-jvm
+          docker build -t streampipes_backend:release-validation ./streampipes-service-core
+
+      - name: Start streampipes
+        run: docker compose -f ./streampipes-client-e2e/docker-compose.yml -p streampipes-client-e2e up -d
+
+      - name: Start go client test
+        run: |
+          cd ./streampipes-client-e2e/tool
+          chmod +x ./start-streampipes-client-e2e.sh
+          ./start-streampipes-client-e2e.sh -t go-client-e2e.sh
\ No newline at end of file
diff --git a/streampipes-client-e2e/README.md b/streampipes-client-e2e/README.md
new file mode 100644
index 0000000000..dcd05bc21b
--- /dev/null
+++ b/streampipes-client-e2e/README.md
@@ -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.
+  ~
+  -->
+
+# Client E2E
+
+## Environment Setup
+Before running, ensure that StreamPipe is operational and the corresponding configuration is input into the `start-streampipes-client-e2e.sh` script.
+**Note**: This script must be executed from within the [tool](./tool) directory.
+```shell
+./start-streampipes-client-e2e.sh -h 127.0.0.1 -p 8030 -u admin@streampipes.apache.org   -pw admin -t go-client-e2e.sh
+```
+
+## Usage Instructions
+| Parameter | Default          | Required | Description                                                                                               |
+|-----------|-------------------|----------|-----------------------------------------------------------------------------------------------------------|
+| `-h`      | `127.0.0.1`       | No       | Host address                                                                                              |
+| `-p`      | `8030`            | No       | Backend port                                                                                              |
+| `-u`      | `admin@streampipes.apache.org` | No       | User                                                                                                      |
+| `-pw`     | `admin`           | No       | Password                                                                                                  |
+| `-t`      | `""`              | Yes      | Name of the client's E2E test startup script (the script must be located in the [tool](./tool) directory) |
+
+## How to add E2E in a new language
+1. If you need to define an E2E test in a new language, you need to download the element you need to use in `install-element.sh`.
+2. To accommodate different languages, you need to write an E2E startup script. You can refer to `go-client-e2e.sh`. The startup script will receive four parameters, so it can easily obtain the information needed to run the Client.
+
+| Parameter | Description |
+|-----------|-------------|
+| `-h`      | Host        |
+| `-p`      | Backend port|
+| `-u`      | User        |
+| `-k`      | API Key     |
\ No newline at end of file
diff --git a/streampipes-client-e2e/docker-compose.yml b/streampipes-client-e2e/docker-compose.yml
new file mode 100644
index 0000000000..09e2c9baa5
--- /dev/null
+++ b/streampipes-client-e2e/docker-compose.yml
@@ -0,0 +1,135 @@
+# 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.
+
+version: "3.8"
+
+# global logging
+x-logging: &default-logging
+  options:
+    max-size: "12m"
+    max-file: "5"
+  driver: json-file
+
+services:
+  #### apache/streampipes
+  backend:
+    image: streampipes_backend:release-validation
+    ports:
+      - "8030:8030"
+    depends_on:
+      - couchdb
+    volumes:
+      - backend:/root/.streampipes
+      - files:/spImages
+    logging: *default-logging
+    networks:
+      spnet:
+
+  extensions-all-jvm:
+    image: streampipes_pipeline-elements-all-jvm:release-validation
+    volumes:
+      - files:/spImages
+    logging: *default-logging
+    networks:
+      spnet:
+
+  couchdb:
+    image: couchdb:2.3.1
+    environment:
+      - COUCHDB_USER=admin
+      - COUCHDB_PASSWORD=admin
+    logging: *default-logging
+    volumes:
+      - couchdb:/opt/couchdb/data
+    networks:
+      spnet:
+
+  kafka:
+    image: fogsyio/kafka:2.2.0
+    hostname: kafka
+    depends_on:
+      - zookeeper
+    environment:
+      # see: https://github.com/confluentinc/schema-registry/issues/648
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://:9092
+      KAFKA_LISTENERS: PLAINTEXT://:9092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_ADVERTISED_HOST_NAME: kafka
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+    volumes:
+      - kafka:/kafka
+      - /var/run/docker.sock:/var/run/docker.sock
+    logging: *default-logging
+    networks:
+      spnet:
+
+  zookeeper:
+    image: fogsyio/zookeeper:3.4.13
+    logging: *default-logging
+    volumes:
+      - zookeeper:/opt/zookeeper-3.4.13
+    networks:
+      spnet:
+
+  influxdb:
+    image: influxdb:2.6
+    environment:
+      - INFLUXDB_DATA_ENGINE=tsm1
+      - INFLUXDB_REPORTING_DISABLED=false
+      - INFLUXDB_ADMIN_ENABLED=true
+      - DOCKER_INFLUXDB_INIT_USERNAME=admin
+      - DOCKER_INFLUXDB_INIT_PASSWORD=sp-admin
+      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=sp-admin
+      - DOCKER_INFLUXDB_INIT_ORG=sp
+      - DOCKER_INFLUXDB_INIT_BUCKET=sp
+      - DOCKER_INFLUXDB_INIT_MODE=setup
+    volumes:
+      - influxdb:/var/lib/influxdb
+      - influxdb2:/var/lib/influxdb2
+    logging: *default-logging
+    networks:
+      spnet:
+
+  mosquitto:
+    image: eclipse-mosquitto:1.5.4
+    logging: *default-logging
+    networks:
+      spnet:
+
+  opcua:
+    image: mcr.microsoft.com/iotedge/opc-plc
+    logging: *default-logging
+    command: --ut=true
+    networks:
+      spnet:
+
+volumes:
+  kafka:
+  files:
+  couchdb:
+  zookeeper:
+  influxdb:
+  influxdb2:
+  backend:
+  nginx:
+
+networks:
+  spnet:
+    name: spnet
+    driver: bridge
+    ipam:
+      config:
+        - subnet: 172.31.0.0/16
diff --git a/streampipes-client-e2e/go-client-e2e/adapter/machine.json b/streampipes-client-e2e/go-client-e2e/adapter/machine.json
new file mode 100644
index 0000000000..d50a60be01
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/adapter/machine.json
@@ -0,0 +1,225 @@
+{
+  "@class": "org.apache.streampipes.model.connect.adapter.AdapterDescription",
+  "appId": "org.apache.streampipes.connect.iiot.adapters.simulator.machine",
+  "connectedTo": null,
+  "description": "",
+  "dom": null,
+  "elementId": "sp:org.apache.streampipes.connect.iiot.adapters.simulator.machine",
+  "includedAssets": [
+    "documentation.md",
+    "icon.png"
+  ],
+  "includedLocales": [
+    "strings.en"
+  ],
+  "includesAssets": true,
+  "includesLocales": true,
+  "internallyManaged": false,
+  "name": "test1",
+  "rev": "1-73ca816b2ed82d8232b864556c7f2d64",
+  "version": 0,
+  "category": [
+    "Debugging"
+  ],
+  "config": [
+    {
+      "@class": "org.apache.streampipes.model.staticproperty.FreeTextStaticProperty",
+      "description": "The time to wait between two events in milliseconds",
+      "internalName": "wait-time-ms",
+      "label": "Wait Time (MS)",
+      "optional": false,
+      "staticPropertyType": "FreeTextStaticProperty",
+      "htmlAllowed": false,
+      "htmlFontFormat": false,
+      "mapsTo": null,
+      "multiLine": false,
+      "placeholdersSupported": false,
+      "requiredDatatype": "http://www.w3.org/2001/XMLSchema#integer",
+      "requiredDomainProperty": null,
+      "value": "1000",
+      "valueSpecification": null
+    },
+    {
+      "@class": "org.apache.streampipes.model.staticproperty.OneOfStaticProperty",
+      "description": "Select simulated sensor data to be published",
+      "internalName": "selected-simulator-option",
+      "label": "Select sensor",
+      "optional": false,
+      "staticPropertyType": "OneOfStaticProperty",
+      "horizontalRendering": false,
+      "options": [
+        {
+          "elementId": "sp:option:pOPYGJ",
+          "internalName": null,
+          "name": "flowrate",
+          "selected": true
+        },
+        {
+          "elementId": "sp:option:CpeHtP",
+          "internalName": null,
+          "name": "pressure",
+          "selected": false
+        },
+        {
+          "elementId": "sp:option:KYnhkL",
+          "internalName": null,
+          "name": "waterlevel",
+          "selected": false
+        }
+      ]
+    }
+  ],
+  "correspondingDataStreamElementId": null,
+  "correspondingServiceGroup": null,
+  "createdAt": 0,
+  "dataStream": {
+    "@class": "org.apache.streampipes.model.SpDataStream",
+    "appId": "urn:streampipes.apache.org:eventstream:hLNLBx",
+    "connectedTo": null,
+    "description": null,
+    "dom": null,
+    "elementId": "urn:streampipes.apache.org:eventstream:hLNLBx",
+    "includedAssets": [],
+    "includedLocales": [],
+    "includesAssets": false,
+    "includesLocales": false,
+    "internallyManaged": false,
+    "name": null,
+    "rev": null,
+    "category": null,
+    "correspondingAdapterId": null,
+    "eventGrounding": null,
+    "eventSchema": {
+      "eventProperties": [
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "Denotes the current density of the fluid",
+          "elementId": "sp:eventproperty:Ruvqlo",
+          "label": "Density",
+          "propertyScope": "MEASUREMENT_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "density",
+          "semanticType": "http://schema.org/Number",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+          "valueSpecification": null,
+          "id": 1510309651547
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "Denotes the current mass flow in the sensor",
+          "elementId": "sp:eventproperty:NyTjgp",
+          "label": "Mass Flow",
+          "propertyScope": "MEASUREMENT_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "mass_flow",
+          "semanticType": "http://schema.org/Number",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+          "valueSpecification": null,
+          "id": 4855204620884
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "The ID of the sensor",
+          "elementId": "sp:eventproperty:ThjvdL",
+          "label": "Sensor ID",
+          "propertyScope": "DIMENSION_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "sensorId",
+          "semanticType": "https://streampipes.org/vocabulary/examples/watertank/v1/hasSensorId",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#string",
+          "valueSpecification": null,
+          "id": 3260657832066
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "Any fault flags of the sensors",
+          "elementId": "sp:eventproperty:vxokoH",
+          "label": "Sensor Fault Flags",
+          "propertyScope": "MEASUREMENT_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "sensor_fault_flags",
+          "semanticType": "http://schema.org/Boolean",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+          "valueSpecification": null,
+          "id": 6931662104350
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "Denotes the current temperature in degrees celsius",
+          "elementId": "sp:eventproperty:cmnXbe",
+          "label": "Temperature",
+          "propertyScope": "MEASUREMENT_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "temperature",
+          "semanticType": "http://schema.org/Number",
+          "measurementUnit": "http://qudt.org/vocab/unit#DegreeCelsius",
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+          "valueSpecification": {
+            "@class": "org.apache.streampipes.model.schema.QuantitativeValue",
+            "maxValue": 100,
+            "minValue": 0,
+            "step": 0.1
+          },
+          "id": 1980836171418
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "The current timestamp value",
+          "elementId": "sp:eventproperty:AYjMkj",
+          "label": "Timestamp",
+          "propertyScope": "HEADER_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "timestamp",
+          "semanticType": "http://schema.org/DateTime",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#long",
+          "valueSpecification": null,
+          "id": 5857972009985
+        },
+        {
+          "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+          "additionalMetadata": {},
+          "description": "Denotes the current volume flow",
+          "elementId": "sp:eventproperty:ORzRsw",
+          "label": "Volume Flow",
+          "propertyScope": "MEASUREMENT_PROPERTY",
+          "runtimeId": null,
+          "runtimeName": "volume_flow",
+          "semanticType": "http://schema.org/Number",
+          "measurementUnit": null,
+          "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+          "valueSpecification": null,
+          "id": 64305844251
+        }
+      ]
+    },
+    "index": 0
+  },
+  "deploymentConfiguration": {
+    "desiredServiceTags": [],
+    "selectedEndpointUrl": null
+  },
+  "eventGrounding": {
+    "transportProtocols": []
+  },
+  "eventSchema": {
+    "eventProperties": []
+  },
+  "icon": null,
+  "rules": [],
+  "running": false,
+  "schemaRules": [],
+  "selectedEndpointUrl": null,
+  "streamRules": [],
+  "valueRules": []
+}
\ No newline at end of file
diff --git a/streampipes-client-e2e/go-client-e2e/adapter_test.go b/streampipes-client-e2e/go-client-e2e/adapter_test.go
new file mode 100644
index 0000000000..88d41fcca2
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/adapter_test.go
@@ -0,0 +1,134 @@
+//
+// 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 go_client_e2e
+
+import (
+	"go-client-e2e/utils"
+	"os"
+	"testing"
+)
+
+const (
+	E2E_ADAPTER_ID             = "e2e-adapter-id"
+	E2E_ADAPTER_NAME           = "e2e-adapter-name"
+	E2E_STREAM_REV             = "e2e-stream-rev"
+	E2E_HOST_NAME              = "e2e-host-name"
+	E2E_ADAPTER_OUT_TOPIC_NAME = "e2e-adapter-out-topic-name"
+	E2E_PORT                   = "\"e2e-port\""
+)
+
+func TestCreateAdapter(t *testing.T) {
+	TestDeleteAdapter(t)
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Log(err)
+		os.Exit(1)
+	}
+	data := utils.CreateData("adapter/machine.json")
+	err = streamPipesClient.Adapter().CreateAdapter(data)
+	if err != nil {
+		t.Log(err)
+		os.Exit(1)
+	}
+}
+
+func TestGetAdapter(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	adapters, err1 := streamPipesClient.Adapter().GetAllAdapter()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	if len(adapters) == 0 {
+		t.Log("adapter is null")
+		os.Exit(1)
+	}
+}
+
+func TestStartAdapter(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	adapters, err1 := streamPipesClient.Adapter().GetAllAdapter()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	if len(adapters) == 0 {
+		t.Log("adapter is null")
+		os.Exit(1)
+	}
+	err = streamPipesClient.Adapter().StartSingleAdapter(adapters[0].ElementID)
+	if err != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+}
+
+func TestStopAdapter(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	adapters, err1 := streamPipesClient.Adapter().GetAllAdapter()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	for _, adapter := range adapters {
+		err = streamPipesClient.Adapter().StopSingleAdapter(adapter.ElementID)
+		if err != nil {
+			t.Error(err)
+			os.Exit(1)
+		}
+		adapter, err = streamPipesClient.Adapter().GetSingleAdapter(adapter.ElementID)
+		if err != nil || adapter.Running {
+			t.Error(err)
+			os.Exit(1)
+		}
+	}
+
+}
+
+func TestDeleteAdapter(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	adapters, err1 := streamPipesClient.Adapter().GetAllAdapter()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	for _, adapter := range adapters {
+		err = streamPipesClient.Adapter().DeleteSingleAdapter(adapter.ElementID)
+		if err != nil {
+			t.Error(err1)
+			os.Exit(1)
+		}
+	}
+
+}
diff --git a/streampipes-client-e2e/go-client-e2e/datalake_test.go b/streampipes-client-e2e/go-client-e2e/datalake_test.go
new file mode 100644
index 0000000000..6168139ed6
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/datalake_test.go
@@ -0,0 +1,52 @@
+//
+// 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 go_client_e2e
+
+import (
+	"go-client-e2e/utils"
+	"os"
+	"testing"
+	"time"
+)
+
+func TestGetDataLake(t *testing.T) {
+	TestCreatePipeline(t)
+	TestStartPipeline(t)
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	dataLake := streamPipesClient.DataLakeMeasures()
+	measures, err := dataLake.GetAllDataLakeMeasure()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+
+	time.Sleep(10 * time.Second)
+
+	if len(measures) == 0 {
+		TestStopPipeline(t)
+		TestDeletePipeline(t)
+		t.Error("No data lake measures found")
+		os.Exit(1)
+	}
+	TestStopPipeline(t)
+	TestDeletePipeline(t)
+}
diff --git a/streampipes-client-e2e/go-client-e2e/go.mod b/streampipes-client-e2e/go-client-e2e/go.mod
new file mode 100644
index 0000000000..f7c1cd0cc1
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/go.mod
@@ -0,0 +1,23 @@
+//
+// 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.
+//
+
+module go-client-e2e
+
+go 1.21.6
+
+require github.com/apache/streampipes/streampipes-client-go v0.0.0 // indirect
+replace "github.com/apache/streampipes/streampipes-client-go" => "../../streampipes-client-go"
\ No newline at end of file
diff --git a/streampipes-client-e2e/go-client-e2e/pipeline_test.go b/streampipes-client-e2e/go-client-e2e/pipeline_test.go
new file mode 100644
index 0000000000..e85fc375fe
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/pipeline_test.go
@@ -0,0 +1,138 @@
+//
+// 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 go_client_e2e
+
+import (
+	"go-client-e2e/utils"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestCreatePipeline(t *testing.T) {
+	TestDeletePipeline(t)
+	TestCreateAdapter(t)
+
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		os.Exit(1)
+	}
+	TestStartAdapter(t)
+	adapters, err := streamPipesClient.Adapter().GetAllAdapter()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	if len(adapters) == 0 {
+		t.Log("adapter is null")
+		os.Exit(1)
+	}
+	adapter := adapters[0]
+
+	data := utils.CreateData("pipelines/pipelines.json")
+	pipeline := string(data)
+	pipeline = strings.Replace(pipeline, E2E_ADAPTER_ID, adapter.ElementID, -1)
+	pipeline = strings.Replace(pipeline, E2E_ADAPTER_NAME, adapter.Name, -1)
+	pipeline = strings.Replace(pipeline, E2E_STREAM_REV, adapter.Rev, -1)
+	pipeline = strings.Replace(pipeline, E2E_ADAPTER_OUT_TOPIC_NAME, adapter.EventGrounding.TransportProtocols[0].TopicDefinition.ActualTopicName, -1)
+
+	message, statusErr := streamPipesClient.Pipeline().CreatePipeline([]byte(pipeline))
+	if statusErr != nil || !message.Success {
+		t.Log(statusErr, !message.Success)
+		os.Exit(1)
+	}
+
+}
+
+func TestGetPipeline(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	pipelines, err1 := streamPipesClient.Pipeline().GetAllPipeline()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	if len(pipelines) == 0 {
+		t.Log("pipelines is null")
+		os.Exit(1)
+	}
+}
+
+func TestStartPipeline(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	pipelines, err1 := streamPipesClient.Pipeline().GetAllPipeline()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	if len(pipelines) == 0 {
+		t.Log("pipelines is null")
+		os.Exit(1)
+	}
+
+	status, statusErr := streamPipesClient.Pipeline().StartSinglePipeline(pipelines[0].PipelineId)
+	if statusErr != nil || !status.Success {
+		t.Error(status.Title)
+		os.Exit(1)
+	}
+}
+
+func TestStopPipeline(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	pipelines, err1 := streamPipesClient.Pipeline().GetAllPipeline()
+	for _, pipeline := range pipelines {
+		operation, operationErr := streamPipesClient.Pipeline().StopSinglePipeline(pipeline.PipelineId)
+		if operationErr != nil || !operation.Success {
+			t.Error(err1)
+			os.Exit(1)
+		}
+	}
+
+}
+
+func TestDeletePipeline(t *testing.T) {
+	streamPipesClient, err := utils.CreateStreamPipesClient()
+	if err != nil {
+		t.Error(err)
+		os.Exit(1)
+	}
+	pipelines, err1 := streamPipesClient.Pipeline().GetAllPipeline()
+	if err1 != nil {
+		t.Error(err1)
+		os.Exit(1)
+	}
+	for _, pipeline := range pipelines {
+		err = streamPipesClient.Pipeline().DeleteSinglePipeline(pipeline.PipelineId)
+		if err != nil {
+			t.Error(err1)
+			os.Exit(1)
+		}
+	}
+	TestDeleteAdapter(t)
+}
diff --git a/streampipes-client-e2e/go-client-e2e/pipelines/pipelines.json b/streampipes-client-e2e/go-client-e2e/pipelines/pipelines.json
new file mode 100644
index 0000000000..3cebc73a5f
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/pipelines/pipelines.json
@@ -0,0 +1,960 @@
+{
+  "actions": [
+    {
+      "@class": "org.apache.streampipes.model.graph.DataSinkInvocation",
+      "appId": "org.apache.streampipes.sinks.internal.jvm.datalake",
+      "connectedTo": [
+        "jsplumb_1_j3BB"
+      ],
+      "description": "Stores events in the internal data lake.",
+      "dom": "jsplumb_2_Fahe",
+      "elementId": "sp:datasinkinvocation:MZnKuj:qULqX",
+      "includedAssets": [
+        "documentation.md",
+        "icon.png"
+      ],
+      "includedLocales": [
+        "strings.en"
+      ],
+      "includesAssets": true,
+      "includesLocales": true,
+      "internallyManaged": false,
+      "name": "Data Lake",
+      "rev": null,
+      "version": 2,
+      "belongsTo": "sp:org.apache.streampipes.sinks.internal.jvm.datalake",
+      "configured": true,
+      "correspondingPipeline": null,
+      "correspondingUser": null,
+      "inputStreams": [
+        {
+          "@class": "org.apache.streampipes.model.SpDataStream",
+          "appId": "urn:streampipes.apache.org:eventstream:mqYwaw",
+          "connectedTo": null,
+          "description": null,
+          "dom": null,
+          "elementId": "urn:streampipes.apache.org:eventstream:mqYwaw",
+          "includedAssets": [],
+          "includedLocales": [],
+          "includesAssets": false,
+          "includesLocales": false,
+          "internallyManaged": false,
+          "name": null,
+          "rev": null,
+          "category": null,
+          "correspondingAdapterId": null,
+          "eventGrounding": {
+            "transportProtocols": [
+              {
+                "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+                "brokerHostname": "kafka",
+                "elementId": null,
+                "topicDefinition": {
+                  "@class": "org.apache.streampipes.model.grounding.SimpleTopicDefinition",
+                  "actualTopicName": "org-apache-streampipes-internal-prbfXSLBvVHpoeX"
+                },
+                "acks": null,
+                "batchSize": null,
+                "groupId": null,
+                "kafkaPort": 9092,
+                "lingerMs": null,
+                "maxRequestSize": null,
+                "messageMaxBytes": null,
+                "offset": null,
+                "zookeeperHost": "zookeeper",
+                "zookeeperPort": 2181
+              }
+            ]
+          },
+          "eventSchema": {
+            "eventProperties": [
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current density of the fluid",
+                "elementId": "sp:eventproperty:Ruvqlo",
+                "label": "Density",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "density",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current mass flow in the sensor",
+                "elementId": "sp:eventproperty:NyTjgp",
+                "label": "Mass Flow",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "mass_flow",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Any fault flags of the sensors",
+                "elementId": "sp:eventproperty:vxokoH",
+                "label": "Sensor Fault Flags",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "sensor_fault_flags",
+                "semanticType": "http://schema.org/Boolean",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "The ID of the sensor",
+                "elementId": "sp:eventproperty:ThjvdL",
+                "label": "Sensor ID",
+                "propertyScope": "DIMENSION_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "sensorId",
+                "semanticType": "https://streampipes.org/vocabulary/examples/watertank/v1/hasSensorId",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#string",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current temperature in degrees celsius",
+                "elementId": "sp:eventproperty:cmnXbe",
+                "label": "Temperature",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "temperature",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": "http://qudt.org/vocab/unit#DegreeCelsius",
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": {
+                  "@class": "org.apache.streampipes.model.schema.QuantitativeValue",
+                  "maxValue": 100,
+                  "minValue": 0,
+                  "step": 0.1
+                }
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "The current timestamp value",
+                "elementId": "sp:eventproperty:AYjMkj",
+                "label": "Timestamp",
+                "propertyScope": "HEADER_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "timestamp",
+                "semanticType": "http://schema.org/DateTime",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#long",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current volume flow",
+                "elementId": "sp:eventproperty:ORzRsw",
+                "label": "Volume Flow",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "volume_flow",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              }
+            ]
+          },
+          "index": 0
+        }
+      ],
+      "selectedEndpointUrl": null,
+      "serviceTagPrefix": "DATA_SINK",
+      "staticProperties": [
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.MappingPropertyUnary",
+          "description": "The value which contains a timestamp",
+          "internalName": "timestamp_mapping",
+          "label": "Timestamp Field",
+          "optional": false,
+          "staticPropertyType": "MappingPropertyUnary",
+          "mapsFromOptions": [
+            "s0::timestamp"
+          ],
+          "propertyScope": "NONE",
+          "requirementSelector": "r0::timestamp_mapping",
+          "selectedProperty": "s0::timestamp"
+        },
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.FreeTextStaticProperty",
+          "description": "The name of the identifier under which the data is to be stored.",
+          "internalName": "db_measurement",
+          "label": "Identifier",
+          "optional": false,
+          "staticPropertyType": "FreeTextStaticProperty",
+          "htmlAllowed": false,
+          "htmlFontFormat": false,
+          "mapsTo": null,
+          "multiLine": false,
+          "placeholdersSupported": false,
+          "requiredDatatype": "http://www.w3.org/2001/XMLSchema#string",
+          "requiredDomainProperty": null,
+          "value": "test",
+          "valueSpecification": null
+        },
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.OneOfStaticProperty",
+          "description": "Update existing schemas with the new one or extend the existing schema with new properties",
+          "internalName": "schema_update",
+          "label": "Schema Update",
+          "optional": false,
+          "staticPropertyType": "OneOfStaticProperty",
+          "horizontalRendering": false,
+          "options": [
+            {
+              "elementId": "sp:option:wzqPbk",
+              "internalName": null,
+              "name": "Update schema",
+              "selected": true
+            },
+            {
+              "elementId": "sp:option:XmVjRp",
+              "internalName": null,
+              "name": "Extend existing schema",
+              "selected": false
+            }
+          ]
+        },
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.RuntimeResolvableAnyStaticProperty",
+          "description": "Selected fields will be stored as dimensions.",
+          "internalName": "dimensions_selection",
+          "label": "Dimensions",
+          "optional": false,
+          "staticPropertyType": "RuntimeResolvableAnyStaticProperty",
+          "horizontalRendering": false,
+          "options": [
+            {
+              "elementId": "sp:option:paBEDz",
+              "internalName": null,
+              "name": "sensor_fault_flags",
+              "selected": false
+            },
+            {
+              "elementId": "sp:option:JYOYId",
+              "internalName": null,
+              "name": "sensorId",
+              "selected": true
+            }
+          ],
+          "dependsOn": null
+        },
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty",
+          "description": "Fields having the same value than the previous event are not stored. This only affects measurements, not tags.",
+          "internalName": "ignore_duplicates",
+          "label": "Ignore duplicates",
+          "optional": false,
+          "staticPropertyType": "SlideToggleStaticProperty",
+          "defaultValue": false,
+          "selected": false
+        }
+      ],
+      "streamRequirements": [
+        {
+          "@class": "org.apache.streampipes.model.SpDataStream",
+          "appId": "urn:streampipes.apache.org:eventstream:mqYwaw",
+          "connectedTo": null,
+          "description": null,
+          "dom": null,
+          "elementId": "urn:streampipes.apache.org:eventstream:mqYwaw",
+          "includedAssets": [],
+          "includedLocales": [],
+          "includesAssets": false,
+          "includesLocales": false,
+          "internallyManaged": false,
+          "name": null,
+          "rev": null,
+          "category": null,
+          "correspondingAdapterId": null,
+          "eventGrounding": null,
+          "eventSchema": {
+            "eventProperties": [
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": null,
+                "elementId": "sp:eventproperty:LrdSOF",
+                "label": null,
+                "propertyScope": null,
+                "runtimeId": null,
+                "runtimeName": "timestamp_mapping",
+                "semanticType": "http://schema.org/DateTime",
+                "measurementUnit": null,
+                "runtimeType": null,
+                "valueSpecification": null
+              }
+            ]
+          },
+          "index": 0
+        }
+      ],
+      "supportedGrounding": {
+        "transportProtocols": [
+          {
+            "@class": "org.apache.streampipes.model.grounding.MqttTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:IxejrD",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.NatsTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:hMYAXY",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.JmsTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:ycSaOT",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.PulsarTransportProtocol",
+            "brokerHostname": "pulsar://localhost:6650",
+            "elementId": "sp:transportprotocol:lvMQWc",
+            "topicDefinition": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:EiyDHT",
+            "topicDefinition": null,
+            "acks": null,
+            "batchSize": null,
+            "groupId": null,
+            "kafkaPort": 0,
+            "lingerMs": null,
+            "maxRequestSize": null,
+            "messageMaxBytes": null,
+            "offset": null,
+            "zookeeperHost": null,
+            "zookeeperPort": 0
+          }
+        ]
+      },
+      "uncompleted": false,
+      "category": [
+        "INTERNAL"
+      ]
+    }
+  ],
+  "description": "",
+  "name": "test",
+  "sepas": [
+    {
+      "@class": "org.apache.streampipes.model.graph.DataProcessorInvocation",
+      "appId": "org.apache.streampipes.processors.transformation.jvm.booloperator.inverter",
+      "connectedTo": [
+        "jsplumb_0_x6Gz"
+      ],
+      "description": "Inverts the boolean value of the selected field",
+      "dom": "jsplumb_1_j3BB",
+      "elementId": "sp:dataprocessorinvocation:JOqFCM:R0okK",
+      "includedAssets": [
+        "documentation.md",
+        "icon.png"
+      ],
+      "includedLocales": [
+        "strings.en"
+      ],
+      "includesAssets": true,
+      "includesLocales": true,
+      "internallyManaged": false,
+      "name": "Boolean Inverter",
+      "rev": null,
+      "version": 0,
+      "belongsTo": "sp:org.apache.streampipes.processors.transformation.jvm.booloperator.inverter",
+      "configured": true,
+      "correspondingPipeline": null,
+      "correspondingUser": null,
+      "inputStreams": [
+        {
+          "@class": "org.apache.streampipes.model.SpDataStream",
+          "appId": "urn:streampipes.apache.org:eventstream:VPyWtS",
+          "connectedTo": null,
+          "description": null,
+          "dom": null,
+          "elementId": "urn:streampipes.apache.org:eventstream:VPyWtS",
+          "includedAssets": [],
+          "includedLocales": [],
+          "includesAssets": false,
+          "includesLocales": false,
+          "internallyManaged": false,
+          "name": null,
+          "rev": null,
+          "category": null,
+          "correspondingAdapterId": null,
+          "eventGrounding": {
+            "transportProtocols": [
+              {
+                "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+                "brokerHostname": "kafka",
+                "elementId": "sp:transportprotocol:KoHuzT",
+                "topicDefinition": {
+                  "@class": "org.apache.streampipes.model.grounding.SimpleTopicDefinition",
+                  "actualTopicName": "e2e-adapter-out-topic-name"
+                },
+                "acks": null,
+                "batchSize": null,
+                "groupId": null,
+                "kafkaPort": 9092,
+                "lingerMs": null,
+                "maxRequestSize": null,
+                "messageMaxBytes": null,
+                "offset": null,
+                "zookeeperHost": null,
+                "zookeeperPort": 0
+              }
+            ]
+          },
+          "eventSchema": {
+            "eventProperties": [
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current density of the fluid",
+                "elementId": "sp:eventproperty:Ruvqlo",
+                "label": "Density",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "density",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current mass flow in the sensor",
+                "elementId": "sp:eventproperty:NyTjgp",
+                "label": "Mass Flow",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "mass_flow",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "The ID of the sensor",
+                "elementId": "sp:eventproperty:ThjvdL",
+                "label": "Sensor ID",
+                "propertyScope": "DIMENSION_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "sensorId",
+                "semanticType": "https://streampipes.org/vocabulary/examples/watertank/v1/hasSensorId",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#string",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Any fault flags of the sensors",
+                "elementId": "sp:eventproperty:vxokoH",
+                "label": "Sensor Fault Flags",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "sensor_fault_flags",
+                "semanticType": "http://schema.org/Boolean",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current temperature in degrees celsius",
+                "elementId": "sp:eventproperty:cmnXbe",
+                "label": "Temperature",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "temperature",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": "http://qudt.org/vocab/unit#DegreeCelsius",
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": {
+                  "@class": "org.apache.streampipes.model.schema.QuantitativeValue",
+                  "maxValue": 100,
+                  "minValue": 0,
+                  "step": 0.1
+                }
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "The current timestamp value",
+                "elementId": "sp:eventproperty:AYjMkj",
+                "label": "Timestamp",
+                "propertyScope": "HEADER_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "timestamp",
+                "semanticType": "http://schema.org/DateTime",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#long",
+                "valueSpecification": null
+              },
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": "Denotes the current volume flow",
+                "elementId": "sp:eventproperty:ORzRsw",
+                "label": "Volume Flow",
+                "propertyScope": "MEASUREMENT_PROPERTY",
+                "runtimeId": null,
+                "runtimeName": "volume_flow",
+                "semanticType": "http://schema.org/Number",
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+                "valueSpecification": null
+              }
+            ]
+          },
+          "index": 0
+        }
+      ],
+      "selectedEndpointUrl": null,
+      "serviceTagPrefix": "DATA_PROCESSOR",
+      "staticProperties": [
+        {
+          "@class": "org.apache.streampipes.model.staticproperty.MappingPropertyUnary",
+          "description": "The field which should be inverted",
+          "internalName": "invert-field",
+          "label": "Invert field",
+          "optional": false,
+          "staticPropertyType": "MappingPropertyUnary",
+          "mapsFromOptions": [
+            "s0::sensor_fault_flags"
+          ],
+          "propertyScope": "NONE",
+          "requirementSelector": "r0::invert-field",
+          "selectedProperty": "s0::sensor_fault_flags"
+        }
+      ],
+      "streamRequirements": [
+        {
+          "@class": "org.apache.streampipes.model.SpDataStream",
+          "appId": "urn:streampipes.apache.org:eventstream:VPyWtS",
+          "connectedTo": null,
+          "description": null,
+          "dom": null,
+          "elementId": "urn:streampipes.apache.org:eventstream:VPyWtS",
+          "includedAssets": [],
+          "includedLocales": [],
+          "includesAssets": false,
+          "includesLocales": false,
+          "internallyManaged": false,
+          "name": null,
+          "rev": null,
+          "category": null,
+          "correspondingAdapterId": null,
+          "eventGrounding": null,
+          "eventSchema": {
+            "eventProperties": [
+              {
+                "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+                "additionalMetadata": {},
+                "description": null,
+                "elementId": "sp:eventproperty:sHuzzM",
+                "label": null,
+                "propertyScope": null,
+                "runtimeId": null,
+                "runtimeName": "invert-field",
+                "semanticType": null,
+                "measurementUnit": null,
+                "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+                "valueSpecification": null
+              }
+            ]
+          },
+          "index": 0
+        }
+      ],
+      "supportedGrounding": {
+        "transportProtocols": [
+          {
+            "@class": "org.apache.streampipes.model.grounding.MqttTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:IxejrD",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.NatsTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:hMYAXY",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.JmsTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:ycSaOT",
+            "topicDefinition": null,
+            "port": 0
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.PulsarTransportProtocol",
+            "brokerHostname": "pulsar://localhost:6650",
+            "elementId": "sp:transportprotocol:lvMQWc",
+            "topicDefinition": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+            "brokerHostname": null,
+            "elementId": "sp:transportprotocol:EiyDHT",
+            "topicDefinition": null,
+            "acks": null,
+            "batchSize": null,
+            "groupId": null,
+            "kafkaPort": 0,
+            "lingerMs": null,
+            "maxRequestSize": null,
+            "messageMaxBytes": null,
+            "offset": null,
+            "zookeeperHost": null,
+            "zookeeperPort": 0
+          }
+        ]
+      },
+      "uncompleted": false,
+      "category": [
+        "BOOLEAN_OPERATOR"
+      ],
+      "outputStrategies": [
+        {
+          "@class": "org.apache.streampipes.model.output.KeepOutputStrategy",
+          "name": null,
+          "renameRules": [],
+          "eventName": null,
+          "keepBoth": false
+        }
+      ],
+      "outputStream": {
+        "@class": "org.apache.streampipes.model.SpDataStream",
+        "appId": "urn:streampipes.apache.org:eventstream:zPxOLe",
+        "connectedTo": null,
+        "description": null,
+        "dom": null,
+        "elementId": "urn:streampipes.apache.org:eventstream:zPxOLe",
+        "includedAssets": [],
+        "includedLocales": [],
+        "includesAssets": false,
+        "includesLocales": false,
+        "internallyManaged": false,
+        "name": null,
+        "rev": null,
+        "category": null,
+        "correspondingAdapterId": null,
+        "eventGrounding": {
+          "transportProtocols": [
+            {
+              "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+              "brokerHostname": "kafka",
+              "elementId": null,
+              "topicDefinition": {
+                "@class": "org.apache.streampipes.model.grounding.SimpleTopicDefinition",
+                "actualTopicName": "org-apache-streampipes-internal-prbfXSLBvVHpoeX"
+              },
+              "acks": null,
+              "batchSize": null,
+              "groupId": null,
+              "kafkaPort": 9092,
+              "lingerMs": null,
+              "maxRequestSize": null,
+              "messageMaxBytes": null,
+              "offset": null,
+              "zookeeperHost": "zookeeper",
+              "zookeeperPort": 2181
+            }
+          ]
+        },
+        "eventSchema": {
+          "eventProperties": [
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "Denotes the current density of the fluid",
+              "elementId": "sp:eventproperty:Ruvqlo",
+              "label": "Density",
+              "propertyScope": "MEASUREMENT_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "density",
+              "semanticType": "http://schema.org/Number",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+              "valueSpecification": null
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "Denotes the current mass flow in the sensor",
+              "elementId": "sp:eventproperty:NyTjgp",
+              "label": "Mass Flow",
+              "propertyScope": "MEASUREMENT_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "mass_flow",
+              "semanticType": "http://schema.org/Number",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+              "valueSpecification": null
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "The ID of the sensor",
+              "elementId": "sp:eventproperty:ThjvdL",
+              "label": "Sensor ID",
+              "propertyScope": "DIMENSION_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "sensorId",
+              "semanticType": "https://streampipes.org/vocabulary/examples/watertank/v1/hasSensorId",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#string",
+              "valueSpecification": null
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "Any fault flags of the sensors",
+              "elementId": "sp:eventproperty:vxokoH",
+              "label": "Sensor Fault Flags",
+              "propertyScope": "MEASUREMENT_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "sensor_fault_flags",
+              "semanticType": "http://schema.org/Boolean",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+              "valueSpecification": null
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "Denotes the current temperature in degrees celsius",
+              "elementId": "sp:eventproperty:cmnXbe",
+              "label": "Temperature",
+              "propertyScope": "MEASUREMENT_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "temperature",
+              "semanticType": "http://schema.org/Number",
+              "measurementUnit": "http://qudt.org/vocab/unit#DegreeCelsius",
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+              "valueSpecification": {
+                "@class": "org.apache.streampipes.model.schema.QuantitativeValue",
+                "maxValue": 100,
+                "minValue": 0,
+                "step": 0.1
+              }
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "The current timestamp value",
+              "elementId": "sp:eventproperty:AYjMkj",
+              "label": "Timestamp",
+              "propertyScope": "HEADER_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "timestamp",
+              "semanticType": "http://schema.org/DateTime",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#long",
+              "valueSpecification": null
+            },
+            {
+              "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+              "additionalMetadata": {},
+              "description": "Denotes the current volume flow",
+              "elementId": "sp:eventproperty:ORzRsw",
+              "label": "Volume Flow",
+              "propertyScope": "MEASUREMENT_PROPERTY",
+              "runtimeId": null,
+              "runtimeName": "volume_flow",
+              "semanticType": "http://schema.org/Number",
+              "measurementUnit": null,
+              "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+              "valueSpecification": null
+            }
+          ]
+        },
+        "index": 0
+      }
+    }
+  ],
+  "streams": [
+    {
+      "@class": "org.apache.streampipes.model.SpDataStream",
+      "appId": "e2e-adapter-id",
+      "connectedTo": null,
+      "description": "",
+      "dom": "jsplumb_0_x6Gz",
+      "elementId": "e2e-adapter-id",
+      "includedAssets": [],
+      "includedLocales": [],
+      "includesAssets": false,
+      "includesLocales": false,
+      "internallyManaged": true,
+      "name": "e2e-adapter-name",
+      "rev": "e2e-stream-rev",
+      "category": null,
+      "correspondingAdapterId": "sp:adapterdescription:OiTAHK",
+      "eventGrounding": {
+        "transportProtocols": [
+          {
+            "@class": "org.apache.streampipes.model.grounding.KafkaTransportProtocol",
+            "brokerHostname": "kafka",
+            "elementId": "sp:transportprotocol:KoHuzT",
+            "topicDefinition": {
+              "@class": "org.apache.streampipes.model.grounding.SimpleTopicDefinition",
+              "actualTopicName": "e2e-adapter-out-topic-name"
+            },
+            "acks": null,
+            "batchSize": null,
+            "groupId": null,
+            "kafkaPort": 9092,
+            "lingerMs": null,
+            "maxRequestSize": null,
+            "messageMaxBytes": null,
+            "offset": null,
+            "zookeeperHost": null,
+            "zookeeperPort": 0
+          }
+        ]
+      },
+      "eventSchema": {
+        "eventProperties": [
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "Denotes the current density of the fluid",
+            "elementId": "sp:eventproperty:Ruvqlo",
+            "label": "Density",
+            "propertyScope": "MEASUREMENT_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "density",
+            "semanticType": "http://schema.org/Number",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+            "valueSpecification": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "Denotes the current mass flow in the sensor",
+            "elementId": "sp:eventproperty:NyTjgp",
+            "label": "Mass Flow",
+            "propertyScope": "MEASUREMENT_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "mass_flow",
+            "semanticType": "http://schema.org/Number",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+            "valueSpecification": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "The ID of the sensor",
+            "elementId": "sp:eventproperty:ThjvdL",
+            "label": "Sensor ID",
+            "propertyScope": "DIMENSION_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "sensorId",
+            "semanticType": "https://streampipes.org/vocabulary/examples/watertank/v1/hasSensorId",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#string",
+            "valueSpecification": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "Any fault flags of the sensors",
+            "elementId": "sp:eventproperty:vxokoH",
+            "label": "Sensor Fault Flags",
+            "propertyScope": "MEASUREMENT_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "sensor_fault_flags",
+            "semanticType": "http://schema.org/Boolean",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#boolean",
+            "valueSpecification": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "Denotes the current temperature in degrees celsius",
+            "elementId": "sp:eventproperty:cmnXbe",
+            "label": "Temperature",
+            "propertyScope": "MEASUREMENT_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "temperature",
+            "semanticType": "http://schema.org/Number",
+            "measurementUnit": "http://qudt.org/vocab/unit#DegreeCelsius",
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+            "valueSpecification": {
+              "@class": "org.apache.streampipes.model.schema.QuantitativeValue",
+              "maxValue": 100,
+              "minValue": 0,
+              "step": 0.1
+            }
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "The current timestamp value",
+            "elementId": "sp:eventproperty:AYjMkj",
+            "label": "Timestamp",
+            "propertyScope": "HEADER_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "timestamp",
+            "semanticType": "http://schema.org/DateTime",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#long",
+            "valueSpecification": null
+          },
+          {
+            "@class": "org.apache.streampipes.model.schema.EventPropertyPrimitive",
+            "additionalMetadata": {},
+            "description": "Denotes the current volume flow",
+            "elementId": "sp:eventproperty:ORzRsw",
+            "label": "Volume Flow",
+            "propertyScope": "MEASUREMENT_PROPERTY",
+            "runtimeId": null,
+            "runtimeName": "volume_flow",
+            "semanticType": "http://schema.org/Number",
+            "measurementUnit": null,
+            "runtimeType": "http://www.w3.org/2001/XMLSchema#float",
+            "valueSpecification": null
+          }
+        ]
+      },
+      "index": 0
+    }
+  ],
+  "valid": true
+}
\ No newline at end of file
diff --git a/streampipes-client-e2e/go-client-e2e/utils/create_machine_apadter.go b/streampipes-client-e2e/go-client-e2e/utils/create_machine_apadter.go
new file mode 100644
index 0000000000..2f9034c1f1
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/utils/create_machine_apadter.go
@@ -0,0 +1,37 @@
+//
+// 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 utils
+
+import (
+	"io/ioutil"
+	"os"
+)
+
+func CreateData(jsonFilePath string) []byte {
+	jsonFile, err := os.Open(jsonFilePath)
+	if err != nil {
+		return nil
+	}
+	defer jsonFile.Close()
+
+	jsonData, err := ioutil.ReadAll(jsonFile)
+	if err != nil {
+		return nil
+	}
+	return jsonData
+}
diff --git a/streampipes-client-e2e/go-client-e2e/utils/streampipes_client.go b/streampipes-client-e2e/go-client-e2e/utils/streampipes_client.go
new file mode 100644
index 0000000000..90c2b87203
--- /dev/null
+++ b/streampipes-client-e2e/go-client-e2e/utils/streampipes_client.go
@@ -0,0 +1,36 @@
+//
+// 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 utils
+
+import (
+	"github.com/apache/streampipes/streampipes-client-go/streampipes"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+	"os"
+)
+
+func CreateStreamPipesClient() (*streampipes.StreamPipesClient, error) {
+	length := len(os.Args)
+	clientConfig := config.StreamPipesClientConfig{
+		Url: "http://" + os.Args[length-4] + ":" + os.Args[length-3],
+		Credential: config.StreamPipesApiKeyCredentials{
+			UserName: os.Args[length-1],
+			ApiKey:   os.Args[length-2],
+		},
+	}
+	return streampipes.NewStreamPipesClient(clientConfig)
+}
diff --git a/streampipes-client-e2e/tool/go-client-e2e.sh b/streampipes-client-e2e/tool/go-client-e2e.sh
new file mode 100644
index 0000000000..3b52e89e2e
--- /dev/null
+++ b/streampipes-client-e2e/tool/go-client-e2e.sh
@@ -0,0 +1,61 @@
+#!/bin/bash
+
+# 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.
+
+# Set environment variables
+HOST=""
+PORT=""
+APIKEY=""
+API_KEY_USER_NAME=""
+
+while true; do
+    case "$1" in
+        -h)
+            HOST="$2"
+            shift 2
+        ;;
+        -p)
+            PORT="$2"
+            shift 2
+        ;;
+        -u)
+            API_KEY_USER_NAME="$2"
+            shift 2
+        ;;
+        -k)
+            APIKEY="$2"
+            shift 2
+        ;;
+        "")
+            #skip directly
+            break
+        ;;
+        *)
+          #skip directly
+            shift
+        ;;
+    esac
+done
+
+cd ../go-client-e2e || exit
+go test -v ../go-client-e2e/... -args "$HOST" "$PORT" "$APIKEY" "$API_KEY_USER_NAME"
+if [ $? -ne 0 ]; then
+    cd ../tool
+    echo "Error: go test failed"
+    exit 1
+fi
+cd ../tool || exit
+echo "All tests passed successfully"
\ No newline at end of file
diff --git a/streampipes-client-e2e/tool/install-element.sh b/streampipes-client-e2e/tool/install-element.sh
new file mode 100644
index 0000000000..2b3c340f7a
--- /dev/null
+++ b/streampipes-client-e2e/tool/install-element.sh
@@ -0,0 +1,113 @@
+#!/bin/bash
+
+# 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.
+
+
+INSTALL_ELEMENT_URL="/streampipes-backend/api/v2/extension-installation"
+
+HOST="127.0.0.1"
+PORT="8030"
+TOKEN=""
+
+
+while true; do
+    case "$1" in
+        -host)
+            HOST="$2"
+            shift 2
+            ;;
+        -port)
+            PORT="$2"
+            shift 2
+        ;;
+        -token)
+            TOKEN="$2"
+            shift 2
+        ;;
+        --help)
+            echo "Usage: $0 [-host <ip>] [-port <port>] [-token <username>]"
+            exit 0
+        ;;
+        "")
+            #skip directly
+            break
+        ;;
+        *)
+          #skip directly
+            shift
+        ;;
+    esac
+done
+
+
+######################Adapters######################
+
+#machine
+installRequestBody='{
+  "appId":"org.apache.streampipes.connect.iiot.adapters.simulator.machine",
+  "publicElement":true,
+  "serviceTagPrefix":"ADAPTER"
+  }'
+
+response=$(curl -s -X POST "http://$HOST:$PORT$INSTALL_ELEMENT_URL" \
+   -H "Content-Type: application/json" \
+   -H "authorization: Bearer $TOKEN" \
+   -d "$installRequestBody")
+if [ $? -ne 0 ]; then
+    echo "$response"
+    echo "Error install machine adapter"
+    exit 1
+fi
+
+
+
+######################processor######################
+
+#bool_inverter
+installRequestBody='{
+  "appId":"org.apache.streampipes.processors.transformation.jvm.booloperator.inverter",
+  "publicElement":true,
+  "serviceTagPrefix":"DATA_PROCESSOR"
+  }'
+
+response=$(curl -s -X POST "http://$HOST:$PORT$INSTALL_ELEMENT_URL" \
+   -H "Content-Type: application/json" \
+   -H "authorization: Bearer $TOKEN" \
+   -d "$installRequestBody")
+if [ $? -ne 0 ]; then
+    echo "$response"
+    echo "Error install bool inverter processor"
+    exit 1
+fi
+
+
+######################sink######################
+#datalake
+installRequestBody='{
+  "appId":"org.apache.streampipes.sinks.internal.jvm.datalake",
+  "publicElement":true,
+  "serviceTagPrefix":"DATA_SINK"
+  }'
+
+response=$(curl -s -X POST "http://$HOST:$PORT$INSTALL_ELEMENT_URL" \
+   -H "Content-Type: application/json" \
+   -H "authorization: Bearer $TOKEN" \
+   -d "$installRequestBody")
+if [ $? -ne 0 ]; then
+    echo "$response"
+    echo "Error install datalake"
+    exit 1
+fi
\ No newline at end of file
diff --git a/streampipes-client-e2e/tool/start-streampipes-client-e2e.sh b/streampipes-client-e2e/tool/start-streampipes-client-e2e.sh
new file mode 100644
index 0000000000..cbab40355f
--- /dev/null
+++ b/streampipes-client-e2e/tool/start-streampipes-client-e2e.sh
@@ -0,0 +1,148 @@
+#!/bin/bash
+
+# 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.
+
+# Set environment variables
+E2E_TEST=""
+HOST="127.0.0.1"
+PORT="8030"
+LOGIN_URL="/streampipes-backend/api/v2/auth/login"
+
+SP_USERNAME="admin@streampipes.apache.org"
+SP_PASSWORD="admin"
+
+while true; do
+    case "$1" in
+        -u)
+            SP_USERNAME="$2"
+            shift 2
+            ;;
+        -pw)
+            SP_PASSWORD="$2"
+            shift 2
+        ;;
+        -h)
+            HOST="$2"
+            shift 2
+        ;;
+        -p)
+            PORT="$2"
+            shift 2
+        ;;
+        -t)
+            E2E_TEST="$2"
+            shift 2
+        ;;
+        --help)
+            echo "Usage: $0 [-h <ip>] [-p <port>] [-u <username>] [-pw <password>] [-t <E2E_TEST>]"
+            exit 0
+        ;;
+        "")
+            #skip directly
+            break
+        ;;
+        *)
+          #skip directly
+            shift
+        ;;
+    esac
+done
+
+if [ E2E_TEST == "" ]; then
+    echo "-t is empty"
+    exit 1
+fi
+
+API_KEY_URL="/streampipes-backend/api/v2/users/$SP_USERNAME/tokens"
+API_KEY_USER_NAME="$SP_USERNAME"
+
+loginRequestBody='{
+  "username": "'"$SP_USERNAME"'",
+  "password": "'"$SP_PASSWORD"'"
+}'
+
+if nc -zv localhost 8030; then
+    echo "Port 8030 is open and listening."
+else
+    echo "Port 8030 is not open or not listening."
+    exit 1
+fi
+
+echo "start login"
+max_attempts=60
+attempt=1
+while [ $attempt -le $max_attempts ]
+do
+    response=$(curl -s -X POST "http://$HOST:$PORT$LOGIN_URL" \
+        -H "Content-Type: application/json" \
+        -d "$loginRequestBody")
+    if [ $? -eq 0 ]; then
+        echo "Login successful"
+        break
+    else
+        echo "$response"
+        echo "Error: Login request failed on attempt $attempt"
+        if [ $attempt -eq $max_attempts ]; then
+            echo "Max attempts reached. Exiting."
+            exit 0
+        else
+            echo "Retrying in 1 second..."
+            sleep 1
+        fi
+    fi
+    attempt=$((attempt+1))
+done
+
+echo "get token"
+accessToken=$(echo "$response" | sed -n 's/.*"accessToken":"\([^"]*\)".*/\1/p')
+if [ -z "$accessToken" ]; then
+    echo "Error: Failed to retrieve access token"
+    exit 1
+fi
+
+apiKeyRequestBody='{
+  "tokenName": "'"$API_KEY_USER_NAME"'"
+}'
+
+echo "install element"
+chmod +x ./install-element.sh
+./install-element.sh -host "$HOST" -port "$PORT" -token "$accessToken"
+
+echo "get apikey"
+# Get APIKEY
+APIKEYRESP=$(curl -s -X POST "http://$HOST:$PORT$API_KEY_URL" \
+   -H "Content-Type: application/json" \
+   -H "authorization: Bearer $accessToken" \
+   -d "$apiKeyRequestBody")
+if [ $? -ne 0 ]; then
+    echo "Error: API Key request failed"
+    exit 1
+fi
+APIKEY=$(echo "$APIKEYRESP" | sed 's/.*"rawToken":"\([^"]*\)".*/\1/')
+if [ -z "$APIKEY" ]; then
+    echo "Error: Failed to retrieve API key"
+    exit 1
+fi
+
+echo "start e2e test"
+chmod +x ./"$E2E_TEST"
+./"$E2E_TEST" -h "$HOST" -p "$PORT" -u "$API_KEY_USER_NAME" -k "$APIKEY"
+if [ $? -ne 0 ]; then
+    echo "start $E2E_TEST failed"
+    exit 1
+fi
+
+echo "All tests passed successfully"
\ No newline at end of file
diff --git a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
index 6d920fb751..e1e18324ce 100644
--- a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
+++ b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java
@@ -36,12 +36,12 @@ import static org.mockito.Mockito.when;
 public class AdapterMasterManagementTest {
 
   @Test
-  public void getAdapterFailNull() {
-    AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
-    AdapterResourceManager resourceManager = mock(AdapterResourceManager.class);
+  public void getAdapter_FailNull() {
+    var adapterStorage = mock(AdapterInstanceStorageImpl.class);
+    var resourceManager = mock(AdapterResourceManager.class);
     when(adapterStorage.findAll()).thenReturn(null);
 
-    AdapterMasterManagement adapterMasterManagement =
+    var adapterMasterManagement =
         new AdapterMasterManagement(
             adapterStorage,
             resourceManager,
@@ -53,13 +53,13 @@ public class AdapterMasterManagementTest {
   }
 
   @Test
-  public void getAdapterFail() {
-    List<AdapterDescription> adapterDescriptions = List.of(new AdapterDescription());
-    AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
-    AdapterResourceManager resourceManager = mock(AdapterResourceManager.class);
+  public void getAdapter_Fail() {
+    var adapterDescriptions = List.of(new AdapterDescription());
+    var adapterStorage = mock(AdapterInstanceStorageImpl.class);
+    var resourceManager = mock(AdapterResourceManager.class);
     when(adapterStorage.findAll()).thenReturn(adapterDescriptions);
 
-    AdapterMasterManagement adapterMasterManagement =
+    var adapterMasterManagement =
         new AdapterMasterManagement(
             adapterStorage,
             resourceManager,
@@ -71,10 +71,10 @@ public class AdapterMasterManagementTest {
   }
 
   @Test
-  public void getAllAdaptersSuccess() throws AdapterException {
-    List<AdapterDescription> adapterDescriptions = List.of(new AdapterDescription());
-    AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
-    AdapterResourceManager resourceManager = mock(AdapterResourceManager.class);
+  public void getAllAdapters_Success() throws AdapterException {
+    var adapterDescriptions = List.of(new AdapterDescription());
+    var adapterStorage = mock(AdapterInstanceStorageImpl.class);
+    var resourceManager = mock(AdapterResourceManager.class);
     when(adapterStorage.findAll()).thenReturn(adapterDescriptions);
 
     AdapterMasterManagement adapterMasterManagement =
@@ -91,12 +91,12 @@ public class AdapterMasterManagementTest {
   }
 
   @Test
-  public void getAllAdaptersFail() {
-    AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
-    AdapterResourceManager resourceManager = mock(AdapterResourceManager.class);
+  public void getAllAdapters_Fail() {
+    var adapterStorage = mock(AdapterInstanceStorageImpl.class);
+    var resourceManager = mock(AdapterResourceManager.class);
     when(adapterStorage.findAll()).thenReturn(null);
 
-    AdapterMasterManagement adapterMasterManagement =
+    var adapterMasterManagement =
         new AdapterMasterManagement(
             adapterStorage,
             resourceManager,
diff --git a/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java
index d815855490..d4616713da 100644
--- a/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java
+++ b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java
@@ -18,25 +18,59 @@
 
 package org.apache.streampipes.resource.management;
 
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.model.Tuple2;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.storage.api.IAdapterStorage;
 
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class AdapterResourceManagerTest {
 
+  private IAdapterStorage storage;
+  private AdapterResourceManager adapterResourceManager;
+
+  @BeforeEach
+  void setUp() {
+    storage = mock(IAdapterStorage.class);
+    adapterResourceManager = new AdapterResourceManager(storage);
+  }
+
   @Test
-  public void encryptAndUpdateValidateDescriptionStored() {
+  public void encryptAndUpdate_ValidateDescriptionStored() throws AdapterException {
 
-    IAdapterStorage storage = mock(IAdapterStorage.class);
-    AdapterResourceManager adapterResourceManager = new AdapterResourceManager(storage);
     adapterResourceManager.encryptAndUpdate(new AdapterDescription());
 
     verify(storage, times(1)).updateElement(any());
   }
+
+  @Test
+  void encryptAndCreate_ReturnsIdOnSuccess() throws AdapterException {
+    var id = "adapterId";
+
+    when(storage.persist(any())).thenReturn(new Tuple2<>(true, id));
+    var result = adapterResourceManager.encryptAndCreate(new AdapterDescription());
+
+    verify(storage, times(1)).persist(any());
+    assertEquals(id, result);
+  }
+
+  @Test
+  void encryptAndCreate_ThrowsAdapterExceptionOnConflict() {
+
+    doThrow(new org.lightcouch.DocumentConflictException("Conflict")).when(storage).persist(any());
+
+    assertThrows(AdapterException.class, () -> adapterResourceManager.encryptAndCreate(new AdapterDescription()));
+  }
+
 }
diff --git a/ui/cypress/support/utils/ProcessingElementTestUtils.ts b/ui/cypress/support/utils/ProcessingElementTestUtils.ts
index f2607a112b..6340635b48 100644
--- a/ui/cypress/support/utils/ProcessingElementTestUtils.ts
+++ b/ui/cypress/support/utils/ProcessingElementTestUtils.ts
@@ -18,7 +18,7 @@
 
 import { FileManagementUtils } from './FileManagementUtils';
 import { ConnectUtils } from './connect/ConnectUtils';
-import { PipelineUtils } from './PipelineUtils';
+import { PipelineUtils } from './pipeline/PipelineUtils';
 import { DataLakeUtils } from './datalake/DataLakeUtils';
 import { PipelineBuilder } from '../builder/PipelineBuilder';
 import { PipelineElementBuilder } from '../builder/PipelineElementBuilder';
diff --git a/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts b/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts
new file mode 100644
index 0000000000..c19e345f8d
--- /dev/null
+++ b/ui/cypress/tests/connect/compact/addCompactAdapter.spec.ts
@@ -0,0 +1,99 @@
+/*
+ * 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 { CompactAdapterUtils } from '../../../support/utils/connect/CompactAdapterUtils';
+import { PipelineUtils } from '../../../support/utils/pipeline/PipelineUtils';
+
+describe('Add Compact Adapters', () => {
+    beforeEach('Setup Test', () => {
+        cy.initStreamPipesTest();
+    });
+
+    it('Add an adapter via the compact API. Do not start', () => {
+        const compactAdapter =
+            CompactAdapterUtils.getMachineDataSimulator().build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(() => {
+            ConnectUtils.validateAdapterIsStopped();
+
+            PipelineUtils.checkAmountOfPipelinesPipeline(0);
+        });
+    });
+
+    it('Add an adapter via the compact API. Start Adapter', () => {
+        const compactAdapter = CompactAdapterUtils.getMachineDataSimulator()
+            .setStart()
+            .build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(() => {
+            ConnectUtils.validateAdapterIsRunning();
+
+            PipelineUtils.checkAmountOfPipelinesPipeline(0);
+        });
+    });
+
+    it('Add an adapter via the compact API. Start Adapter and start persist pipeline', () => {
+        const compactAdapter = CompactAdapterUtils.getMachineDataSimulator()
+            .setStart()
+            .setPersist()
+            .build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(() => {
+            ConnectUtils.validateAdapterIsRunning();
+
+            PipelineUtils.checkAmountOfPipelinesPipeline(1);
+        });
+    });
+
+    it('Add an adapter via the compact API via yml API.', () => {
+        cy.readFile(
+            'cypress/fixtures/connect/compact/machineDataSimulator.yml',
+        ).then(ymlDescription => {
+            CompactAdapterUtils.storeCompactYmlAdapter(ymlDescription).then(
+                () => {
+                    ConnectUtils.validateAdapterIsStopped();
+                },
+            );
+        });
+    });
+
+    it('Ensure correct error code when adapter with the same id already exists', () => {
+        const compactAdapter = CompactAdapterUtils.getMachineDataSimulator()
+            .setStart()
+            .setPersist()
+            .build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(
+            response => {
+                expect(response.status).to.equal(200);
+                ConnectUtils.checkAmountOfAdapters(1);
+
+                // Store the same adapter a second time and validate that resource returns status of conflict
+                CompactAdapterUtils.storeCompactAdapter(
+                    compactAdapter,
+                    false,
+                ).then(response => {
+                    expect(response.status).to.equal(409);
+
+                    ConnectUtils.checkAmountOfAdapters(1);
+                });
+            },
+        );
+    });
+});
diff --git a/ui/cypress/tests/connect/compact/compactAdapterWithTransformation.spec.ts b/ui/cypress/tests/connect/compact/compactAdapterWithTransformation.spec.ts
new file mode 100644
index 0000000000..4af2f359e2
--- /dev/null
+++ b/ui/cypress/tests/connect/compact/compactAdapterWithTransformation.spec.ts
@@ -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.
+ *
+ */
+
+import { ConnectUtils } from '../../../support/utils/connect/ConnectUtils';
+import { CompactAdapterUtils } from '../../../support/utils/connect/CompactAdapterUtils';
+import { ConnectBtns } from '../../../support/utils/connect/ConnectBtns';
+
+describe('Add Compact Adapters', () => {
+    beforeEach('Setup Test', () => {
+        cy.initStreamPipesTest();
+    });
+
+    it('Add an adapter and rename a property', () => {
+        const newPropertyName = 'temperature_renamed';
+        const compactAdapter = CompactAdapterUtils.getMachineDataSimulator()
+            .withRename('temperature', newPropertyName)
+            .setStart()
+            .build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(() => {
+            const runtimeNames = [
+                'density',
+                'mass_flow',
+                'sensor_fault_flags',
+                'sensorId',
+                newPropertyName,
+                'timestamp',
+                'volume_flow',
+            ];
+
+            ConnectUtils.validateEventSchema(runtimeNames);
+        });
+    });
+
+    it('Add an adapter and change measurement unit', () => {
+        const compactAdapter = CompactAdapterUtils.getMachineDataSimulator()
+            .withMeasurementUnit(
+                'temperature',
+                'http://qudt.org/vocab/unit#DegreeFahrenheit',
+            )
+            .setStart()
+            .build();
+
+        CompactAdapterUtils.storeCompactAdapter(compactAdapter).then(() => {
+            ConnectUtils.goToConnect();
+            ConnectBtns.detailsAdapter().click();
+
+            // This assertion works because the original value is below 100
+            // with the transformation the value is above 100
+            ConnectUtils.getLivePreviewValue('temperature')
+                .invoke('text')
+                .then(text => {
+                    const value = parseFloat(text.trim());
+                    expect(value).to.be.greaterThan(100);
+                });
+        });
+    });
+});
diff --git a/ui/cypress/tests/connect/compact/uiConfiguration.spec.ts b/ui/cypress/tests/connect/compact/uiConfiguration.spec.ts
new file mode 100644
index 0000000000..798d8a8caf
--- /dev/null
+++ b/ui/cypress/tests/connect/compact/uiConfiguration.spec.ts
@@ -0,0 +1,55 @@
+/*
+ * 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 { AdapterBuilder } from '../../../support/builder/AdapterBuilder';
+import { ConnectBtns } from '../../../support/utils/connect/ConnectBtns';
+import { GeneralUtils } from '../../../support/utils/GeneralUtils';
+import { CompactUtils } from '../../../support/utils/CompactUtils';
+
+describe('Test Compact Adapters', () => {
+    beforeEach('Setup Test', () => {
+        cy.initStreamPipesTest();
+    });
+
+    it('Validate that code for existing adapter is displayed correctly', () => {
+        const adapterInput = AdapterBuilder.create('Machine_Data_Simulator')
+            .setName('Machine Data Simulator Test')
+            .addInput('input', 'wait-time-ms', '1000')
+            .build();
+
+        ConnectUtils.addAdapter(adapterInput);
+
+        // Validate code editor in start dialog
+        ConnectBtns.showCodeCheckbox().parent().click();
+        validateCodeEditor();
+
+        ConnectUtils.startAdapter(adapterInput);
+
+        // Validate code editor in adapter details
+        ConnectBtns.detailsAdapter().click();
+        GeneralUtils.tab('Code');
+        validateCodeEditor();
+    });
+
+    const validateCodeEditor = () => {
+        CompactUtils.ymlConfiguration().should('be.visible');
+        cy.contains('span', 'JSON').click();
+        CompactUtils.jsonConfiguration().should('be.visible');
+    };
+});
diff --git a/ui/cypress/tests/connect/deleteAdapterWithMultipleUsers.smoke.spec.ts b/ui/cypress/tests/connect/deleteAdapterWithMultipleUsers.smoke.spec.ts
index 29c854cb23..a11332ac9e 100644
--- a/ui/cypress/tests/connect/deleteAdapterWithMultipleUsers.smoke.spec.ts
+++ b/ui/cypress/tests/connect/deleteAdapterWithMultipleUsers.smoke.spec.ts
@@ -20,7 +20,7 @@ import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { UserUtils } from '../../support/utils/UserUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 
 const adapterName = 'simulator';
 
diff --git a/ui/cypress/tests/connect/editAdapter.smoke.spec.ts b/ui/cypress/tests/connect/editAdapter.smoke.spec.ts
index 214e8433c8..a1d20139eb 100644
--- a/ui/cypress/tests/connect/editAdapter.smoke.spec.ts
+++ b/ui/cypress/tests/connect/editAdapter.smoke.spec.ts
@@ -57,6 +57,8 @@ describe('Test Edit Adapter', () => {
 
         cy.dataCy('sp-adapter-name').clear().type(newAdapterName);
 
+        // This wait is required to ensure that there is no couch db update conflict
+        cy.wait(1000);
         ConnectBtns.storeEditAdapter().click();
 
         cy.dataCy('sp-connect-adapter-success-added', {
diff --git a/ui/cypress/tests/connect/editAdapterSettingsAndPipeline.spec.ts b/ui/cypress/tests/connect/editAdapterSettingsAndPipeline.spec.ts
index c30f355474..c3f14b7a29 100644
--- a/ui/cypress/tests/connect/editAdapterSettingsAndPipeline.spec.ts
+++ b/ui/cypress/tests/connect/editAdapterSettingsAndPipeline.spec.ts
@@ -18,7 +18,7 @@
 
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
 import { ConnectBtns } from '../../support/utils/connect/ConnectBtns';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
diff --git a/ui/cypress/tests/connect/persistInDataLake.smoke.spec.ts b/ui/cypress/tests/connect/persistInDataLake.smoke.spec.ts
index 39c672b6a7..939bf44312 100644
--- a/ui/cypress/tests/connect/persistInDataLake.smoke.spec.ts
+++ b/ui/cypress/tests/connect/persistInDataLake.smoke.spec.ts
@@ -17,7 +17,7 @@
  */
 
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { FileManagementUtils } from '../../support/utils/FileManagementUtils';
 import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
 import { ConnectBtns } from '../../support/utils/connect/ConnectBtns';
diff --git a/ui/cypress/tests/datalake/configuration.smoke.spec.ts b/ui/cypress/tests/datalake/configuration.smoke.spec.ts
index 3b5532e154..1b095ab07e 100644
--- a/ui/cypress/tests/datalake/configuration.smoke.spec.ts
+++ b/ui/cypress/tests/datalake/configuration.smoke.spec.ts
@@ -16,7 +16,7 @@
  *
  */
 
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { DataLakeUtils } from '../../support/utils/datalake/DataLakeUtils';
 
 describe('Test Truncate data in datalake', () => {
diff --git a/ui/cypress/tests/experimental/restartStreamPipes/restartStreamPipes1.ts b/ui/cypress/tests/experimental/restartStreamPipes/restartStreamPipes1.ts
index 3f098de23a..d8eafe8bc1 100644
--- a/ui/cypress/tests/experimental/restartStreamPipes/restartStreamPipes1.ts
+++ b/ui/cypress/tests/experimental/restartStreamPipes/restartStreamPipes1.ts
@@ -17,7 +17,7 @@
  */
 
 import { ConnectUtils } from '../../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../../support/builder/PipelineBuilder';
 import { DashboardUtils } from '../../../support/utils/DashboardUtils';
diff --git a/ui/cypress/tests/experimental/testJvmArchetype/testJvmArchetype.ts b/ui/cypress/tests/experimental/testJvmArchetype/testJvmArchetype.ts
index 2a41d84eec..c15490d2e4 100644
--- a/ui/cypress/tests/experimental/testJvmArchetype/testJvmArchetype.ts
+++ b/ui/cypress/tests/experimental/testJvmArchetype/testJvmArchetype.ts
@@ -17,7 +17,7 @@
  */
 
 import { ConnectUtils } from '../../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../../support/builder/PipelineBuilder';
 import { DashboardUtils } from '../../../support/utils/DashboardUtils';
diff --git a/ui/cypress/tests/pipeline/pipelineTest.smoke.spec.ts b/ui/cypress/tests/pipeline/pipelineTest.smoke.spec.ts
index 1de708d27f..010f7695f9 100644
--- a/ui/cypress/tests/pipeline/pipelineTest.smoke.spec.ts
+++ b/ui/cypress/tests/pipeline/pipelineTest.smoke.spec.ts
@@ -17,7 +17,7 @@
  */
 
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 
diff --git a/ui/cypress/tests/pipeline/updatePipelineTest.smoke.spec.ts b/ui/cypress/tests/pipeline/updatePipelineTest.smoke.spec.ts
index b2e69bd0b4..5de2359906 100644
--- a/ui/cypress/tests/pipeline/updatePipelineTest.smoke.spec.ts
+++ b/ui/cypress/tests/pipeline/updatePipelineTest.smoke.spec.ts
@@ -17,7 +17,7 @@
  */
 
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 
diff --git a/ui/cypress/tests/pipelineElement/PipelineElementDocumentation.spec.ts b/ui/cypress/tests/pipelineElement/PipelineElementDocumentation.spec.ts
index 726ed1c5e2..4387a2e5e6 100644
--- a/ui/cypress/tests/pipelineElement/PipelineElementDocumentation.spec.ts
+++ b/ui/cypress/tests/pipelineElement/PipelineElementDocumentation.spec.ts
@@ -16,7 +16,7 @@
  *
  */
 
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 
 describe('Validate that the markdown documentation for pipeline elements works', () => {
     beforeEach('Setup Test', () => {
diff --git a/ui/cypress/tests/pipelineElementConfigurationTemplate/pipelineElementConfigurationTemplate.ts b/ui/cypress/tests/pipelineElementConfigurationTemplate/pipelineElementConfigurationTemplate.ts
index 2b9fff9442..16ea5c3543 100644
--- a/ui/cypress/tests/pipelineElementConfigurationTemplate/pipelineElementConfigurationTemplate.ts
+++ b/ui/cypress/tests/pipelineElementConfigurationTemplate/pipelineElementConfigurationTemplate.ts
@@ -16,7 +16,7 @@
  *
  */
 
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
diff --git a/ui/cypress/tests/userManagement/testGroupManagement.spec.ts b/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
index 2f4de9e7bf..03d7159568 100644
--- a/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
+++ b/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
@@ -20,7 +20,7 @@ import { UserBuilder } from '../../support/builder/UserBuilder';
 import { UserRole } from '../../../src/app/_enums/user-role.enum';
 import { UserUtils } from '../../support/utils/UserUtils';
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 
diff --git a/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts b/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
index a992334de8..94d83418ff 100644
--- a/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
+++ b/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
@@ -20,7 +20,7 @@ import { UserBuilder } from '../../support/builder/UserBuilder';
 import { UserRole } from '../../../src/app/_enums/user-role.enum';
 import { UserUtils } from '../../support/utils/UserUtils';
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
-import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { PipelineUtils } from '../../support/utils/pipeline/PipelineUtils';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';