Repository: apache/streampipes
Pull request #2848 (2024-05-10 07:40:50)

fix: Add error handling to OI4 and MQTT adapters
<!--
  ~ 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.
-->

Add some improvements to the MQTT and OI4 adapters:
* Error messages don't get lost and are sent to the UI
* Avoid infinite loops in case schema retrieval fails
* Support two different formats as OI4 payload

### 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>

PR introduces (a) deprecation(s): <yes/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..52d0b41fc1 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 < ReceiveSchemaSleepTime) {
       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,40 +313,36 @@ 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);
           }
+        } else {
+          throw new ParseException(String.format("No sensor of type %s found in message", givenSensorType));
         }
       }
     }
@@ -335,16 +351,15 @@ public class Oi4Adapter implements StreamPipesAdapter {
 
   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-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 {
 


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