Repository: apache/streampipes
Pull request #2845 (2024-05-09 19:15:57)

refactor(#2843): Enhance user handling in cypress tests
<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~    http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
This PR cleans up the user handling in the e2e tests a bit.

### 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/installer/k8s/README.md b/installer/k8s/README.md
index e7597c61a8..3d1c0563d0 100644
--- a/installer/k8s/README.md
+++ b/installer/k8s/README.md
@@ -82,7 +82,7 @@ For **minikube users**:
 > to a privileged port).
 
 ```bash
-kubectl port-forward svc/ui --address=0.0.0.0 80:80
+kubectl port-forward svc/ui --address=0.0.0.0 8088:8088
 ```
 
 **Deleting** the current helm chart deployment:
diff --git a/pom.xml b/pom.xml
index c2b603c2ec..ba022a2c95 100644
--- a/pom.xml
+++ b/pom.xml
@@ -67,6 +67,7 @@
         <httpcore.version>4.4.9</httpcore.version>
         <influxdb.version>2.24</influxdb.version>
         <inlong.version>1.12.0</inlong.version>
+        <iotdb.version>1.3.0</iotdb.version>
         <jackson.version>2.17.0</jackson.version>
         <jackson.databind.version>2.17.0</jackson.databind.version>
         <jakarta-annotation.version>3.0.0</jakarta-annotation.version>
@@ -95,7 +96,7 @@
         <micrometer-prometheus.version>1.12.0</micrometer-prometheus.version>
         <micrometer-observation.version>1.12.0</micrometer-observation.version>
         <mqtt-client.version>1.12</mqtt-client.version>
-        <nats.version>2.17.0</nats.version>
+        <nats.version>2.18.0</nats.version>
         <netty.version>4.1.72.Final</netty.version>
         <okhttp.version>3.13.1</okhttp.version>
         <opencsv.version>5.9</opencsv.version>
@@ -128,7 +129,7 @@
 
         <!-- Test dependencies -->
         <junit.version>5.10.1</junit.version>
-        <mockito.version>5.11.0</mockito.version>
+        <mockito.version>5.12.0</mockito.version>
 
         <!-- Maven settings & dependencies -->
         <apache-rat-plugin.version>0.13</apache-rat-plugin.version>
@@ -139,7 +140,6 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <maven-checkstyle-plugin.version>3.3.1</maven-checkstyle-plugin.version>
-        <maven.dependency.check.plugin.version>9.1.0</maven.dependency.check.plugin.version>
         <maven-shade-plugin.version>3.5.1</maven-shade-plugin.version>
         <maven.version>3.8.7</maven.version>
         <typescript-generator-maven-plugin.version>3.2.1263</typescript-generator-maven-plugin.version>
@@ -147,7 +147,6 @@
 
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <dependency.locations.enabled>false</dependency.locations.enabled>
-        <owasp.check.skip>true</owasp.check.skip>
     </properties>
 
     <dependencyManagement>
@@ -389,6 +388,11 @@
                 <artifactId>httpcore</artifactId>
                 <version>${httpcore.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.iotdb</groupId>
+                <artifactId>iotdb-session</artifactId>
+                <version>${iotdb.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.kafka</groupId>
                 <artifactId>kafka-clients</artifactId>
@@ -825,6 +829,7 @@
         <module>streampipes-data-explorer-api</module>
         <module>streampipes-data-explorer-export</module>
         <module>streampipes-data-explorer-influx</module>
+        <module>streampipes-data-explorer-iotdb</module>
         <module>streampipes-data-explorer-management</module>
         <module>streampipes-data-export</module>
         <module>streampipes-dataformat</module>
@@ -868,7 +873,6 @@
         <module>streampipes-storage-management</module>
         <module>streampipes-test-utils</module>
         <module>streampipes-test-utils-executors</module>
-        <module>streampipes-ts-store-iotdb</module>
         <module>streampipes-user-management</module>
         <module>streampipes-vocabulary</module>
         <module>streampipes-wrapper</module>
@@ -1027,11 +1031,6 @@
                         <failOnViolation>true</failOnViolation>
                     </configuration>
                 </plugin>
-                <plugin>
-                    <groupId>org.owasp</groupId>
-                    <artifactId>dependency-check-maven</artifactId>
-                    <version>${maven.dependency.check.plugin.version}</version>
-                </plugin>
                 <plugin>
                     <groupId>org.springframework.boot</groupId>
                     <artifactId>spring-boot-maven-plugin</artifactId>
@@ -1281,28 +1280,6 @@
                     </archive>
                 </configuration>
             </plugin>
-            <plugin>
-                <groupId>org.owasp</groupId>
-                <artifactId>dependency-check-maven</artifactId>
-                <configuration>
-                    <!-- use -Dowasp.check.skip=true to skip dependency check -->
-                    <skip>${owasp.check.skip}</skip>
-                    <format>HTML</format>
-                    <bundleAuditAnalyzerEnabled>false</bundleAuditAnalyzerEnabled>
-                    <assemblyAnalyzerEnabled>false</assemblyAnalyzerEnabled>
-                    <failOnError>false</failOnError>
-                    <suppressionFile>tools/maven/owasp-dependency-check-suppression.xml</suppressionFile>
-                </configuration>
-                <executions>
-                    <execution>
-                        <goals>
-                            <!-- generates aggregated reports in target/ -->
-                            <goal>check</goal>
-                            <goal>aggregate</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
             <plugin>
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
diff --git a/streampipes-client-python/mkdocs.yml b/streampipes-client-python/mkdocs.yml
index 763ef120ec..3e774bb565 100644
--- a/streampipes-client-python/mkdocs.yml
+++ b/streampipes-client-python/mkdocs.yml
@@ -29,7 +29,7 @@ copyright: "Apache License 2.0"
 theme:
   name: material
   logo: https://streampipes.apache.org/img/sp-logo-white-right.png
-  favicon: https://streampipes.apache.org/img/favicon/favicon.png
+  favicon: https://streampipes.apache.org/img/favicon.png
   custom_dir: docs/overrides
   features:
     - navigation.tabs
@@ -39,7 +39,6 @@ theme:
     - navigation.tracking
   font: false
   palette:
-
     # Palette toggle for light mode
     - scheme: default
       toggle:
@@ -62,6 +61,9 @@ extra:
   version:
     provider: mike
 
+exclude_docs: |
+  /scripts/**
+
 # Extensions
 markdown_extensions:
   - admonition
@@ -94,9 +96,10 @@ plugins:
             docstring_style: numpy
             docstring_section_style: spacy
             show_source: False
-            merge_init_into_class: True
+            merge_init_into_class: true
+            inherited_members: true
   - mike:
-      canonical_version: 'latest'
+      canonical_version: "latest"
       version_selector: true
       css_dir: css
       javascript_dir: js
diff --git a/streampipes-client-python/streampipes/client/credential_provider.py b/streampipes-client-python/streampipes/client/credential_provider.py
index 461088e8fc..d7493b40c2 100644
--- a/streampipes-client-python/streampipes/client/credential_provider.py
+++ b/streampipes-client-python/streampipes/client/credential_provider.py
@@ -81,7 +81,7 @@ class CredentialProvider(ABC):
 class StreamPipesApiKeyCredentials(CredentialProvider):
     """A credential provider that allows authentication via a StreamPipes API Token.
 
-    The required token can be generated via the StreamPipes UI (see the description on our [start-page](../../../).
+    The required token can be generated via the StreamPipes UI (see the description on our [start-page](../../index.md).
 
     Both parameters can either be passed as arguments or remain unset.
     If they are not passed, they are retrieved from environment variables:
diff --git a/streampipes-client-python/streampipes/model/resource/function_definition.py b/streampipes-client-python/streampipes/model/resource/function_definition.py
index 7917180797..b055f4fc3e 100644
--- a/streampipes-client-python/streampipes/model/resource/function_definition.py
+++ b/streampipes-client-python/streampipes/model/resource/function_definition.py
@@ -34,7 +34,7 @@ class FunctionId(BasicModel):
 
     Maps to the `FunctionId` class defined in the StreamPipes model.
 
-    Parameters
+    Attributes
     ----------
     id: str
         unique identifier of the function instance
@@ -56,15 +56,12 @@ class FunctionDefinition(Resource):
     This class maps to the `FunctionDefinition` class in the StreamPipes model.
     It contains all metadata that are required to register a function at the StreamPipes backend.
 
-    Parameters
+    Attributes
     ----------
     consumed_streams: List[str]
         List of data streams the function is consuming from
     function_id: FunctionId
         identifier object of a StreamPipes function
-
-    Attributes
-    ----------
     output_data_streams: Dict[str, DataStream]
         Map off all output data streams added to the function definition
 
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
index 472cb7eca8..14b5ccf1c1 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.connect.management.management;
 
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.commons.exceptions.SpConfigurationException;
 import org.apache.streampipes.commons.exceptions.connect.ParseException;
 import org.apache.streampipes.connect.management.AdapterEventPreviewPipeline;
 import org.apache.streampipes.connect.management.util.WorkerPaths;
@@ -31,6 +30,7 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.guess.AdapterEventPreview;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.model.monitoring.SpLogMessage;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
 import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
@@ -75,8 +75,8 @@ public class GuessManagement {
     if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
       return objectMapper.readValue(responseString, GuessSchema.class);
     } else {
-      var exception = objectMapper.readValue(responseString, SpConfigurationException.class);
-      throw new WorkerAdapterException(exception.getMessage(), exception.getCause());
+      var exception = objectMapper.readValue(responseString, SpLogMessage.class);
+      throw new WorkerAdapterException(exception);
     }
   }
 
diff --git a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ToTransformedSchemaConverter.java b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ToTransformedSchemaConverter.java
index b8bec63fa6..37715185ee 100644
--- a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ToTransformedSchemaConverter.java
+++ b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/convert/ToTransformedSchemaConverter.java
@@ -36,6 +36,7 @@ import org.apache.streampipes.model.connect.rules.value.UnitTransformRuleDescrip
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventPropertyNested;
 import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.PropertyScope;
 import org.apache.streampipes.model.util.Cloner;
 import org.apache.streampipes.sdk.helpers.EpProperties;
 import org.apache.streampipes.sdk.utils.Datatypes;
@@ -149,6 +150,7 @@ public class ToTransformedSchemaConverter implements ITransformationRuleVisitor,
     var property = findPrimitiveProperty(properties, rule.getRuntimeKey());
     property.setDomainProperties(List.of(URI.create("http://schema.org/DateTime")));
     property.setRuntimeType(Datatypes.Long.toString());
+    property.setPropertyScope(PropertyScope.HEADER_PROPERTY.toString());
     var metadata = property.getAdditionalMetadata();
     metadata.put("mode", rule.getMode());
     metadata.put("formatString", rule.getFormatString());
diff --git a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/StatelessTransformationRuleGeneratorVisitor.java b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/StatelessTransformationRuleGeneratorVisitor.java
index 7adf1b3765..1728d8db27 100644
--- a/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/StatelessTransformationRuleGeneratorVisitor.java
+++ b/streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/StatelessTransformationRuleGeneratorVisitor.java
@@ -117,7 +117,7 @@ public class StatelessTransformationRuleGeneratorVisitor extends TransformationR
     if (ruleDesc.getMode().equals(TimestampTranformationRuleMode.FORMAT_STRING.internalName())) {
       mode = TimestampTranformationRuleMode.FORMAT_STRING;
     } else {
-      mode = TimestampTranformationRuleMode.FORMAT_STRING;
+      mode = TimestampTranformationRuleMode.TIME_UNIT;
     }
 
     rules.add(new TimestampTransformationRule(Utils.toKeyArray(
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
index e79cb8d2bc..1c25d5fbc5 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerManagerInflux.java
@@ -34,9 +34,7 @@ import org.apache.streampipes.storage.management.StorageDispatcher;
 
 import java.util.List;
 
-public enum DataExplorerManagerInflux implements IDataExplorerManager {
-
-  INSTANCE;
+public class DataExplorerManagerInflux implements IDataExplorerManager {
 
   @Override
   public IDataLakeMeasurementCounter getMeasurementCounter(
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
index 4f516f52c1..db11222324 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCounterInflux.java
@@ -18,86 +18,36 @@
 
 package org.apache.streampipes.dataexplorer.influx;
 
-import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter;
+import org.apache.streampipes.dataexplorer.query.DataLakeMeasurementCounter;
 import org.apache.streampipes.model.datalake.AggregationFunction;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
-import org.apache.streampipes.model.datalake.SpQueryResult;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.PropertyScope;
 
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
 
-public class DataLakeMeasurementCounterInflux implements IDataLakeMeasurementCounter {
-
-  private final List<DataLakeMeasure> allMeasurements;
-  private final List<String> measurementNames;
+public class DataLakeMeasurementCounterInflux extends DataLakeMeasurementCounter {
 
   private static final String COUNT_FIELD = "count";
 
   public DataLakeMeasurementCounterInflux(List<DataLakeMeasure> allMeasurements,
                                           List<String> measurementNames) {
-    this.allMeasurements = allMeasurements;
-    this.measurementNames = measurementNames;
+    super(allMeasurements, measurementNames);
   }
 
-  public Map<String, Integer> countMeasurementSizes() {
-    Map<String, CompletableFuture<Integer>> futures = measurementNames.stream()
-        .distinct()
-        .map(this::getMeasure)
-        .collect(Collectors.toMap(DataLakeMeasure::getMeasureName, m -> CompletableFuture.supplyAsync(() -> {
-          var firstColumn = getFirstColumn(m);
-          var builder = DataLakeInfluxQueryBuilder
-              .create(m.getMeasureName()).withEndTime(System.currentTimeMillis())
-              .withAggregatedColumn(firstColumn, AggregationFunction.COUNT);
-          var queryResult = new DataExplorerInfluxQueryExecutor().executeQuery(builder.build(), Optional.empty(), true);
-          if (queryResult.getTotal() > 0) {
-            var headers = queryResult.getHeaders();
-            return extractResult(queryResult, headers);
-          } else {
-            return 0;
-          }
-        })));
-
-    Map<String, Integer> result = new HashMap<>();
-    futures.entrySet().forEach((entry -> {
-      try {
-        result.put(entry.getKey(), entry.getValue().get());
-      } catch (InterruptedException | ExecutionException e) {
-        result.put(entry.getKey(), 0);
+  @Override
+  protected CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure) {
+    return CompletableFuture.supplyAsync(() -> {
+      var firstColumn = getFirstMeasurementProperty(measure);
+      var builder = DataLakeInfluxQueryBuilder
+          .create(measure.getMeasureName()).withEndTime(System.currentTimeMillis())
+          .withAggregatedColumn(firstColumn, AggregationFunction.COUNT);
+      var queryResult = new DataExplorerInfluxQueryExecutor().executeQuery(builder.build(), Optional.empty(), true);
+      if (queryResult.getTotal() > 0) {
+        return extractResult(queryResult, COUNT_FIELD);
+      } else {
+        return 0;
       }
-    }));
-
-    return result;
-  }
-
-  private Integer extractResult(SpQueryResult queryResult,
-                                List<String> headers) {
-    return ((Double) (
-        queryResult.getAllDataSeries().get(0).getRows().get(0).get(headers.indexOf(COUNT_FIELD)))
-    ).intValue();
-  }
-
-  private DataLakeMeasure getMeasure(String measureName) {
-    return allMeasurements
-        .stream()
-        .filter(m -> m.getMeasureName().equals(measureName))
-        .findFirst()
-        .orElse(null);
-  }
-
-  private String getFirstColumn(DataLakeMeasure measure) {
-    return measure.getEventSchema().getEventProperties()
-        .stream()
-        .filter(ep -> ep.getPropertyScope() != null
-            && ep.getPropertyScope().equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
-        .map(EventProperty::getRuntimeName)
-        .findFirst()
-        .orElse(null);
+    });
   }
 }
diff --git a/streampipes-ts-store-iotdb/pom.xml b/streampipes-data-explorer-iotdb/pom.xml
similarity index 74%
rename from streampipes-ts-store-iotdb/pom.xml
rename to streampipes-data-explorer-iotdb/pom.xml
index ff294b445d..763c932a36 100644
--- a/streampipes-ts-store-iotdb/pom.xml
+++ b/streampipes-data-explorer-iotdb/pom.xml
@@ -24,7 +24,7 @@
         <version>0.97.0-SNAPSHOT</version>
     </parent>
 
-    <artifactId>streampipes-ts-store-iotdb</artifactId>
+    <artifactId>streampipes-data-explorer-iotdb</artifactId>
 
     <properties>
         <maven.compiler.source>17</maven.compiler.source>
@@ -44,6 +44,11 @@
             <artifactId>streampipes-commons</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-api</artifactId>
+            <version>0.97.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-data-explorer</artifactId>
@@ -64,7 +69,16 @@
         <dependency>
             <groupId>org.apache.iotdb</groupId>
             <artifactId>iotdb-session</artifactId>
-            <version>1.3.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.xerial.snappy</groupId>
+                    <artifactId>snappy-java</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.github.luben</groupId>
+                    <artifactId>zstd-jni</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
@@ -72,11 +86,22 @@
         </dependency>
 
         <!--Test dependencies -->
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.97.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java
new file mode 100644
index 0000000000..9622aebd48
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerIotDbQueryExecutor.java
@@ -0,0 +1,96 @@
+/*
+ * 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.dataexplorer.iotdb;
+
+import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
+import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Optional;
+
+public class DataExplorerIotDbQueryExecutor extends DataExplorerQueryExecutor<String, SessionDataSetWrapper> {
+
+  private final SessionPool sessionPool;
+
+  public DataExplorerIotDbQueryExecutor(SessionPool sessionPool){
+    this.sessionPool = sessionPool;
+  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(DataExplorerIotDbQueryExecutor.class);
+  @Override
+  protected SpQueryResult postQuery(SessionDataSetWrapper queryResult, Optional<String> forIdOpt, boolean ignoreMissingValues) {
+    return null;
+  }
+
+  @Override
+  public SessionDataSetWrapper executeQuery(String query) {
+    try (var result = sessionPool.executeQueryStatement(query)) {
+      return result;
+    } catch (IoTDBConnectionException | StatementExecutionException e) {
+      throw new SpRuntimeException(e);
+    }
+  }
+
+  public boolean executeNonQueryStatement(String statement) {
+    try {
+      sessionPool.executeNonQueryStatement(statement);
+    } catch (StatementExecutionException | IoTDBConnectionException e) {
+      LOG.error("Error while executing non-query statement '{}': {}", statement, e.getMessage());
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  protected String asQueryString(String query) {
+    return "";
+  }
+
+  @Override
+  protected String makeDeleteQuery(DeleteQueryParams params) {
+    LOG.error("This method is not required for IoTDB.");
+    return "";
+  }
+
+  @Override
+  protected String makeSelectQuery(SelectQueryParams params) {
+    return "";
+  }
+
+  @Override
+  public Map<String, Object> getTagValues(String measurementId, String fields) {
+    return Map.of();
+  }
+
+  @Override
+  public boolean deleteData(DataLakeMeasure measure) {
+    var deleteTimeSeriesQuery = "DELETE timeseries root.streampipes.%s.*".formatted(measure.getMeasureName());
+    return executeNonQueryStatement(deleteTimeSeriesQuery);
+  }
+}
\ No newline at end of file
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
new file mode 100644
index 0000000000..e29b570f4a
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
@@ -0,0 +1,58 @@
+/*
+ * 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.dataexplorer.iotdb;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
+import org.apache.streampipes.dataexplorer.api.*;
+import org.apache.streampipes.dataexplorer.iotdb.sanitize.DataLakeMeasurementSanitizerIotDb;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.List;
+
+public class DataExplorerManagerIotDb implements IDataExplorerManager {
+
+  @Override
+  public IDataLakeMeasurementCounter getMeasurementCounter(List<DataLakeMeasure> allMeasurements, List<String> measurementsToCount) {
+    return new DataLakeMeasurementCounterIotDb(allMeasurements, measurementsToCount);
+  }
+
+  @Override
+  public IDataExplorerQueryManagement getQueryManagement(IDataExplorerSchemaManagement dataExplorerSchemaManagement) {
+    return null;
+  }
+
+  @Override
+  public IDataExplorerSchemaManagement getSchemaManagement() {
+    return new DataExplorerSchemaManagement(StorageDispatcher.INSTANCE
+                                                             .getNoSqlStore()
+                                                             .getDataLakeStorage());
+  }
+
+  @Override
+  public ITimeSeriesStorage getTimeseriesStorage(DataLakeMeasure measure) {
+    return new TimeSeriesStorageIotDb(measure, new IotDbPropertyConverter(), new IotDbSessionProvider());
+  }
+
+  @Override
+  public IDataLakeMeasurementSanitizer getMeasurementSanitizer(IStreamPipesClient client, DataLakeMeasure measure) {
+    return new DataLakeMeasurementSanitizerIotDb(client, measure);
+  }
+}
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java
new file mode 100644
index 0000000000..95ed59d6ac
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataLakeMeasurementCounterIotDb.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.iotdb;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.dataexplorer.query.DataLakeMeasurementCounter;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+public class DataLakeMeasurementCounterIotDb extends DataLakeMeasurementCounter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(DataLakeMeasurementCounterIotDb.class);
+
+  public DataLakeMeasurementCounterIotDb(List<DataLakeMeasure> allMeasurements,
+                                         List<String> measurementNames) {
+    super(allMeasurements, measurementNames);
+  }
+
+  /**
+   * Creates a CompletableFuture to execute a count query on a DataLakeMeasure asynchronously.
+   *
+   * @param measure The DataLakeMeasure object representing the measure to query.
+   * @return A CompletableFuture<Integer> representing the count query result as a future.
+   */
+  @Override
+  protected CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure) {
+    var sessionPool = new IotDbSessionProvider().getSessionPool(Environments.getEnvironment());
+    return CompletableFuture.supplyAsync(() -> {
+
+      // We want to apply the count query to only one property of the measurement, as this is sufficient and
+      // significantly reduces the complexity of the query compared to counting all available properties.
+      // So we can just take the first measurement
+      var propertyName = getFirstMeasurementProperty(measure);
+
+      try (var result = new DataExplorerIotDbQueryExecutor(sessionPool).executeQuery(
+          "Select count(%s) from root.streampipes.%s".formatted(propertyName, measure.getMeasureName())
+      )) {
+        var resultRecord = result.next();
+        if (resultRecord == null) {
+          LOG.error("Result of count query does not contain any row - empty result");
+          return 0;
+        }
+        return (int) resultRecord.getFields().get(0).getLongV();
+      } catch (IoTDBConnectionException | StatementExecutionException e) {
+        LOG.error("Error during count query execution: {}", e.getMessage());
+        return 0;
+      }
+    });
+  }
+}
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
similarity index 96%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
index 63cb18f4b4..24591da3bd 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbMeasurementRecord.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbMeasurementRecord.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
 
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
similarity index 96%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
index c0793ba9eb..4bb46a2cfd 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverter.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverter.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
 
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
@@ -32,7 +32,7 @@ import org.apache.streampipes.vocabulary.XSD;
 public class IotDbPropertyConverter {
 
   /**
-   * Converts a {@link EventPropertyPrimitive }into an {@link IotDbMeasurementRecord}.
+   * Converts a {@link EventPropertyPrimitive } into an {@link IotDbMeasurementRecord}.
    *
    * @param eventPropertyPrimitive The primitive event property to be converted
    * @param primitiveField         The primitive field containing the property value
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
similarity index 97%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
index 4a46ede043..6ff4bc38ca 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbSessionProvider.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/IotDbSessionProvider.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
 
 import org.apache.iotdb.session.pool.SessionPool;
 import org.apache.streampipes.commons.environment.Environment;
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java
new file mode 100644
index 0000000000..c233580d1f
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/TimeSeriesStorageIotDb.java
@@ -0,0 +1,175 @@
+/*
+ * 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.dataexplorer.iotdb;
+
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.TimeSeriesStorage;
+import org.apache.streampipes.dataexplorer.iotdb.sanitize.IotDbNameSanitizer;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TimeSeriesStorageIotDb extends TimeSeriesStorage {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TimeSeriesStorageIotDb.class);
+
+  private final IotDbPropertyConverter propertyConverter;
+  private final SessionPool sessionPool;
+
+  public TimeSeriesStorageIotDb(DataLakeMeasure measure,
+                                IotDbPropertyConverter propertyConverter,
+                                IotDbSessionProvider iotDbSessionProvider) {
+    super(measure);
+    this.propertyConverter = propertyConverter;
+    this.sessionPool = iotDbSessionProvider.getSessionPool(Environments.getEnvironment());
+  }
+
+  @Override
+  protected void storeSanitizedRuntimeNames() {
+    measure.getEventSchema()
+        .getEventProperties()
+        .forEach(ep -> sanitizedRuntimeNames.put(
+            ep.getRuntimeName(),
+            new IotDbNameSanitizer().renameReservedKeywords(ep.getRuntimeName())
+        ));
+  }
+
+  @Override
+  protected void sanitizeRuntimeNamesInEvent(Event event) {
+    event.getRaw()
+        .keySet()
+        .forEach(runtimeName -> {
+              // timestamp field does not need to be renamed since it is not written as measurement
+              // and taken care of separately
+              if (!runtimeName.equals(measure.getTimestampFieldName())) {
+                event.renameFieldByRuntimeName(
+                    runtimeName,
+                    new IotDbNameSanitizer().renameReservedKeywords(runtimeName)
+                );
+              }
+            }
+        );
+  }
+
+  @Override
+  protected void writeToTimeSeriesStorage(Event event) throws SpRuntimeException {
+    if (event == null) {
+      LOG.warn("Input event is null - skipping event");
+      return;
+    }
+
+    var timestampValue = event.getFieldBySelector(measure.getTimestampField())
+        .getAsPrimitive()
+        .getAsLong();
+
+    if (timestampValue == null) {
+      LOG.warn("Timestamp of input event is null - skipping event");
+      return;
+    }
+
+    if (event.getRaw().size() <= 1) {
+      LOG.warn("Event only consists of timestamp and does not include any measurement - skipping event");
+      return;
+    }
+
+    var iotDbRecords = extractMeasurementRecords(event);
+
+    if (iotDbRecords.isEmpty()) {
+      return;
+    }
+
+    insertIntoIotDb(timestampValue, iotDbRecords);
+  }
+
+  @Override
+  public void close() throws SpRuntimeException {
+    this.sessionPool.close();
+  }
+
+  /**
+   * Extracts all relevant information for the IotDb from the incoming event.
+   *
+   * @param event The event from which measurement records are extracted.
+   * @return An ArrayList of IotDbMeasurementRecord objects containing the extracted measurement records.
+   */
+  private ArrayList<IotDbMeasurementRecord> extractMeasurementRecords(Event event) {
+    var iotDbRecords = new ArrayList<IotDbMeasurementRecord>();
+    allEventProperties.forEach(ep -> {
+      try {
+        // timestamp is already known and is not part of the measurement thus we ignore it here
+        if (!ep.getRuntimeName().equals(measure.getTimestampFieldName())) {
+
+          try {
+            var eventField = event.getFieldByRuntimeName(ep.getRuntimeName()).getAsPrimitive();
+
+            if (ep instanceof EventPropertyPrimitive) {
+              iotDbRecords.add(propertyConverter.convertPrimitiveProperty(
+                  (EventPropertyPrimitive) ep,
+                  event.getFieldByRuntimeName(ep.getRuntimeName()).getAsPrimitive(),
+                  sanitizedRuntimeNames.get(ep.getRuntimeName())
+              ));
+            } else {
+              iotDbRecords.add(propertyConverter.convertNonPrimitiveProperty(
+                  ep,
+                  sanitizedRuntimeNames.get(ep.getRuntimeName())
+              ));
+            }
+          } catch (SpRuntimeException e) {
+            LOG.debug("Value and metadata for event field '{}' could not be extracted - "
+                    + "property will not be written to storage",
+                ep.getRuntimeName()
+            );
+          }
+        }
+      } catch (SpRuntimeException e) {
+        LOG.error("Event could not be converted to the IotDB representation: {}", e.getMessage());
+      }
+    });
+    return iotDbRecords;
+  }
+
+  /**
+   * Inserts measurement records into an IoTDB database as an aligned time series of one device.
+   *
+   * @param timestampValue The timestamp value associated with the measurement records.
+   * @param iotDbRecords   An ArrayList of IotDbMeasurementRecord objects containing the measurement records to be inserted.
+   */
+  private void insertIntoIotDb(long timestampValue, ArrayList<IotDbMeasurementRecord> iotDbRecords) {
+    try {
+      sessionPool.insertAlignedRecordsOfOneDevice(
+          "root.streampipes.%s".formatted(measure.getMeasureName()),
+          List.of(timestampValue),
+          List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::measurementName).toList()),
+          List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::dataType).toList()),
+          List.of(iotDbRecords.stream().map(IotDbMeasurementRecord::value).toList())
+      );
+    } catch (IoTDBConnectionException | StatementExecutionException e) {
+      LOG.error("Failed to write event to IoTDB  - {}", e.getMessage());
+    }
+  }
+}
diff --git a/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java
new file mode 100644
index 0000000000..bd6637be27
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDb.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.DataLakeMeasurementSanitizer;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+/**
+ * Ensures proper sanitization of data lake measurements with respect to Apache IoTDB specifics.
+ * <p>
+ * This includes both sanitizing the measurement name according to IoTDB's path specifications,
+ * because the measurement name is used as part of the path in IoTDB,
+ * and sanitizing the runtime names of the event properties so that they do not match any reserved keywords.
+ *
+ * @see <a href="https://iotdb.apache.org/UserGuide/latest/Basic-Concept/Data-Model-and-Terminology.html#path">IotDB Path Spec</a>
+ */
+public class DataLakeMeasurementSanitizerIotDb extends DataLakeMeasurementSanitizer {
+  public DataLakeMeasurementSanitizerIotDb(IStreamPipesClient client, DataLakeMeasure measure) {
+    super(client, measure);
+  }
+
+  @Override
+  protected void cleanDataLakeMeasure() throws SpRuntimeException {
+      measure.setMeasureName(new MeasureNameSanitizerIotDb().sanitize(measure.getMeasureName()));
+
+      measure.getEventSchema()
+             .getEventProperties()
+             .forEach(eventProperty -> eventProperty.setRuntimeName(
+                 new IotDbNameSanitizer().renameReservedKeywords(eventProperty.getRuntimeName()))
+             );
+  }
+
+  protected DataLakeMeasure getMeasure() {
+    return measure;
+  }
+}
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
similarity index 96%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
index 8d66978dc9..0f9c4cfe36 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizer.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizer.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
 
 public class IotDbNameSanitizer {
 
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
similarity index 98%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
index 5813cffce1..11c2530b2a 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/IotDbReservedKeywords.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbReservedKeywords.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
 
 import java.util.Arrays;
 import java.util.List;
diff --git a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
similarity index 97%
rename from streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java
rename to streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
index bc30a4ed2d..0cd93a45f9 100644
--- a/streampipes-ts-store-iotdb/src/main/java/org/apache/streampipes/ts/store/iotdb/MeasureNameSanitizerIotDb.java
+++ b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDb.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
 
 /**
  * Ensures that measurement names comply with IoTDB path specifications.
diff --git a/streampipes-data-explorer-management/pom.xml b/streampipes-data-explorer-management/pom.xml
index 9e1d6a9600..e82a032919 100644
--- a/streampipes-data-explorer-management/pom.xml
+++ b/streampipes-data-explorer-management/pom.xml
@@ -47,6 +47,11 @@
             <artifactId>streampipes-data-explorer-influx</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-iotdb</artifactId>
+            <version>0.97.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <properties>
diff --git a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
index 96ba5b70ae..8614c15b57 100644
--- a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
+++ b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/DataExplorerDispatcher.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.dataexplorer.management;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerManager;
 import org.apache.streampipes.dataexplorer.influx.DataExplorerManagerInflux;
+import org.apache.streampipes.dataexplorer.iotdb.DataExplorerManagerIotDb;
 
 public class DataExplorerDispatcher {
 
@@ -31,8 +32,8 @@ public class DataExplorerDispatcher {
     return switch (Environments.getEnvironment()
                                .getTsStorage()
                                .getValueOrDefault()) {
-      case SupportedDataExplorerStorages.INFLUX_DB -> DataExplorerManagerInflux.INSTANCE;
-      default -> DataExplorerManagerInflux.INSTANCE;
+      case SupportedDataExplorerStorages.IOT_DB -> new DataExplorerManagerIotDb();
+      default -> new DataExplorerManagerInflux();
     };
   }
 }
diff --git a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
index 70af654be8..1ee48e2acc 100644
--- a/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
+++ b/streampipes-data-explorer-management/src/main/java/org/apache/streampipes/dataexplorer/management/SupportedDataExplorerStorages.java
@@ -28,4 +28,5 @@ package org.apache.streampipes.dataexplorer.management;
  */
 public class SupportedDataExplorerStorages {
   public static final String INFLUX_DB = "influxdb";
+  public static final String IOT_DB = "iotdb";
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
new file mode 100644
index 0000000000..0c1026a627
--- /dev/null
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.query;
+
+import org.apache.streampipes.dataexplorer.api.IDataLakeMeasurementCounter;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.PropertyScope;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurementCounter {
+
+  private static final Logger LOG = LoggerFactory.getLogger(DataLakeMeasurementCounter.class);
+
+  protected final List<DataLakeMeasure> allMeasurements;
+  protected final List<String> measurementNames;
+
+  public DataLakeMeasurementCounter(List<DataLakeMeasure> allMeasurements,
+                                         List<String> measurementNames) {
+    this.allMeasurements = allMeasurements;
+    this.measurementNames = measurementNames;
+  }
+
+  @Override
+  public Map<String, Integer> countMeasurementSizes() {
+
+    // create async futures so that count queries can be executed parallel
+    Map<String, CompletableFuture<Integer>> countQueriesFutures = measurementNames.stream()
+        .map(this::getMeasure)
+        .collect(Collectors.toMap(
+            DataLakeMeasure::getMeasureName,
+            this::createQueryAsAsyncFuture)
+        );
+
+    return getQueryResults(countQueriesFutures);
+  }
+
+  /**
+   * Retrieves the {@link DataLakeMeasure} with the specified measure name from the collection of all measurements.
+   *
+   * @param measureName The name of the measure to retrieve.
+   * @return The DataLakeMeasure corresponding to the provided measure name, or null if no such measure is found.
+   */
+  private DataLakeMeasure getMeasure(String measureName) {
+    return allMeasurements
+        .stream()
+        .filter(m -> m.getMeasureName().equals(measureName))
+        .findFirst()
+        .orElse(null);
+  }
+
+  /**
+   * Retrieves the results of asynchronous count queries from the given CompletableFuture map.
+   *
+   * @param queryFutures A Map containing the futures of
+   *                     asynchronous count queries mapped by their respective keys.
+   * @return A Map representing the results of the queries, where each key corresponds to
+   *         a measure name and the value is the count result.
+   */
+  private Map<String, Integer> getQueryResults(Map<String, CompletableFuture<Integer>> queryFutures) {
+    Map<String, Integer> resultPerMeasure = new HashMap<>();
+    queryFutures.forEach((key, value) -> {
+      try {
+        resultPerMeasure.put(key, value.get());
+      } catch (InterruptedException | ExecutionException e) {
+        LOG.error("Async execution of count query failed: {}", e.getMessage());
+        resultPerMeasure.put(key, 0);
+      }
+    });
+
+    return resultPerMeasure;
+  }
+
+  private Integer getQueryResult(Map<String, CompletableFuture<Integer>> queryFuture) {
+    return null;
+  }
+
+  /**
+   * Retrieves the runtime name of the first measurement property from the event schema of the given DataLakeMeasure.
+   *
+   * @param measure The {@link DataLakeMeasure} from which to retrieve the first measurement property.
+   * @return The runtime name of the first measurement property, or null if no such property is found.
+   */
+  protected String getFirstMeasurementProperty(DataLakeMeasure measure) {
+    return measure.getEventSchema().getEventProperties()
+        .stream()
+        .filter(ep -> ep.getPropertyScope() != null
+            && ep.getPropertyScope().equals(PropertyScope.MEASUREMENT_PROPERTY.name()))
+        .map(EventProperty::getRuntimeName)
+        .findFirst()
+        .orElse(null);
+  }
+
+  protected Integer extractResult(SpQueryResult queryResult, String fieldName) {
+    return ((Double) (
+        queryResult.getAllDataSeries().get(0).getRows().get(0).get(queryResult.getHeaders().indexOf(fieldName)))
+    ).intValue();
+  }
+
+  /**
+   * Create the count query for the given DataLakeMeasure and return it as a {@link CompletableFuture}.
+   */
+  protected abstract CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure);
+}
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/exception/WorkerAdapterException.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/exception/WorkerAdapterException.java
index ef3ecfb34b..1cc848a686 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/exception/WorkerAdapterException.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/exception/WorkerAdapterException.java
@@ -18,24 +18,20 @@
 
 package org.apache.streampipes.extensions.api.connect.exception;
 
-import org.apache.streampipes.model.message.Message;
+import org.apache.streampipes.model.monitoring.SpLogMessage;
 
 public class WorkerAdapterException extends Exception {
-  private Message content;
 
-  public WorkerAdapterException() {
-  }
+  private SpLogMessage exceptionMessage;
 
-  public WorkerAdapterException(Message message) {
-    super(message.getElementName());
-    this.content = message;
+  public WorkerAdapterException() {
   }
 
-  public WorkerAdapterException(String message, Throwable cause) {
-    super(message, cause);
+  public WorkerAdapterException(SpLogMessage exceptionMessage) {
+    this.exceptionMessage = exceptionMessage;
   }
 
-  public Message getContent() {
-    return content;
+  public SpLogMessage getExceptionMessage() {
+    return exceptionMessage;
   }
 }
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
index 6f6a0477ae..06bd974c6c 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
@@ -60,8 +60,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import static org.apache.streampipes.sdk.helpers.EpProperties.timestampProperty;
@@ -77,6 +77,8 @@ public class Oi4Adapter implements StreamPipesAdapter {
   public static final String ID = "org.apache.streampipes.connect.iiot.adapters.oi4";
 
   private static final Logger LOG = LoggerFactory.getLogger(Oi4Adapter.class);
+  private static final long ReceiveSchemaSleepTime = 100;
+  private static final long ReceiveSchemaMaxTimeout = 5000;
 
   // Information about the topic structure can be found at page 57 of the above-mentioned development guide
   // The app id (missing here) needs to be provided by the user
@@ -204,13 +206,17 @@ public class Oi4Adapter implements StreamPipesAdapter {
       IAdapterParameterExtractor extractor,
       IAdapterGuessSchemaContext adapterGuessSchemaContext
   ) throws AdapterException {
-    this.applyConfiguration(extractor.getStaticPropertyExtractor());
+    try {
+      this.applyConfiguration(extractor.getStaticPropertyExtractor());
 
-    var sampleMessage = getSampleMessage();
-    var guessSchema = guessSchemaFromSampleMessage(sampleMessage);
-    updateTimestampPropertyIfExists(guessSchema);
+      var sampleMessage = getSampleMessage();
+      var guessSchema = guessSchemaFromSampleMessage(sampleMessage);
+      updateTimestampPropertyIfExists(guessSchema);
 
-    return guessSchema;
+      return guessSchema;
+    } catch (RuntimeException e) {
+      throw new AdapterException(e.getMessage(), e);
+    }
   }
 
   /**
@@ -256,23 +262,37 @@ public class Oi4Adapter implements StreamPipesAdapter {
     }
   }
 
-  private byte[] getSampleMessage() {
+  private byte[] getSampleMessage() throws AdapterException {
     List<byte[]> sampleMessages = new ArrayList<>();
+    long timeElapsed = 0;
+    AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
     MqttConsumer guessConsumer = getGuessMqttConsumer(sampleMessages);
 
-    var executor = Executors.newSingleThreadExecutor();
-    executor.submit(guessConsumer);
+    var thread = new Thread(guessConsumer);
+    thread.setUncaughtExceptionHandler((t, e) -> exceptionRef.set(e.getCause()));
+    thread.start();
 
-    while (sampleMessages.isEmpty()) {
+    while (sampleMessages.isEmpty() && exceptionRef.get() == null && timeElapsed < ReceiveSchemaMaxTimeout) {
       try {
-        TimeUnit.MILLISECONDS.sleep(100);
+        TimeUnit.MILLISECONDS.sleep(ReceiveSchemaSleepTime);
+        timeElapsed += ReceiveSchemaSleepTime;
       } catch (InterruptedException e) {
         LOG.error("Schema guessing failed during waiting for an incoming event: {}", e.getMessage());
+        break;
       }
     }
     guessConsumer.close();
-    executor.shutdown();
-    return sampleMessages.get(0);
+
+    Throwable threadException = exceptionRef.get();
+    if (threadException != null) {
+      throw new AdapterException(threadException.getMessage(), threadException);
+    }
+
+    if (!sampleMessages.isEmpty()) {
+      return sampleMessages.get(0);
+    } else {
+      throw new AdapterException("No messages received");
+    }
   }
 
   /**
@@ -293,58 +313,50 @@ public class Oi4Adapter implements StreamPipesAdapter {
         LOG.error("Error during parsing of incoming MQTT event.");
         throw new RuntimeException(e);
       }
-      try {
-        // Attempt to extract payload from the NetworkMessage
-        extractPayload(networkMessage);
-        // If successful, add the event to the sampleMessages array
-        sampleMessages.add(event);
-      } catch (ParseException e) {
-        LOG.debug("Collected sample could not be parsed successfully - "
-            + "this could be a message from a different sensor type.");
-      }
+      // Attempt to extract payload from the NetworkMessage
+      extractPayload(networkMessage);
+      // If successful, add the event to the sampleMessages array
+      sampleMessages.add(event);
     };
     return new MqttConsumer(this.mqttConfig, eventProcessor);
   }
 
-
   private Map<String, Object> extractPayload(NetworkMessage message) throws ParseException {
 
     var dataMessages = findProcessDataInputMessage(message);
 
     if (!dataMessages.isEmpty()) {
 
-      for (var dataMessage: dataMessages) {
+      for (var dataMessage : dataMessages) {
 
         var sensorId = getSensorIdFromSource(dataMessage.source());
 
         // Verify that the message corresponds to the designated sensor type.
         // This validation relies on the assumption that the source information includes the sensor type.
         if (dataMessage.source()
-                       .contains(givenSensorType)) {
+            .contains(givenSensorType)) {
 
           // an empty list of selected sensors means that we want to collect data from all sensors available
           if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
-
             return extractAndEnrichMessagePayload(dataMessage, sensorId);
           }
         }
       }
     }
-    throw new ParseException("No process data message");
+    throw new ParseException(String.format("No sensor of type %s found in message", givenSensorType));
   }
 
   private List<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
     return message.messages()
-                  .stream()
-                  .filter(msg -> msg.filter()
-                  .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
-                  .toList();
+        .stream()
+        .filter(msg -> msg.filter()
+            .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
+        .toList();
   }
 
   private Map<String, Object> extractAndEnrichMessagePayload(DataSetMessage dataSetMessage, String sensorId) {
     var payload = dataSetMessage
-        .payload()
-        .get(0);
+        .payload();
     try {
       payload.put(
           OI4AdapterLabels.EVENT_KEY_TIMESTAMP,
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
index f50dd2f76a..ca33d4053e 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
@@ -18,10 +18,12 @@
 
 package org.apache.streampipes.connect.iiot.adapters.oi4.model;
 
+import org.apache.streampipes.connect.iiot.adapters.oi4.serializer.PayloadDeserializer;
+
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -37,7 +39,7 @@ public record DataSetMessage(
     @JsonProperty("Timestamp") String timestamp,
     @JsonProperty("Filter") String filter,
     @JsonProperty("Source") String source,
-    @JsonProperty("Payload") List<Map<String, Object>> payload
+    @JsonProperty("Payload") @JsonDeserialize(using = PayloadDeserializer.class) Map<String, Object> payload
 ) {
 
 }
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java
similarity index 50%
rename from streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java
rename to streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java
index 9b852168fe..ce27d80dd8 100644
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/model/PeConfig.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java
@@ -16,49 +16,26 @@
  *
  */
 
-package org.apache.streampipes.svcdiscovery.api.model;
+package org.apache.streampipes.connect.iiot.adapters.oi4.serializer;
 
-import org.apache.streampipes.model.extensions.configuration.ConfigItem;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
 
-import java.util.List;
+import java.io.IOException;
 import java.util.Map;
 
-public class PeConfig {
-
-  String mainKey;
-  String name;
-  List<ConfigItem> configs;
-  Map<String, String> meta;
-
-  public List<ConfigItem> getConfigs() {
-    return configs;
-  }
-
-  public void setConfigs(List<ConfigItem> configs) {
-    this.configs = configs;
-  }
-
-  public Map<String, String> getMeta() {
-    return meta;
-  }
-
-  public void setMeta(Map<String, String> meta) {
-    this.meta = meta;
-  }
-
-  public String getMainKey() {
-    return mainKey;
-  }
-
-  public void setMainKey(String mainKey) {
-    this.mainKey = mainKey;
-  }
-
-  public String getName() {
-    return name;
-  }
-
-  public void setName(String name) {
-    this.name = name;
+public class PayloadDeserializer extends JsonDeserializer<Map<String, Object>> {
+
+  @Override
+  public Map<String, Object> deserialize(JsonParser jp, DeserializationContext ctx) throws IOException {
+    JsonNode node = jp.getCodec().readTree(jp);
+    if (node.isObject()) {
+      return jp.getCodec().treeToValue(node, Map.class);
+    } else if (node.isArray() && !node.isEmpty()) {
+      return jp.getCodec().treeToValue(node.get(0), Map.class);
+    }
+    throw new IllegalArgumentException("Unexpected JSON type for payload");
   }
 }
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
index 04f32643a3..dd0647c6d5 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
@@ -33,6 +33,7 @@ import org.apache.streampipes.extensions.management.connect.adapter.parser.xml.X
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 import org.apache.streampipes.model.connect.rules.schema.RenameRuleDescription;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
 import org.apache.streampipes.sdk.StaticProperties;
 import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
 import org.apache.streampipes.sdk.helpers.Alternatives;
@@ -40,7 +41,6 @@ import org.apache.streampipes.sdk.helpers.Filetypes;
 import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
 import org.apache.streampipes.sdk.helpers.Options;
-import org.apache.streampipes.sdk.utils.Assets;
 import org.apache.streampipes.vocabulary.SO;
 
 import org.slf4j.Logger;
@@ -91,7 +91,7 @@ public class FileReplayAdapter implements StreamPipesAdapter {
             new XmlParser(),
             new ImageParser()
         )
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
         .withLocales(Locales.EN)
         .withCategory(AdapterType.Generic)
         .requiredFile(
@@ -192,7 +192,7 @@ public class FileReplayAdapter implements StreamPipesAdapter {
    *
    * @param extractor An instance of {@code IAdapterParameterExtractor} used for extracting adapter parameters.
    * @throws AdapterException If there are multiple renaming rules detected affecting the timestamp property,
-   * indicating an invalid configuration.
+   *                          indicating an invalid configuration.
    */
   private void determineSourceTimestampField(IAdapterParameterExtractor extractor) throws AdapterException {
     var renamingRulesTimestamp = getRenamingRulesTimestamp(extractor);
@@ -245,23 +245,37 @@ public class FileReplayAdapter implements StreamPipesAdapter {
           .parse(inputStream, (event) -> {
 
             long actualEventTimestamp = -1;
-            if (event.get(timestampSourceFieldName) instanceof Long) {
-              actualEventTimestamp = (long) event.get(timestampSourceFieldName);
-            } else {
-              LOG.error(
-                  "The timestamp field is not a unix timestamp in ms. Value: %s"
-                      .formatted(event.get(timestampSourceFieldName)));
+            var timestampFieldValue = event.get(timestampSourceFieldName);
+            if (timestampFieldValue instanceof Long) {
+              actualEventTimestamp = (Long) timestampFieldValue;
+            } else if (timestampFieldValue instanceof Integer) {
+              actualEventTimestamp = (Integer) timestampFieldValue;
+            } else if (!(timestampFieldValue == null && replaceTimestamp)){
+              adapterRuntimeContext
+                  .getLogger()
+                  .error(
+                      new AdapterException("Timestamp field is not a unix timestamp in ms, skipping event. Value: %s"
+                          .formatted(event.get(timestampSourceFieldName))
+                      ));
+            }
+
+            if (actualEventTimestamp == -1 && !replaceTimestamp) {
+              // Do not emit any data if timestamp could not be processed
+              return;
             }
 
-            if (timestampLastEvent != -1) {
-              long sleepTime = (long) ((actualEventTimestamp - timestampLastEvent) / speedUp);
-              // speed up is set to Float.MAX_VALUE when user selected fastest option
-              if (sleepTime > 0 && speedUp != Float.MAX_VALUE) {
-                try {
-                  Thread.sleep(sleepTime);
-                } catch (InterruptedException e) {
-                  LOG.info("File stream adapter was stopped, the current replay is interuppted", e);
-                }
+            long sleepTime;
+            if (timestampLastEvent != -1 && actualEventTimestamp != -1) {
+              sleepTime = (long) ((actualEventTimestamp - timestampLastEvent) / speedUp);
+            } else {
+              sleepTime = 1;
+            }
+            // speed up is set to Float.MAX_VALUE when user selected fastest option
+            if (sleepTime > 0 && speedUp != Float.MAX_VALUE) {
+              try {
+                Thread.sleep(sleepTime);
+              } catch (InterruptedException e) {
+                LOG.info("File stream adapter was stopped, the current replay is interrupted", e);
               }
             }
 
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
index fccb004c9c..1a9bf14903 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
@@ -79,3 +79,4 @@ The output consists of all values from the selected sensor type. Below is an exa
     "Device status": 0,
     "timestamp": 1685525380729
 }
+```
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
index a03969e8fa..34f3bde459 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
@@ -26,7 +26,59 @@
 
 ## Description
 
-Continuously streams the content from a file
+The File Stream Adapter enables continuous streaming of file contents to Apache StreamPipes, creating a data stream for utilization within StreamPipes. It's particularly handy when you prefer not to connect directly to the data source via StreamPipes or for testing and demonstration purposes. Currently, it supports the following file types:
 
-***
+- CSV
+- JSON
+- XML
+
+### Example
+
+Suppose we have a CSV file (`temperature.csv`) containing data from a temperature sensor recording data every second:
+
+```text
+time,temperature
+1715593295000,36.3
+1715593296000,37.5
+1715593297000,37.0
+1715593298000,37.2
+1715593299000,37.2
+1715593210000,37.6
+1715593211000,37.4
+1715593212000,37.5
+1715593213000,37.5
+1715593214000,37.7
+```
+
+When creating a new File Stream Adapter:
+- Upload the file
+- Select `yes` for `Replay Once`
+- Choose `CSV` as the `Format` with `,` as the `delimiter`, check `Header`
+
+After creating the adapter, it will output one line of the CSV as an event every second.
+Further details on configuration options are provided below.
+
+---
+
+## Configuration
+
+### File
+
+This section determines the file to be streamed by the adapter. Options include:
+
+- `Choose existing file`: Select from files already present in StreamPipes.
+- `Upload new file`: Upload a new file, also available for other adapters. Supports `.csv`, `.json`, and `.xml` file types.
+
+### Overwrite file time
+Enable this option to always pass the current system time as the timestamp when emitting an event. If your file lacks timestamp information, this should be enabled. Conversely, if your file has timestamp information, enabling this option will overwrite it with the current system time. By default, this option is disabled, leaving timestamp information unaffected.
+
+### Replay Once
+Distinguishes between replaying all data contained in the file only once or in a loop until the adapter is manually stopped.
+If enabled, this will cause events from the file to be emitted multiple times. In this case, it is recommended to enable `Overwrite file time` if the resulting stream is to be persisted in StreamPipes, otherwise existing events with the same timestamp will be overwritten.
+
+### Replay Speed
 
+Configures the event frequency:
+- **Keep original time**: Events are emitted based on the timestamp information in the file. 
+- **Fastest**: All data in the file is replayed as quickly as possible, with no waiting time. 
+- **Speed Up Factor**: Adjusts the waiting time of the adapter based on the provided speed up factor, considering the time between two events in the file.
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
index 1f6fecb2d8..86a6a38d31 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
@@ -17,16 +17,16 @@
 
 
 org.apache.streampipes.connect.iiot.protocol.stream.file.title=File Stream
-org.apache.streampipes.connect.iiot.protocol.stream.file.description=Continuously streams the content from a file.
+org.apache.streampipes.connect.iiot.protocol.stream.file.description=allows you to continuously stream the contents of a file to Apache StreamPipes, creating a data stream that allows you to use the data within StreamPipes
 
 filePath.title=File
-filePath.description=File Path
+filePath.description=Choose between an already existing file or upload a new file
 
-replaceTimestamp.title=Use current time
-replaceTimestamp.description=Replace Event Time with Current Timestamp
+replaceTimestamp.title=Overwrite file time
+replaceTimestamp.description=Always use the current time as the timestamp when emitting the event, regardless of any time value in the file. Needs to be activated if file does not contain a timestamp.
 
 speed.title=Replay Speed
-speed.description=
+speed.description=Determine the event frequency.
 
 speedUp.title=Speed Up
 speedUp.description=original = 1; speedup 2x = 2; half speed  =  0.5
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
index 1399794a89..af9f02893a 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
@@ -33,20 +33,25 @@ import org.apache.streampipes.extensions.management.connect.adapter.parser.Parse
 import org.apache.streampipes.messaging.InternalEventProcessor;
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
 import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
 import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class MqttProtocol implements StreamPipesAdapter {
 
+  private static final Logger LOG = LoggerFactory.getLogger(MqttProtocol.class);
+
   public static final String ID = "org.apache.streampipes.connect.iiot.protocol.stream.mqtt";
 
-  private Thread thread;
   private MqttConsumer mqttConsumer;
   private MqttConfig mqttConfig;
 
@@ -63,7 +68,7 @@ public class MqttProtocol implements StreamPipesAdapter {
         .create(ID, 0, MqttProtocol::new)
         .withSupportedParsers(Parsers.defaultParsers())
         .withLocales(Locales.EN)
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
         .withCategory(AdapterType.Generic, AdapterType.Manufacturing)
         .requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel())
         .requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAlternativesOne(),
@@ -80,12 +85,10 @@ public class MqttProtocol implements StreamPipesAdapter {
     this.applyConfiguration(extractor.getStaticPropertyExtractor());
     this.mqttConsumer = new MqttConsumer(
         this.mqttConfig,
-        new BrokerEventProcessor(extractor.selectedParser(), (event) -> {
-          collector.collect(event);
-        })
+        new BrokerEventProcessor(extractor.selectedParser(), collector)
     );
 
-    thread = new Thread(this.mqttConsumer);
+    Thread thread = new Thread(this.mqttConsumer);
     thread.start();
   }
 
@@ -98,22 +101,36 @@ public class MqttProtocol implements StreamPipesAdapter {
   @Override
   public GuessSchema onSchemaRequested(IAdapterParameterExtractor extractor,
                                        IAdapterGuessSchemaContext adapterGuessSchemaContext) throws AdapterException {
-    this.applyConfiguration(extractor.getStaticPropertyExtractor());
-    List<byte[]> elements = new ArrayList<>();
-    InternalEventProcessor<byte[]> eventProcessor = elements::add;
+    try {
+      AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
+      this.applyConfiguration(extractor.getStaticPropertyExtractor());
+      List<byte[]> elements = new ArrayList<>();
+      InternalEventProcessor<byte[]> eventProcessor = elements::add;
 
-    MqttConsumer consumer = new MqttConsumer(this.mqttConfig, eventProcessor);
 
-    Thread thread = new Thread(consumer);
-    thread.start();
+      MqttConsumer consumer = new MqttConsumer(this.mqttConfig, eventProcessor);
 
-    while (consumer.getMessageCount() < 1) {
-      try {
-        TimeUnit.MILLISECONDS.sleep(100);
-      } catch (InterruptedException e) {
-        e.printStackTrace();
+      Thread thread = new Thread(consumer);
+      thread.setUncaughtExceptionHandler((t, e) -> exceptionRef.set(e.getCause()));
+      thread.start();
+
+      while (consumer.getMessageCount() < 1 && exceptionRef.get() == null) {
+        try {
+          TimeUnit.MILLISECONDS.sleep(100);
+        } catch (InterruptedException e) {
+          break;
+        }
+      }
+      consumer.close();
+
+      Throwable threadException = exceptionRef.get();
+      if (threadException != null) {
+        throw new AdapterException(threadException.getMessage(), threadException);
       }
+
+      return extractor.selectedParser().getGuessSchema(new ByteArrayInputStream(elements.get(0)));
+    } catch (Exception e) {
+      throw new AdapterException(e.getMessage(), e);
     }
-    return extractor.selectedParser().getGuessSchema(new ByteArrayInputStream(elements.get(0)));
   }
 }
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
index 48c45798c6..db81aab8ba 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
@@ -27,19 +27,21 @@ import org.fusesource.mqtt.client.Topic;
 
 public class MqttConsumer implements Runnable {
 
-  private InternalEventProcessor<byte[]> consumer;
+  private final InternalEventProcessor<byte[]> consumer;
   private boolean running;
   private int maxElementsToReceive = -1;
   private int messageCount = 0;
 
-  private MqttConfig mqttConfig;
+  private final MqttConfig mqttConfig;
 
-  public MqttConsumer(MqttConfig mqttConfig, InternalEventProcessor<byte[]> consumer) {
+  public MqttConsumer(MqttConfig mqttConfig,
+                      InternalEventProcessor<byte[]> consumer) {
     this.mqttConfig = mqttConfig;
     this.consumer = consumer;
   }
 
-  public MqttConsumer(MqttConfig mqttConfig, InternalEventProcessor<byte[]> consumer,
+  public MqttConsumer(MqttConfig mqttConfig,
+                      InternalEventProcessor<byte[]> consumer,
                       int maxElementsToReceive) {
     this(mqttConfig, consumer);
     this.maxElementsToReceive = maxElementsToReceive;
@@ -51,6 +53,7 @@ public class MqttConsumer implements Runnable {
     MQTT mqtt = new MQTT();
     try {
       mqtt.setHost(mqttConfig.getUrl());
+      mqtt.setConnectAttemptsMax(1);
       if (mqttConfig.getAuthenticated()) {
         mqtt.setUserName(mqttConfig.getUsername());
         mqtt.setPassword(mqttConfig.getPassword());
@@ -69,7 +72,7 @@ public class MqttConsumer implements Runnable {
       }
       connection.disconnect();
     } catch (Exception e) {
-      e.printStackTrace();
+      throw new RuntimeException("Error when receiving data from MQTT", e);
     }
   }
 
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
index b4f2449ac4..e948bf1e28 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/pom.xml
@@ -52,13 +52,18 @@
         <dependency>
             <groupId>org.apache.iotdb</groupId>
             <artifactId>iotdb-session</artifactId>
-            <!-- 0.13.3 is the only recommended production ready version for edge side (standalone) deployment -->
-            <!-- Should be worked for IoTDB version >= 0.13.0. -->
-            <version>0.13.3</version>
             <exclusions>
                 <exclusion>
-                    <groupId>ch.qos.logback</groupId>
-                    <artifactId>logback-classic</artifactId>
+                    <groupId>org.xerial.snappy</groupId>
+                    <artifactId>snappy-java</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.antlr</groupId>
+                    <artifactId>antlr4-runtime</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.github.luben</groupId>
+                    <artifactId>zstd-jni</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
index d326376c1a..44dd0dda0e 100644
--- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
+++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.sinks.databases.jvm.iotdb;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
 import org.apache.streampipes.model.graph.DataSinkDescription;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.runtime.field.AbstractField;
@@ -30,7 +31,6 @@ import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
 import org.apache.streampipes.sdk.helpers.EpRequirements;
 import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
 import org.apache.streampipes.wrapper.params.compat.SinkParams;
 import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
 
@@ -38,7 +38,6 @@ import org.apache.iotdb.rpc.IoTDBConnectionException;
 import org.apache.iotdb.rpc.StatementExecutionException;
 import org.apache.iotdb.session.pool.SessionPool;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.utils.Binary;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -77,7 +76,7 @@ public class IotDbSink extends StreamPipesDataSink {
     return DataSinkBuilder
         .create("org.apache.streampipes.sinks.databases.jvm.iotdb", 0)
         .withLocales(Locales.EN)
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON).
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON).
         category(DataSinkType.DATABASE)
         .requiredStream(
             StreamRequirementsBuilder.create()
@@ -169,7 +168,7 @@ public class IotDbSink extends StreamPipesDataSink {
         values.add(value);
       } else {
         types.add(TSDataType.TEXT);
-        values.add(Binary.valueOf(value.toString()));
+        values.add(value);
       }
     }
 
diff --git a/streampipes-messaging/src/main/java/org/apache/streampipes/messaging/SpProtocolManager.java b/streampipes-messaging/src/main/java/org/apache/streampipes/messaging/SpProtocolManager.java
index 35e2d9b6e6..279c522eb9 100644
--- a/streampipes-messaging/src/main/java/org/apache/streampipes/messaging/SpProtocolManager.java
+++ b/streampipes-messaging/src/main/java/org/apache/streampipes/messaging/SpProtocolManager.java
@@ -43,7 +43,6 @@ public enum SpProtocolManager {
   }
 
   public <T extends TransportProtocol> Optional<SpProtocolDefinition<T>> findDefinition(T transportProtocol) {
-    // TODO add RDF URI for protocol in model
     return this.availableProtocols
         .stream()
         .filter
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
index 742417ddd8..8f8c2cf2de 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineElementTemplateVisitor.java
@@ -39,6 +39,7 @@ import org.apache.streampipes.model.staticproperty.StaticPropertyAlternatives;
 import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
 import org.apache.streampipes.model.staticproperty.StaticPropertyVisitor;
 import org.apache.streampipes.model.util.Cloner;
+import org.apache.streampipes.user.management.encryption.SecretEncryptionManager;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -139,8 +140,16 @@ public class PipelineElementTemplateVisitor implements StaticPropertyVisitor {
   @Override
   public void visit(SecretStaticProperty secretStaticProperty) {
     if (hasKey(secretStaticProperty)) {
+      Map<String, Object> values = getAsMap(secretStaticProperty);
+      boolean encrypted = Boolean.parseBoolean(String.valueOf(values.get("encrypted")));
+      String value = String.valueOf(values.get("value"));
+      if (encrypted) {
+        secretStaticProperty.setValue(value);
+      } else {
+        String newValue = SecretEncryptionManager.encrypt(value);
+        secretStaticProperty.setValue(newValue);
+      }
       secretStaticProperty.setEncrypted(true);
-      secretStaticProperty.setValue(getAsString(secretStaticProperty));
     }
   }
 
diff --git a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/connect/GuessResource.java b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/connect/GuessResource.java
index 25adc18edc..42a177dfc5 100644
--- a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/connect/GuessResource.java
+++ b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/connect/GuessResource.java
@@ -24,7 +24,8 @@ import org.apache.streampipes.extensions.management.connect.GuessManagement;
 import org.apache.streampipes.extensions.management.context.AdapterContextGenerator;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
-import org.apache.streampipes.rest.shared.exception.SpMessageException;
+import org.apache.streampipes.model.monitoring.SpLogMessage;
+import org.apache.streampipes.rest.shared.exception.SpLogMessageException;
 import org.apache.streampipes.rest.shared.impl.AbstractSharedRestInterface;
 
 import org.slf4j.Logger;
@@ -65,11 +66,11 @@ public class GuessResource extends AbstractSharedRestInterface {
       return ok(result);
     } catch (ParseException e) {
       logger.error("Error while parsing events: ", e);
-      throw new SpMessageException(HttpStatus.INTERNAL_SERVER_ERROR, e);
+      throw new SpLogMessageException(HttpStatus.INTERNAL_SERVER_ERROR, SpLogMessage.from(e));
     } catch (AdapterException e) {
       logger.error("Error while guessing schema for AdapterDescription: {}, {}", adapterDescription.getElementId(),
           e.getMessage());
-      throw new SpMessageException(HttpStatus.INTERNAL_SERVER_ERROR, e);
+      throw new SpLogMessageException(HttpStatus.INTERNAL_SERVER_ERROR, SpLogMessage.from(e));
     }
 
   }
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/GuessResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/GuessResource.java
index f8756e587c..ddb51a9a92 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/GuessResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/GuessResource.java
@@ -63,7 +63,7 @@ public class GuessResource extends AbstractAdapterResource<GuessManagement> {
       LOG.error("Error while parsing events: ", e);
       return badRequest(SpLogMessage.from(e));
     } catch (WorkerAdapterException e) {
-      return serverError(SpLogMessage.from(e));
+      return serverError(e.getExceptionMessage());
     } catch (NoServiceEndpointsAvailableException | IOException e) {
       LOG.error(e.getMessage());
       return serverError(SpLogMessage.from(e));
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java
deleted file mode 100644
index 6ba4a2acd5..0000000000
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/SpConfig.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.svcdiscovery.api;
-
-import org.apache.streampipes.model.extensions.configuration.ConfigItem;
-import org.apache.streampipes.model.extensions.configuration.ConfigurationScope;
-
-public interface SpConfig {
-
-  <T> void register(String key, T defaultValue, String description, ConfigurationScope configurationScope);
-
-  void register(String key, boolean defaultValue, String description);
-
-  void register(String key, int defaultValue, String description);
-
-  void register(String key, double defaultValue, String description);
-
-  void register(String key, String defaultValue, String description);
-
-  void register(ConfigItem configItem);
-
-  void registerObject(String key, Object defaultValue, String description);
-
-  void registerPassword(String key, String defaultValue, String description);
-
-  boolean getBoolean(String key);
-
-  int getInteger(String key);
-
-  double getDouble(String key);
-
-  String getString(String key);
-
-  boolean exists(String key);
-
-  <T> T getObject(String key, Class<T> clazz, T defaultValue);
-
-  ConfigItem getConfigItem(String key);
-
-  void setBoolean(String key, Boolean value);
-
-  void setInteger(String key, int value);
-
-  void setDouble(String key, double value);
-
-  void setString(String key, String value);
-
-  void setObject(String key, Object value);
-
-}
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
index 0bdb9c63ed..ee57736951 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
@@ -31,6 +31,7 @@ import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistratio
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
 import org.apache.streampipes.rest.extensions.WelcomePage;
+import org.apache.streampipes.rest.shared.exception.RestResponseLogMessageExceptionHandler;
 import org.apache.streampipes.service.base.BaseNetworkingConfig;
 import org.apache.streampipes.service.base.StreamPipesServiceBase;
 import org.apache.streampipes.service.base.rest.ServiceHealthResource;
@@ -55,7 +56,12 @@ import java.util.stream.Collectors;
 
 @Configuration
 @EnableAutoConfiguration
-@Import({WebSecurityConfig.class, WelcomePage.class, ServiceHealthResource.class})
+@Import({
+    WebSecurityConfig.class,
+    WelcomePage.class,
+    ServiceHealthResource.class,
+    RestResponseLogMessageExceptionHandler.class
+})
 @ComponentScan({"org.apache.streampipes.rest.extensions.*", "org.apache.streampipes.service.base.rest.*"})
 public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServiceBase {
 
diff --git a/ui/cypress/README.md b/ui/cypress/README.md
index 075cbaeaef..faf4e25194 100644
--- a/ui/cypress/README.md
+++ b/ui/cypress/README.md
@@ -51,7 +51,7 @@ This folder contains a WIP framework for automated E2E tests of StreamPipes.
 
 ## Automated test runs
 
-- Each night the whole test suite is run within github actions
+- Each night the whole test suite is run within GitHub actions
   - See:
   - To add a test to the test suite add the suffix .spec.ts
 - Each PR triggers the smoke tests to detect errors before the branch is merged into the development
diff --git a/ui/cypress/support/general/login.ts b/ui/cypress/support/general/login.ts
index 13be5dc188..22452a5daa 100644
--- a/ui/cypress/support/general/login.ts
+++ b/ui/cypress/support/general/login.ts
@@ -19,7 +19,6 @@
 import '@angular/compiler';
 import { UserUtils } from '../utils/UserUtils';
 import { JwtHelperService } from '@auth0/angular-jwt';
-import { User } from '../model/User';
 
 declare global {
     namespace Cypress {
@@ -33,13 +32,10 @@ declare global {
     }
 }
 
-export const login = (user?: User) => {
-    let _user;
-    _user = !user ? UserUtils.adminUser : user;
-
+export const login = (user = UserUtils.adminUser) => {
     cy.request('POST', '/streampipes-backend/api/v2/auth/login', {
-        username: _user.email,
-        password: _user.password,
+        username: user.email,
+        password: user.password,
     }).then(res => {
         const decodedToken = new JwtHelperService({}).decodeToken(
             res.body.accessToken,
@@ -49,6 +45,5 @@ export const login = (user?: User) => {
             JSON.stringify(decodedToken.user),
         );
         window.localStorage.setItem('auth-token', res.body.accessToken);
-        console.log(user);
     });
 };
diff --git a/ui/cypress/support/general/switchUser.ts b/ui/cypress/support/general/switchUser.ts
deleted file mode 100644
index bb8612ac9f..0000000000
--- a/ui/cypress/support/general/switchUser.ts
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-import { User } from '../model/User';
-
-declare global {
-    namespace Cypress {
-        interface Chainable {
-            /**
-             * Logout and login as new user
-             * @example cy.switchUser();
-             */
-            switchUser: typeof switchUser;
-        }
-    }
-}
-
-export const switchUser = (user: User) => {
-    cy.logout();
-    cy.visit('#/login');
-    cy.dataCy('login-email').type(user.email);
-    cy.dataCy('login-password').type(user.password);
-    cy.dataCy('login-button').click();
-    cy.wait(1000);
-};
diff --git a/ui/cypress/support/utils/UserUtils.ts b/ui/cypress/support/utils/UserUtils.ts
index 9db82d6410..b91f046f07 100644
--- a/ui/cypress/support/utils/UserUtils.ts
+++ b/ui/cypress/support/utils/UserUtils.ts
@@ -27,7 +27,7 @@ export class UserUtils {
         .addRole(UserRole.ROLE_ADMIN)
         .build();
 
-    public static adapterAndPipelineAdminUser = UserBuilder.create(
+    public static userWithAdapterAndPipelineAdminRights = UserBuilder.create(
         'anpadmin@streampipes.apache.org',
     )
         .setName('anpadmin')
@@ -62,6 +62,15 @@ export class UserUtils {
         cy.dataCy('sp-element-edit-user-save').click();
     }
 
+    public static switchUser(user: User) {
+        cy.logout();
+        cy.visit('#/login');
+        cy.dataCy('login-email').type(user.email);
+        cy.dataCy('login-password').type(user.password);
+        cy.dataCy('login-button').click();
+        cy.wait(1000);
+    }
+
     public static deleteUser(user: User) {
         this.goToUserConfiguration();
 
diff --git a/ui/cypress/support/utils/connect/ConnectUtils.ts b/ui/cypress/support/utils/connect/ConnectUtils.ts
index 5a8a77a57c..407475c331 100644
--- a/ui/cypress/support/utils/connect/ConnectUtils.ts
+++ b/ui/cypress/support/utils/connect/ConnectUtils.ts
@@ -223,7 +223,9 @@ export class ConnectUtils {
             'be.visible',
         );
         if (switchUserCheck) {
-            cy.switchUser(UserUtils.adapterAndPipelineAdminUser);
+            UserUtils.switchUser(
+                UserUtils.userWithAdapterAndPipelineAdminRights,
+            );
         }
         this.checkAdapterAndAssociatedPipelinesDeleted();
     }
@@ -274,7 +276,9 @@ export class ConnectUtils {
         );
     }
 
-    public static setUpPreprocessingRuleTest(): AdapterInput {
+    public static setUpPreprocessingRuleTest(
+        overwriteTimestamp: boolean,
+    ): AdapterInput {
         const adapterConfiguration = AdapterBuilder.create('File_Stream')
             .setStoreInDataLake()
             .setTimestampProperty('timestamp')
@@ -287,8 +291,16 @@ export class ConnectUtils {
             .setName('Adapter to test rules')
             .setFormat('csv')
             .addFormatInput('input', ConnectBtns.csvDelimiter(), ';')
-            .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check')
-            .build();
+            .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check');
+
+        if (overwriteTimestamp) {
+            adapterConfiguration.addProtocolInput(
+                'checkbox',
+                'replaceTimestamp',
+                'check',
+            );
+        }
+        adapterConfiguration = adapterConfiguration.build();
 
         ConnectUtils.goToConnect();
         ConnectUtils.goToNewAdapterPage();
diff --git a/ui/src/app/services/tour/create-pipeline-tour.constants.ts b/ui/src/app/services/tour/create-pipeline-tour.constants.ts
index a755a79606..c99c9f5d69 100644
--- a/ui/src/app/services/tour/create-pipeline-tour.constants.ts
+++ b/ui/src/app/services/tour/create-pipeline-tour.constants.ts
@@ -40,7 +40,7 @@ export default {
                 stepId: 'step-3',
                 title: 'Pipeline assembly',
                 classes: 'shepherd shepherd-welcome',
-                text: '<p>The pipeline assembly is the place where you can create pipelines. <br>Pipelines combine data streams, processors and sinks and can be used to trigger alarms, to transform data or to apply data analytics algoritms.<br> Click <b>Next</b> to continue.</p>',
+                text: '<p>The pipeline assembly is the place where you can create pipelines. <br>Pipelines combine data streams, processors and sinks and can be used to trigger alarms, to transform data or to apply data analytics algorithms.<br> Click <b>Next</b> to continue.</p>',
                 buttons: ['cancel', 'next'],
             },
             {
@@ -68,7 +68,7 @@ export default {
             {
                 stepId: 'step-7',
                 title: 'Data processors',
-                text: 'No we will add a data processor. This is the data processor list which shows all currently available data processors.<p>Click <b>Next</b> to continue.</p>',
+                text: 'Now we will add a data processor. This is the data processor list which shows all currently available data processors.<p>Click <b>Next</b> to continue.</p>',
                 attachToElement: '#panel-1',
                 attachPosition: 'left',
                 buttons: ['cancel', 'next'],
diff --git a/ui/src/app/services/tour/dashboard-tour.constants.ts b/ui/src/app/services/tour/dashboard-tour.constants.ts
index 095d3c0610..c9a6f1cc83 100644
--- a/ui/src/app/services/tour/dashboard-tour.constants.ts
+++ b/ui/src/app/services/tour/dashboard-tour.constants.ts
@@ -24,13 +24,13 @@ export default {
                 stepId: 'step-1',
                 title: 'Welcome to the Dashboard Tutorial!',
                 classes: 'shepherd shepherd-welcome',
-                text: "<p>In this tutorial, you'll learn how to create a live dashboard that shows data from a running pipeline.</p><b>Note:</b>This tutorial requires a running pipeline that makes use of the <i>Dashboard Sink</i>. If you don't have such a pipeine started yet, go to the pipeline editor and create one.<p>Press <i>Next</i> to start the tutorial!</p>",
+                text: "<p>In this tutorial, you'll learn how to create a live dashboard that shows data from a running pipeline.</p><b>Note:</b>This tutorial requires a running pipeline that makes use of the <i>Dashboard Sink</i>. If you don't have such a pipeline started yet, go to the pipeline editor and create one.<p>Press <i>Next</i> to start the tutorial!</p>",
                 buttons: ['cancel', 'next'],
             },
             {
                 stepId: 'step-2',
                 title: 'Add visualization',
-                text: "<p>Let's start!</p><p>To create a new visulization, click the <i>Add visualization</i> button.</p>",
+                text: "<p>Let's start!</p><p>To create a new visualization, click the <i>Add visualization</i> button.</p>",
                 attachToElement: '#add-viz-button',
                 attachPosition: 'bottom',
                 buttons: ['cancel'],
@@ -46,7 +46,7 @@ export default {
             {
                 stepId: 'step-4',
                 title: 'Select visualization',
-                text: "<p>The second step requires the selection of a visulization. You can select from tables, gauges, maps or charts.</p><p>In this tutorial, we'll create a table visualization, so click the <i>Table</i> visualization on the left.</p>",
+                text: "<p>The second step requires the selection of a visualization. You can select from tables, gauges, maps or charts.</p><p>In this tutorial, we'll create a table visualization, so click the <i>Table</i> visualization on the left.</p>",
                 attachToElement: 'md-grid-tile:nth-of-type(1)',
                 attachPosition: 'right',
                 buttons: ['cancel'],


Test patch:
diff --git a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
similarity index 96%
rename from streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java
rename to streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
index 29d0aab2ae..5b7085784e 100644
--- a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbPropertyConverterTest.java
+++ b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/IotDbPropertyConverterTest.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb;
 
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
@@ -81,7 +81,7 @@ public class IotDbPropertyConverterTest {
   @Test
   public void convertPrimitivePropertyNumber() {
 
-    var property = new EventPropertyPrimitive(SO.NUMBER.toString(), "test", null, null);
+    var property = new EventPropertyPrimitive(SO.NUMBER, "test", null, null);
     var field = new PrimitiveField("test", "test", 5.24);
 
     var result = new IotDbPropertyConverter().convertPrimitiveProperty(property, field, "sanitizedTest");
diff --git a/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java
new file mode 100644
index 0000000000..12e491637d
--- /dev/null
+++ b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/DataLakeMeasurementSanitizerIotDbTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.dataexplorer.iotdb.sanitize;
+
+import org.apache.streampipes.client.api.IStreamPipesClient;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder;
+import org.apache.streampipes.test.generator.EventSchemaTestBuilder;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+public class DataLakeMeasurementSanitizerIotDbTest {
+
+  private IStreamPipesClient clientMock;
+
+  @BeforeEach
+  public void setUp() {
+    this.clientMock = mock(IStreamPipesClient.class);
+  }
+
+  @Test
+  public void cleanDataLakeMeasure() {
+    var eventSchema = EventSchemaTestBuilder.create()
+        .withEventProperty(
+            EventPropertyPrimitiveTestBuilder
+                .create()
+                .withRuntimeName("timestamp")
+                .build())
+        .withEventProperty(
+            EventPropertyPrimitiveTestBuilder
+                .create()
+                .withRuntimeName("all")
+                .build())
+        .withEventProperty(
+            EventPropertyPrimitiveTestBuilder
+                .create()
+                .withRuntimeName("pressure")
+                .build())
+        .build();
+    var measure = new DataLakeMeasure(
+        "invalid.Measure",
+        "s0::%s".formatted("timestamp"),
+        eventSchema
+    );
+    var sanitizer = new DataLakeMeasurementSanitizerIotDb(clientMock, measure);
+
+    sanitizer.cleanDataLakeMeasure();
+    var result = sanitizer.getMeasure();
+
+    assertEquals("invalid_Measure", result.getMeasureName());
+    assertEquals(3, result.getEventSchema().getEventProperties().size());
+    assertEquals("timestamp_", result.getEventSchema().getEventProperties().get(0).getRuntimeName());
+    assertEquals("all_", result.getEventSchema().getEventProperties().get(1).getRuntimeName());
+    assertEquals("pressure", result.getEventSchema().getEventProperties().get(2).getRuntimeName());
+  }
+}
diff --git a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
similarity index 95%
rename from streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java
rename to streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
index 16cbf865b2..5cc0a4ccdc 100644
--- a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/IotDbNameSanitizerTest.java
+++ b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/IotDbNameSanitizerTest.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
similarity index 91%
rename from streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
rename to streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
index 6a11f4dee6..886e924391 100644
--- a/streampipes-ts-store-iotdb/src/test/java/org/apache/streampipes/ts/store/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
+++ b/streampipes-data-explorer-iotdb/src/test/java/org/apache/streampipes/dataexplorer/iotdb/sanitize/MeasureNameSanitizerIotDbTest.java
@@ -16,9 +16,8 @@
  *
  */
 
-package org.apache.streampipes.ts.store.iotdb.sanitize;
+package org.apache.streampipes.dataexplorer.iotdb.sanitize;
 
-import org.apache.streampipes.ts.store.iotdb.MeasureNameSanitizerIotDb;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -30,6 +29,7 @@ public class MeasureNameSanitizerIotDbTest {
     var sanitizer = new MeasureNameSanitizerIotDb();
 
     assertEquals("myMeasure", sanitizer.sanitize("myMeasure"));
+    assertEquals("my_Measure", sanitizer.sanitize("my.Measure"));
     assertEquals("我的措施", sanitizer.sanitize("我的措施"));
     assertEquals("my_Measure", sanitizer.sanitize("my-Measure"));
     assertEquals("my_Measure_", sanitizer.sanitize("my-Measure?"));
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java
new file mode 100644
index 0000000000..d699fd8ad8
--- /dev/null
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.connect.iiot.adapters.oi4.model;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DataSetMessageDeserializationTest {
+
+  private String datasetMessageObjectPayload = """
+      {
+        "DataSetWriterId": 1,
+        "SequenceNumber": 123,
+        "Timestamp": "2024-05-10T07:28:01.711Z",
+        "Filter": "processDataInput",
+        "Source": "ABC",
+        "Payload": {
+          "Device status": 0,
+          "Level": 163,
+          "OUT1": false,
+          "OUT2": false
+        }
+      }
+      """;
+
+
+  private String datasetMessageArrayPayload = """
+      {
+        "DataSetWriterId": 1,
+        "SequenceNumber": 123,
+        "Timestamp": "2024-05-10T07:28:01.711Z",
+        "Filter": "processDataInput",
+        "Source": "ABC",
+        "Payload": [
+          {
+            "Device status": 0,
+            "Level": 163,
+            "OUT1": false,
+            "OUT2": false
+          }
+        ]
+      }
+      """;
+
+
+  @Test
+  public void testObjectDeserialization() throws JsonProcessingException {
+    var mapper = new ObjectMapper();
+    var deserialized = mapper.readValue(datasetMessageObjectPayload, DataSetMessage.class);
+    Assertions.assertEquals(4, deserialized.payload().size());
+  }
+
+  @Test
+  public void testArrayDeserialization() throws JsonProcessingException {
+    var mapper = new ObjectMapper();
+    var deserialized = mapper.readValue(datasetMessageArrayPayload, DataSetMessage.class);
+    Assertions.assertEquals(4, deserialized.payload().size());
+  }
+}
diff --git a/ui/cypress/support/e2e.ts b/ui/cypress/support/e2e.ts
index 561b34f7c0..c37c43ac26 100644
--- a/ui/cypress/support/e2e.ts
+++ b/ui/cypress/support/e2e.ts
@@ -19,7 +19,6 @@
 import { login } from './general/login';
 import { logout } from './general/logout';
 import { dataCy } from './general/dataCy';
-import { switchUser } from './general/switchUser';
 // tslint:disable-next-line:no-implicit-dependencies
 import 'cypress-file-upload';
 import { resetStreamPipes } from './general/resetStreamPipes';
@@ -29,7 +28,6 @@ import { removeDownloadDirectory } from './general/removeDownloadDirectory';
 // General commands
 Cypress.Commands.add('login', login);
 Cypress.Commands.add('logout', logout);
-Cypress.Commands.add('switchUser', switchUser);
 Cypress.Commands.add('dataCy', dataCy);
 Cypress.Commands.add('resetStreamPipes', resetStreamPipes);
 Cypress.Commands.add('initStreamPipesTest', initStreamPipesTest);
diff --git a/ui/cypress/tests/adapter/enhancedDeleteAdapter.smoke.spec.ts b/ui/cypress/tests/adapter/deleteAdapterWithMultipleUsers.smoke.spec.ts
similarity index 69%
rename from ui/cypress/tests/adapter/enhancedDeleteAdapter.smoke.spec.ts
rename to ui/cypress/tests/adapter/deleteAdapterWithMultipleUsers.smoke.spec.ts
index 6bb2cc3c59..29c854cb23 100644
--- a/ui/cypress/tests/adapter/enhancedDeleteAdapter.smoke.spec.ts
+++ b/ui/cypress/tests/adapter/deleteAdapterWithMultipleUsers.smoke.spec.ts
@@ -16,7 +16,6 @@
  *
  */
 
-import { AdapterBuilder } from '../../support/builder/AdapterBuilder';
 import { ConnectUtils } from '../../support/utils/connect/ConnectUtils';
 import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
 import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
@@ -25,11 +24,6 @@ import { PipelineUtils } from '../../support/utils/PipelineUtils';
 
 const adapterName = 'simulator';
 
-const adapterInput = AdapterBuilder.create('Machine_Data_Simulator')
-    .setName(adapterName)
-    .addInput('input', 'wait-time-ms', '1000')
-    .build();
-
 const pipelineInput = PipelineBuilder.create('Pipeline Test')
     .addSource(adapterName)
     .addProcessingElement(
@@ -46,50 +40,36 @@ const pipelineInput = PipelineBuilder.create('Pipeline Test')
     .build();
 
 describe('Test Enhanced Adapter Deletion', () => {
-    // In the beginning, create the non-admin user
-    before(() => {
-        cy.initStreamPipesTest();
-        UserUtils.addUser(UserUtils.adapterAndPipelineAdminUser);
-        cy.logout();
-    });
-
     beforeEach('Setup Test', () => {
-        cy.visit('#/login');
-        cy.dataCy('login-email').type(
-            UserUtils.adapterAndPipelineAdminUser.email,
-        );
-        cy.dataCy('login-password').type(
-            UserUtils.adapterAndPipelineAdminUser.password,
-        );
-        cy.dataCy('login-button').click();
-        cy.wait(1000);
+        cy.initStreamPipesTest();
+        UserUtils.addUser(UserUtils.userWithAdapterAndPipelineAdminRights);
     });
 
     it('Test Delete Adapter and Associated Pipelines', () => {
-        ConnectUtils.testAdapter(adapterInput);
-        PipelineUtils.addPipeline(pipelineInput);
+        ConnectUtils.addMachineDataSimulator(adapterName, false);
         PipelineUtils.addPipeline(pipelineInput);
+
         ConnectUtils.deleteAdapterAndAssociatedPipelines();
     });
 
     it('Test Admin Should Be Able to Delete Adapter and Not Owned Associated Pipelines', () => {
         // Let the user create the adapter and the pipeline
-        ConnectUtils.testAdapter(adapterInput);
-        PipelineUtils.addPipeline(pipelineInput);
+        ConnectUtils.addMachineDataSimulator(adapterName, false);
         PipelineUtils.addPipeline(pipelineInput);
+
         // Then let the admin delete them
-        cy.switchUser(UserUtils.adminUser);
+        UserUtils.switchUser(UserUtils.adminUser);
         ConnectUtils.deleteAdapterAndAssociatedPipelines(true);
     });
 
     it('Test Delete Adapter and Associated Pipelines Permission Denied', () => {
         // Let the admin create the adapter and the pipeline
-        cy.switchUser(UserUtils.adminUser);
-        ConnectUtils.testAdapter(adapterInput);
-        PipelineUtils.addPipeline(pipelineInput);
+        UserUtils.switchUser(UserUtils.adminUser);
+        ConnectUtils.addMachineDataSimulator(adapterName, false);
         PipelineUtils.addPipeline(pipelineInput);
+
         // Then the user shouldn't be able to delete them
-        cy.switchUser(UserUtils.adapterAndPipelineAdminUser);
+        UserUtils.switchUser(UserUtils.userWithAdapterAndPipelineAdminRights);
         ConnectUtils.deleteAdapterAndAssociatedPipelinesPermissionDenied();
     });
 });
diff --git a/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts b/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
index 3399ffd41b..39462cfc06 100644
--- a/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
+++ b/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
@@ -27,7 +27,8 @@ describe('Connect schema rule transformations', () => {
     });
 
     it('Perform Test', () => {
-        const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+        const adapterConfiguration =
+            ConnectUtils.setUpPreprocessingRuleTest(true);
 
         // Add static value to event
         ConnectEventSchemaUtils.addStaticProperty('staticPropertyName', 'id1');
diff --git a/ui/cypress/tests/adapter/rules/streamRules.spec.ts b/ui/cypress/tests/adapter/rules/streamRules.spec.ts
index 48867115d6..dbfab9dab7 100644
--- a/ui/cypress/tests/adapter/rules/streamRules.spec.ts
+++ b/ui/cypress/tests/adapter/rules/streamRules.spec.ts
@@ -27,7 +27,8 @@ describe('Connect aggregation rule transformations', () => {
     });
 
     it('Perform Test', () => {
-        const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+        const adapterConfiguration =
+            ConnectUtils.setUpPreprocessingRuleTest(false);
 
         ConnectEventSchemaUtils.markPropertyAsTimestamp('timestamp');
         ConnectEventSchemaUtils.finishEventSchemaConfiguration();
@@ -50,7 +51,8 @@ describe('Remove duplicates rule transformations', () => {
     });
 
     it('Perform Test', () => {
-        const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+        const adapterConfiguration =
+            ConnectUtils.setUpPreprocessingRuleTest(false);
 
         ConnectEventSchemaUtils.markPropertyAsTimestamp('timestamp');
         ConnectEventSchemaUtils.finishEventSchemaConfiguration();
diff --git a/ui/cypress/tests/adapter/rules/valueRules.ts b/ui/cypress/tests/adapter/rules/valueRules.ts
index d53de16de0..60ffdbf21a 100644
--- a/ui/cypress/tests/adapter/rules/valueRules.ts
+++ b/ui/cypress/tests/adapter/rules/valueRules.ts
@@ -27,7 +27,8 @@ describe('Connect value rule transformations', () => {
     });
 
     it('Perform Test', () => {
-        const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+        const adapterConfiguration =
+            ConnectUtils.setUpPreprocessingRuleTest(false);
 
         // Edit timestamp property
         ConnectEventSchemaUtils.editTimestampProperty(
diff --git a/ui/cypress/tests/userManagement/addUserAdmin.smoke.spec.ts b/ui/cypress/tests/userManagement/addUserAdmin.smoke.spec.ts
index be962555c4..91d5c788e2 100644
--- a/ui/cypress/tests/userManagement/addUserAdmin.smoke.spec.ts
+++ b/ui/cypress/tests/userManagement/addUserAdmin.smoke.spec.ts
@@ -50,12 +50,12 @@ describe('Test User Management', () => {
         );
 
         // Login as user
-        cy.switchUser(user);
+        UserUtils.switchUser(user);
 
         UserUtils.goToUserConfiguration();
         cy.dataCy('close-tutorial-button').click();
 
-        cy.switchUser(UserUtils.adminUser);
+        UserUtils.switchUser(UserUtils.adminUser);
         UserUtils.goToUserConfiguration();
 
         UserUtils.deleteUser(user);
diff --git a/ui/cypress/tests/userManagement/testGroupManagement.spec.ts b/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
index 1df9d793bf..86e78d4f79 100644
--- a/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
+++ b/ui/cypress/tests/userManagement/testGroupManagement.spec.ts
@@ -100,7 +100,7 @@ describe('Test Group Management for Pipelines', () => {
         cy.dataCy('sp-element-edit-user-save').click();
 
         // Login as first user which belongs to user group with pipeline admin role
-        cy.switchUser(user);
+        UserUtils.switchUser(user);
 
         cy.dataCy('navigation-icon', { timeout: 10000 }).should(
             'have.length',
@@ -119,7 +119,7 @@ describe('Test Group Management for Pipelines', () => {
         );
 
         // Login as user2
-        cy.switchUser(user2);
+        UserUtils.switchUser(user2);
 
         cy.dataCy('navigation-icon', { timeout: 10000 }).should(
             'have.length',
@@ -134,7 +134,7 @@ describe('Test Group Management for Pipelines', () => {
         );
 
         // Log in as admin and delete users
-        cy.switchUser(UserUtils.adminUser);
+        UserUtils.switchUser(UserUtils.adminUser);
         UserUtils.deleteUser(user);
         UserUtils.deleteUser(user2);
 
diff --git a/ui/cypress/tests/userManagement/testUserRoleConnect.spec.ts b/ui/cypress/tests/userManagement/testUserRoleConnect.spec.ts
index ce7f66a3cc..5b0dace368 100644
--- a/ui/cypress/tests/userManagement/testUserRoleConnect.spec.ts
+++ b/ui/cypress/tests/userManagement/testUserRoleConnect.spec.ts
@@ -37,7 +37,7 @@ describe('Test User Roles for Connect', () => {
         UserUtils.addUser(connect_admin);
 
         // Login as user and check if connect is visible to user
-        cy.switchUser(connect_admin);
+        UserUtils.switchUser(connect_admin);
 
         cy.dataCy('navigation-icon', { timeout: 10000 }).should(
             'have.length',
diff --git a/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts b/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
index 9da4a9dab7..6d4a9b9cb7 100644
--- a/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
+++ b/ui/cypress/tests/userManagement/testUserRolePipeline.spec.ts
@@ -79,7 +79,7 @@ describe('Test User Roles for Pipelines', () => {
         cy.dataCy('sp-element-edit-user-save').click();
 
         // Login as user and check if pipeline is visible to user
-        cy.switchUser(user);
+        UserUtils.switchUser(user);
 
         cy.dataCy('navigation-icon', { timeout: 10000 }).should(
             'have.length',
@@ -97,7 +97,7 @@ describe('Test User Roles for Pipelines', () => {
         );
 
         // Delete user
-        cy.switchUser(UserUtils.adminUser);
+        UserUtils.switchUser(UserUtils.adminUser);
         UserUtils.deleteUser(user);
     });
 });
diff --git a/ui/cypress/tests/userManagement/testVariousUserRoles.smoke.spec.ts b/ui/cypress/tests/userManagement/testVariousUserRoles.smoke.spec.ts
index d64bfef8d7..7f0049b87f 100644
--- a/ui/cypress/tests/userManagement/testVariousUserRoles.smoke.spec.ts
+++ b/ui/cypress/tests/userManagement/testVariousUserRoles.smoke.spec.ts
@@ -64,7 +64,7 @@ for (var i = 0; i < testedRoles.length; i++) {
             );
 
             // Login as user
-            cy.switchUser(user);
+            UserUtils.switchUser(user);
 
             // Check if every role displays correct navigation menu
             if (testRole == UserRole.ROLE_PIPELINE_ADMIN) {
@@ -95,7 +95,7 @@ for (var i = 0; i < testedRoles.length; i++) {
             }
 
             // Login as admin and delete user
-            cy.switchUser(UserUtils.adminUser);
+            UserUtils.switchUser(UserUtils.adminUser);
             UserUtils.deleteUser(user);
         });
     });