Repository: apache/streampipes
Pull request #2846 (2024-05-10 03:00:09)

fix StatelessTransformationRuleGeneratorVisitor set rule mode
<!--
  ~ 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.
-->



![image](https://github.com/apache/streampipes/assets/89568107/6b60b0c9-7c47-4dda-a923-558eb770bcb8)


![image](https://github.com/apache/streampipes/assets/89568107/f5213e64-7127-4b80-b6b4-0ae98a7f8485)


### 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): <yes/no>
no
PR introduces (a) deprecation(s): <yes/no>
no

tenthe: Hi @yanfeng1992,

Thank you for providing the PR. The change looks good to me.
Can you briefly describe how to reproduce this error?
How did you configure the adapter, maybe we can create a test for this functionality.

Cheers,
Philipp

Patch:
diff --git a/streampipes-client-python/mkdocs.yml b/streampipes-client-python/mkdocs.yml
index 763ef120ec..81ece22708 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
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/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-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-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-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/src/app/services/tour/create-pipeline-tour.constants.ts b/ui/src/app/services/tour/create-pipeline-tour.constants.ts
index a755a79606..c99c9f5d69 100644
--- a/ui/src/app/services/tour/create-pipeline-tour.constants.ts
+++ b/ui/src/app/services/tour/create-pipeline-tour.constants.ts
@@ -40,7 +40,7 @@ export default {
                 stepId: 'step-3',
                 title: 'Pipeline assembly',
                 classes: 'shepherd shepherd-welcome',
-                text: '<p>The pipeline assembly is the place where you can create pipelines. <br>Pipelines combine data streams, processors and sinks and can be used to trigger alarms, to transform data or to apply data analytics algoritms.<br> Click <b>Next</b> to continue.</p>',
+                text: '<p>The pipeline assembly is the place where you can create pipelines. <br>Pipelines combine data streams, processors and sinks and can be used to trigger alarms, to transform data or to apply data analytics algorithms.<br> Click <b>Next</b> to continue.</p>',
                 buttons: ['cancel', 'next'],
             },
             {
@@ -68,7 +68,7 @@ export default {
             {
                 stepId: 'step-7',
                 title: 'Data processors',
-                text: 'No we will add a data processor. This is the data processor list which shows all currently available data processors.<p>Click <b>Next</b> to continue.</p>',
+                text: 'Now we will add a data processor. This is the data processor list which shows all currently available data processors.<p>Click <b>Next</b> to continue.</p>',
                 attachToElement: '#panel-1',
                 attachPosition: 'left',
                 buttons: ['cancel', 'next'],
diff --git a/ui/src/app/services/tour/dashboard-tour.constants.ts b/ui/src/app/services/tour/dashboard-tour.constants.ts
index 095d3c0610..c9a6f1cc83 100644
--- a/ui/src/app/services/tour/dashboard-tour.constants.ts
+++ b/ui/src/app/services/tour/dashboard-tour.constants.ts
@@ -24,13 +24,13 @@ export default {
                 stepId: 'step-1',
                 title: 'Welcome to the Dashboard Tutorial!',
                 classes: 'shepherd shepherd-welcome',
-                text: "<p>In this tutorial, you'll learn how to create a live dashboard that shows data from a running pipeline.</p><b>Note:</b>This tutorial requires a running pipeline that makes use of the <i>Dashboard Sink</i>. If you don't have such a pipeine started yet, go to the pipeline editor and create one.<p>Press <i>Next</i> to start the tutorial!</p>",
+                text: "<p>In this tutorial, you'll learn how to create a live dashboard that shows data from a running pipeline.</p><b>Note:</b>This tutorial requires a running pipeline that makes use of the <i>Dashboard Sink</i>. If you don't have such a pipeline started yet, go to the pipeline editor and create one.<p>Press <i>Next</i> to start the tutorial!</p>",
                 buttons: ['cancel', 'next'],
             },
             {
                 stepId: 'step-2',
                 title: 'Add visualization',
-                text: "<p>Let's start!</p><p>To create a new visulization, click the <i>Add visualization</i> button.</p>",
+                text: "<p>Let's start!</p><p>To create a new visualization, click the <i>Add visualization</i> button.</p>",
                 attachToElement: '#add-viz-button',
                 attachPosition: 'bottom',
                 buttons: ['cancel'],
@@ -46,7 +46,7 @@ export default {
             {
                 stepId: 'step-4',
                 title: 'Select visualization',
-                text: "<p>The second step requires the selection of a visulization. You can select from tables, gauges, maps or charts.</p><p>In this tutorial, we'll create a table visualization, so click the <i>Table</i> visualization on the left.</p>",
+                text: "<p>The second step requires the selection of a visualization. You can select from tables, gauges, maps or charts.</p><p>In this tutorial, we'll create a table visualization, so click the <i>Table</i> visualization on the left.</p>",
                 attachToElement: 'md-grid-tile:nth-of-type(1)',
                 attachPosition: 'right',
                 buttons: ['cancel'],


Test patch:
diff --git a/streampipes-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());
+  }
+}