Repository: apache/streampipes
Pull request #2842 (2024-05-09 07:18:17)

fix PipelineElementTemplateVisitor SecretStaticProperty value
<!--
  ~ 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.
-->

When the adapter uses templates, the Secret Static Property value should be encrypted

Now after importing the template, it becomes an object

`{
          "@class" : "org.apache.streampipes.model.staticproperty.SecretStaticProperty",
          "valueRequired" : false,
          "staticPropertyType" : "SecretStaticProperty",
          "index" : 1,
          "label" : "Password",
          "description" : "The password for the InfluxDB Server",
          "internalName" : "db_password",
          "predefined" : false,
          "value" : "{encrypted=false, value=test}",
          "encrypted" : true
 } `

![image](https://github.com/apache/streampipes/assets/89568107/15e01afd-c9ce-4ea1-8a85-51322b0d27d4)


### 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



Patch:
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-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());
+  }
+}