This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch fix-mqtt-oi4 in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 6c70f68ec19c65bfd504b9cf1e2139d0915c523b Author: Dominik Riemer <[email protected]> AuthorDate: Fri May 10 09:38:36 2024 +0200 fix: Add error handling to OI4 and MQTT adapters --- .../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 | 71 ++++++++++++++++++++ .../connectors/mqtt/adapter/MqttProtocol.java | 55 ++++++++++------ .../connectors/mqtt/shared/MqttConsumer.java | 13 ++-- .../rest/extensions/connect/GuessResource.java | 8 ++- .../rest/impl/connect/GuessResource.java | 2 +- .../StreamPipesExtensionsServiceBase.java | 4 +- 11 files changed, 225 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 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-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..4fbb0028ff --- /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,71 @@ +/* + * 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 = "{\n" + + " \"DataSetWriterId\": 1,\n" + + " \"SequenceNumber\": 123,\n" + + " \"Timestamp\": \"2024-05-10T07:28:01.711Z\",\n" + + " \"Filter\": \"processDataInput\",\n" + + " \"Source\": \"ABC\",\n" + + " \"Payload\": {\n" + + " \"Device status\": 0,\n" + + " \"Level\": 163,\n" + + " \"OUT1\": false,\n" + + " \"OUT2\": false\n" + + " }\n" + + " }"; + + + private String datasetMessageArrayPayload = "{\n" + + " \"DataSetWriterId\": 1,\n" + + " \"SequenceNumber\": 123,\n" + + " \"Timestamp\": \"2024-05-10T07:28:01.711Z\",\n" + + " \"Filter\": \"processDataInput\",\n" + + " \"Source\": \"ABC\",\n" + + " \"Payload\": [{\n" + + " \"Device status\": 0,\n" + + " \"Level\": 163,\n" + + " \"OUT1\": false,\n" + + " \"OUT2\": false\n" + + " }]\n" + + " }"; + + + @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..d81e8d9fc2 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,11 +24,13 @@ 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; import org.slf4j.LoggerFactory; + import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; @@ -65,11 +67,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..cf9a32925e 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; @@ -40,6 +41,7 @@ import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; @@ -55,7 +57,7 @@ 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 {
