Repository: apache/streampipes
Pull request #2835 (2024-05-07 08:24:42)

feat: provide basic support for IoTDB as time series storage
<!--
  ~ 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 introduces very basic and experimental support for IoTDB as time series store for StreamPipes. By default, InfluxDB is still the time series storage of choice but by setting the following environment variables IoTDB can be used as storage.

- `SP_TS_STORAGE`: `iotdb`
- `SP_TS_STORAGE_HOST`: `<THE-HOST-OF-YOUR-IOTDB-INSTANCE>`
- `SP_TS_STORAGE_PORT`: `<THE-PORT-OF-YOUR-IOTDB-INSTANCE>`

These need to be set in the core and the extensions services.

> [!IMPORTANT]
> As of now this only supports persisting data into IoTDB via the Data Lake sink. Several other functionalities are not yet implemented and cause breakages, e.g., in the Data Explorer. These functionalities will be provided with upcoming PRs.

### 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/pom.xml b/pom.xml
index c2b603c2ec..f631832ea6 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>
@@ -389,6 +390,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 +831,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 +875,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>
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-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/DataExplorerManagerIotDb.java b/streampipes-data-explorer-iotdb/src/main/java/org/apache/streampipes/dataexplorer/iotdb/DataExplorerManagerIotDb.java
new file mode 100644
index 0000000000..5457ec166d
--- /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 null;
+  }
+
+  @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-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 98%
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..9772c39c44 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;
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-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-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java
new file mode 100644
index 0000000000..ce27d80dd8
--- /dev/null
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/serializer/PayloadDeserializer.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.connect.iiot.adapters.oi4.serializer;
+
+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.io.IOException;
+import java.util.Map;
+
+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-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-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/utils/connect/ConnectUtils.ts b/ui/cypress/support/utils/connect/ConnectUtils.ts
index 5a8a77a57c..cfe06bfc6a 100644
--- a/ui/cypress/support/utils/connect/ConnectUtils.ts
+++ b/ui/cypress/support/utils/connect/ConnectUtils.ts
@@ -274,7 +274,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 +289,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();


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/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(