This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch rel/0.95.0
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/rel/0.95.0 by this push:
new a2f8f3a0e7 fix: Add error handling to OI4 and MQTT adapters (#2848)
a2f8f3a0e7 is described below
commit a2f8f3a0e79ae062a35dfe3aac83ea036453e8f6
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri May 10 10:36:27 2024 +0200
fix: Add error handling to OI4 and MQTT adapters (#2848)
* fix: Add error handling to OI4 and MQTT adapters
* Fix checkstyle
* Fix checkstyle
* style: improve JSON strings
---------
Co-authored-by: bossenti <[email protected]>
---
.../management/management/GuessManagement.java | 6 +-
.../connect/exception/WorkerAdapterException.java | 18 ++---
.../connect/iiot/adapters/oi4/Oi4Adapter.java | 77 +++++++++++++---------
.../iiot/adapters/oi4/model/DataSetMessage.java | 6 +-
.../oi4/serializer/PayloadDeserializer.java | 41 ++++++++++++
.../model/DataSetMessageDeserializationTest.java | 77 ++++++++++++++++++++++
.../connectors/mqtt/adapter/MqttProtocol.java | 55 ++++++++++------
.../connectors/mqtt/shared/MqttConsumer.java | 13 ++--
.../rest/extensions/connect/GuessResource.java | 7 +-
.../rest/impl/connect/GuessResource.java | 2 +-
.../StreamPipesExtensionsServiceBase.java | 8 ++-
11 files changed, 234 insertions(+), 76 deletions(-)
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 104bae82e0..97ce035dae 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;
@@ -28,6 +27,7 @@ import
org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
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.monitoring.SpLogMessage;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -67,8 +67,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-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java
new file mode 100644
index 0000000000..d699fd8ad8
--- /dev/null
+++
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessageDeserializationTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.connect.iiot.adapters.oi4.model;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DataSetMessageDeserializationTest {
+
+ private String datasetMessageObjectPayload = """
+ {
+ "DataSetWriterId": 1,
+ "SequenceNumber": 123,
+ "Timestamp": "2024-05-10T07:28:01.711Z",
+ "Filter": "processDataInput",
+ "Source": "ABC",
+ "Payload": {
+ "Device status": 0,
+ "Level": 163,
+ "OUT1": false,
+ "OUT2": false
+ }
+ }
+ """;
+
+
+ private String datasetMessageArrayPayload = """
+ {
+ "DataSetWriterId": 1,
+ "SequenceNumber": 123,
+ "Timestamp": "2024-05-10T07:28:01.711Z",
+ "Filter": "processDataInput",
+ "Source": "ABC",
+ "Payload": [
+ {
+ "Device status": 0,
+ "Level": 163,
+ "OUT1": false,
+ "OUT2": false
+ }
+ ]
+ }
+ """;
+
+
+ @Test
+ public void testObjectDeserialization() throws JsonProcessingException {
+ var mapper = new ObjectMapper();
+ var deserialized = mapper.readValue(datasetMessageObjectPayload,
DataSetMessage.class);
+ Assertions.assertEquals(4, deserialized.payload().size());
+ }
+
+ @Test
+ public void testArrayDeserialization() throws JsonProcessingException {
+ var mapper = new ObjectMapper();
+ var deserialized = mapper.readValue(datasetMessageArrayPayload,
DataSetMessage.class);
+ Assertions.assertEquals(4, deserialized.payload().size());
+ }
+}
diff --git
a/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 81775f2271..bae43a0efa 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
@@ -30,6 +30,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;
@@ -54,7 +55,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 {
