You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2022/09/20 14:10:53 UTC
[incubator-streampipes] 01/02: [STREAMPIPES-585] Improve error handling for runtime exceptions
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch STREAMPIPES-585
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit a032126fb59f8c1e77e834f585b121956cd77c33
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Tue Sep 20 15:43:06 2022 +0200
[STREAMPIPES-585] Improve error handling for runtime exceptions
---
.../dataexplorer/commons/influx/InfluxStore.java | 287 +++++++++++----------
.../apache/streampipes/model/runtime/Event.java | 3 +-
.../routing/StandaloneSpInputCollector.java | 4 -
.../runtime/StandaloneEventProcessorRuntime.java | 40 +--
.../runtime/StandaloneEventSinkRuntime.java | 20 +-
.../StandaloneExternalEventProcessorRuntime.java | 2 +-
.../StandaloneExternalEventSinkRuntime.java | 2 +-
.../wrapper/routing/RawDataProcessor.java | 4 +-
8 files changed, 192 insertions(+), 170 deletions(-)
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
index d0bfb7928..fbbe3d7c7 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
+++ b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
@@ -42,158 +42,169 @@ import java.util.concurrent.TimeUnit;
public class InfluxStore {
- private static final Logger LOG = LoggerFactory.getLogger(InfluxStore.class);
-
- private InfluxDB influxDb = null;
- DataLakeMeasure measure;
-
- Map<String, String> targetRuntimeNames = new HashMap<>();
-
- public InfluxStore(DataLakeMeasure measure,
- SpConfig configStore) throws SpRuntimeException {
-
- this.measure = measure;
- InfluxConnectionSettings settings = InfluxConnectionSettings.from(configStore);
-
- // store sanitized target property runtime names in local variable
- measure.getEventSchema()
- .getEventProperties()
- .forEach(ep -> targetRuntimeNames.put(ep.getRuntimeName(),
- InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())));
-
- connect(settings);
+ private static final Logger LOG = LoggerFactory.getLogger(InfluxStore.class);
+
+ private InfluxDB influxDb = null;
+ DataLakeMeasure measure;
+
+ Map<String, String> targetRuntimeNames = new HashMap<>();
+
+ public InfluxStore(DataLakeMeasure measure,
+ SpConfig configStore) throws SpRuntimeException {
+
+ this.measure = measure;
+ InfluxConnectionSettings settings = InfluxConnectionSettings.from(configStore);
+
+ // store sanitized target property runtime names in local variable
+ measure.getEventSchema()
+ .getEventProperties()
+ .forEach(ep -> targetRuntimeNames.put(ep.getRuntimeName(),
+ InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())));
+
+ connect(settings);
+ }
+
+ /**
+ * Connects to the InfluxDB Server, sets the database and initializes the batch-behaviour
+ *
+ * @throws SpRuntimeException If not connection can be established or if the database could not
+ * be found
+ */
+ private void connect(InfluxConnectionSettings settings) throws SpRuntimeException {
+ // Connecting to the server
+ // "http://" must be in front
+ String urlAndPort = settings.getInfluxDbHost() + ":" + settings.getInfluxDbPort();
+ influxDb = InfluxDBFactory.connect(urlAndPort, settings.getUser(), settings.getPassword());
+
+ // Checking, if server is available
+ Pong response = influxDb.ping();
+ if (response.getVersion().equalsIgnoreCase("unknown")) {
+ throw new SpRuntimeException("Could not connect to InfluxDb Server: " + urlAndPort);
}
- /**
- * Connects to the InfluxDB Server, sets the database and initializes the batch-behaviour
- *
- * @throws SpRuntimeException If not connection can be established or if the database could not
- * be found
- */
- private void connect(InfluxConnectionSettings settings) throws SpRuntimeException {
- // Connecting to the server
- // "http://" must be in front
- String urlAndPort = settings.getInfluxDbHost() + ":" + settings.getInfluxDbPort();
- influxDb = InfluxDBFactory.connect(urlAndPort, settings.getUser(), settings.getPassword());
-
- // Checking, if server is available
- Pong response = influxDb.ping();
- if (response.getVersion().equalsIgnoreCase("unknown")) {
- throw new SpRuntimeException("Could not connect to InfluxDb Server: " + urlAndPort);
- }
-
- String databaseName = settings.getDatabaseName();
- // Checking whether the database exists
- if (!databaseExists(databaseName)) {
- LOG.info("Database '" + databaseName + "' not found. Gets created ...");
- createDatabase(databaseName);
- }
-
- // setting up the database
- influxDb.setDatabase(databaseName);
- int batchSize = 2000;
- int flushDuration = 500;
- influxDb.enableBatch(batchSize, flushDuration, TimeUnit.MILLISECONDS);
+ String databaseName = settings.getDatabaseName();
+ // Checking whether the database exists
+ if (!databaseExists(databaseName)) {
+ LOG.info("Database '" + databaseName + "' not found. Gets created ...");
+ createDatabase(databaseName);
}
- private boolean databaseExists(String dbName) {
- QueryResult queryResult = influxDb.query(new Query("SHOW DATABASES", ""));
- for (List<Object> a : queryResult.getResults().get(0).getSeries().get(0).getValues()) {
- if (a.get(0).equals(dbName)) {
- return true;
- }
- }
- return false;
+ // setting up the database
+ influxDb.setDatabase(databaseName);
+ int batchSize = 2000;
+ int flushDuration = 500;
+ influxDb.enableBatch(batchSize, flushDuration, TimeUnit.MILLISECONDS);
+ }
+
+ private boolean databaseExists(String dbName) {
+ QueryResult queryResult = influxDb.query(new Query("SHOW DATABASES", ""));
+ for (List<Object> a : queryResult.getResults().get(0).getSeries().get(0).getValues()) {
+ if (a.get(0).equals(dbName)) {
+ return true;
+ }
}
-
- /**
- * Creates a new database with the given name
- *
- * @param dbName The name of the database which should be created
- */
- private void createDatabase(String dbName) throws SpRuntimeException {
- if (!dbName.matches("^[a-zA-Z_]\\w*$")) {
- throw new SpRuntimeException(
- "Database name '" + dbName + "' not allowed. Allowed names: ^[a-zA-Z_][a-zA-Z0-9_]*$");
- }
- influxDb.query(new Query("CREATE DATABASE \"" + dbName + "\"", ""));
+ return false;
+ }
+
+ /**
+ * Creates a new database with the given name
+ *
+ * @param dbName The name of the database which should be created
+ */
+ private void createDatabase(String dbName) throws SpRuntimeException {
+ if (!dbName.matches("^[a-zA-Z_]\\w*$")) {
+ throw new SpRuntimeException(
+ "Database name '" + dbName + "' not allowed. Allowed names: ^[a-zA-Z_][a-zA-Z0-9_]*$");
+ }
+ influxDb.query(new Query("CREATE DATABASE \"" + dbName + "\"", ""));
+ }
+
+ /**
+ * Saves an event to the connected InfluxDB database
+ *
+ * @param event The event which should be saved
+ * @throws SpRuntimeException If the column name (key-value of the event map) is not allowed
+ */
+ public void onEvent(Event event) throws SpRuntimeException {
+ if (event == null) {
+ throw new SpRuntimeException("event is null");
}
- /**
- * Saves an event to the connected InfluxDB database
- *
- * @param event The event which should be saved
- * @throws SpRuntimeException If the column name (key-value of the event map) is not allowed
- */
- public void onEvent(Event event) throws SpRuntimeException {
- if (event == null) {
- throw new SpRuntimeException("event is null");
- }
-
- Long timestampValue = event.getFieldBySelector(measure.getTimestampField()).getAsPrimitive().getAsLong();
- Point.Builder p =
- Point.measurement(measure.getMeasureName()).time((long) timestampValue, TimeUnit.MILLISECONDS);
-
- for (EventProperty ep : measure.getEventSchema().getEventProperties()) {
- if (ep instanceof EventPropertyPrimitive) {
- String runtimeName = ep.getRuntimeName();
-
- // timestamp should not be added as a field
- if (!measure.getTimestampField().endsWith(runtimeName)) {
- String preparedRuntimeName = targetRuntimeNames.get(runtimeName);
- PrimitiveField eventPropertyPrimitiveField =
- event.getFieldByRuntimeName(runtimeName).getAsPrimitive();
-
- // store property as tag when the field is a dimension property
- if ("DIMENSION_PROPERTY".equals(ep.getPropertyScope())) {
- p.tag(preparedRuntimeName, eventPropertyPrimitiveField.getAsString());
- } else {
- try {
- // Store property according to property type
- String runtimeType = ((EventPropertyPrimitive) ep).getRuntimeType();
- if (XSD._integer.toString().equals(runtimeType)) {
- try {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsInt());
- } catch (NumberFormatException ef) {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
- }
- } else if (XSD._float.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
- } else if (XSD._double.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsDouble());
- } else if (XSD._boolean.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsBoolean());
- } else if (XSD._long.toString().equals(runtimeType)) {
- try {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsLong());
- } catch (NumberFormatException ef) {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
- }
- } else {
- p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsString());
- }
- } catch (NumberFormatException e) {
- LOG.warn("Wrong number format for field {}, ignoring.", preparedRuntimeName);
- }
+ Long timestampValue = event.getFieldBySelector(measure.getTimestampField()).getAsPrimitive().getAsLong();
+ Point.Builder p =
+ Point.measurement(measure.getMeasureName()).time((long) timestampValue, TimeUnit.MILLISECONDS);
+
+ for (EventProperty ep : measure.getEventSchema().getEventProperties()) {
+ if (ep instanceof EventPropertyPrimitive) {
+ String runtimeName = ep.getRuntimeName();
+
+ // timestamp should not be added as a field
+ if (!measure.getTimestampField().endsWith(runtimeName)) {
+ String preparedRuntimeName = targetRuntimeNames.get(runtimeName);
+
+ try {
+ PrimitiveField eventPropertyPrimitiveField = event.getFieldByRuntimeName(runtimeName).getAsPrimitive();
+ if (eventPropertyPrimitiveField.getRawValue() == null) {
+ LOG.warn("Field value for {} is null, ignoring value.", preparedRuntimeName);
+ } else {
+
+ // store property as tag when the field is a dimension property
+ if ("DIMENSION_PROPERTY".equals(ep.getPropertyScope())) {
+ p.tag(preparedRuntimeName, eventPropertyPrimitiveField.getAsString());
+ } else {
+
+ try {
+ // Store property according to property type
+ String runtimeType = ((EventPropertyPrimitive) ep).getRuntimeType();
+ if (XSD._integer.toString().equals(runtimeType)) {
+ try {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsInt());
+ } catch (NumberFormatException ef) {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
+ }
+ } else if (XSD._float.toString().equals(runtimeType)) {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
+ } else if (XSD._double.toString().equals(runtimeType)) {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsDouble());
+ } else if (XSD._boolean.toString().equals(runtimeType)) {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsBoolean());
+ } else if (XSD._long.toString().equals(runtimeType)) {
+ try {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsLong());
+ } catch (NumberFormatException ef) {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
}
+ } else {
+ p.addField(preparedRuntimeName, eventPropertyPrimitiveField.getAsString());
+ }
+ } catch (NumberFormatException e) {
+ LOG.warn("Wrong number format for field {}, ignoring.", preparedRuntimeName);
}
+ }
}
+ } catch (SpRuntimeException iae) {
+ LOG.warn("Field {} was missing in event and will be ignored", runtimeName, iae);
+ }
+
}
- influxDb.write(p.build());
+ }
}
- /**
- * Shuts down the connection to the InfluxDB server
- */
- public void close() throws SpRuntimeException {
- influxDb.flush();
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- throw new SpRuntimeException(e);
- }
- influxDb.close();
+ influxDb.write(p.build());
+ }
+
+ /**
+ * Shuts down the connection to the InfluxDB server
+ */
+ public void close() throws SpRuntimeException {
+ influxDb.flush();
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ throw new SpRuntimeException(e);
}
+ influxDb.close();
+ }
}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java b/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java
index 23143feed..84fd0101b 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.model.runtime;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.model.constants.PropertySelectorConstants;
import org.apache.streampipes.model.runtime.field.AbstractField;
import org.apache.streampipes.model.runtime.field.PrimitiveField;
@@ -71,7 +72,7 @@ public class Event {
.map(Map.Entry::getValue)
.filter(entry -> entry.getFieldNameIn().equals(runtimeName))
.findFirst()
- .orElseThrow(() -> new IllegalArgumentException("Field " + runtimeName + " not found"));
+ .orElseThrow(() -> new SpRuntimeException("Field " + runtimeName + " not found"));
}
public void removeFieldBySelector(String fieldSelector) {
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
index 47f095640..77277c6c1 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
@@ -49,11 +49,7 @@ public class StandaloneSpInputCollector<T extends TransportProtocol> extends
}
private void send(RawDataProcessor rawDataProcessor, byte[] event) {
- try {
rawDataProcessor.process(dataFormatDefinition.toMap(event), topic);
- } catch (SpRuntimeException e) {
- e.printStackTrace();
- }
}
@Override
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
index 196b3969d..e297d473d 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
@@ -27,13 +27,17 @@ import org.apache.streampipes.wrapper.routing.SpInputCollector;
import org.apache.streampipes.wrapper.routing.SpOutputCollector;
import org.apache.streampipes.wrapper.runtime.EventProcessor;
import org.apache.streampipes.wrapper.standalone.manager.ProtocolManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.function.Supplier;
public class StandaloneEventProcessorRuntime<B extends EventProcessorBindingParams> extends
- StandalonePipelineElementRuntime<B, DataProcessorInvocation,
- EventProcessorRuntimeParams<B>, EventProcessorRuntimeContext, EventProcessor<B>> {
+ StandalonePipelineElementRuntime<B, DataProcessorInvocation,
+ EventProcessorRuntimeParams<B>, EventProcessorRuntimeContext, EventProcessor<B>> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StandaloneEventProcessorRuntime.class);
public StandaloneEventProcessorRuntime(Supplier<EventProcessor<B>> supplier,
EventProcessorRuntimeParams<B> params) {
@@ -43,18 +47,18 @@ public class StandaloneEventProcessorRuntime<B extends EventProcessorBindingPara
public SpOutputCollector getOutputCollector() throws SpRuntimeException {
return ProtocolManager.findOutputCollector(
- params
- .getBindingParams()
- .getGraph()
- .getOutputStream()
- .getEventGrounding()
- .getTransportProtocol(),
- params.getBindingParams()
- .getGraph()
- .getOutputStream()
- .getEventGrounding()
- .getTransportFormats()
- .get(0));
+ params
+ .getBindingParams()
+ .getGraph()
+ .getOutputStream()
+ .getEventGrounding()
+ .getTransportProtocol(),
+ params.getBindingParams()
+ .getGraph()
+ .getOutputStream()
+ .getEventGrounding()
+ .getTransportFormats()
+ .get(0));
}
@Override
@@ -65,8 +69,12 @@ public class StandaloneEventProcessorRuntime<B extends EventProcessorBindingPara
}
@Override
- public void process(Map<String, Object> rawEvent, String sourceInfo) throws SpRuntimeException {
- getEngine().onEvent(params.makeEvent(rawEvent, sourceInfo), getOutputCollector());
+ public void process(Map<String, Object> rawEvent, String sourceInfo) {
+ try {
+ engine.onEvent(params.makeEvent(rawEvent, sourceInfo), getOutputCollector());
+ } catch (RuntimeException e) {
+ LOG.error("RuntimeException while processing event in {}", engine.getClass().getCanonicalName(), e);
+ }
}
@Override
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
index 75b0b8e79..c1fd982ce 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
@@ -25,16 +25,20 @@ import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
import org.apache.streampipes.wrapper.params.runtime.EventSinkRuntimeParams;
import org.apache.streampipes.wrapper.routing.SpInputCollector;
import org.apache.streampipes.wrapper.runtime.EventSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.function.Supplier;
public class StandaloneEventSinkRuntime<B extends EventSinkBindingParams> extends
- StandalonePipelineElementRuntime<B, DataSinkInvocation,
- EventSinkRuntimeParams<B>, EventSinkRuntimeContext, EventSink<B>> {
+ StandalonePipelineElementRuntime<B, DataSinkInvocation,
+ EventSinkRuntimeParams<B>, EventSinkRuntimeContext, EventSink<B>> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StandaloneEventSinkRuntime.class);
public StandaloneEventSinkRuntime(Supplier<EventSink<B>> supplier, EventSinkRuntimeParams<B>
- params) {
+ params) {
super(supplier, params);
}
@@ -46,8 +50,12 @@ public class StandaloneEventSinkRuntime<B extends EventSinkBindingParams> extend
}
@Override
- public void process(Map<String, Object> rawEvent, String sourceInfo) throws SpRuntimeException {
- engine.onEvent(params.makeEvent(rawEvent, sourceInfo));
+ public void process(Map<String, Object> rawEvent, String sourceInfo) {
+ try {
+ engine.onEvent(params.makeEvent(rawEvent, sourceInfo));
+ } catch (RuntimeException e) {
+ LOG.error("RuntimeException while processing event in {}", engine.getClass().getCanonicalName() , e);
+ }
}
@Override
@@ -66,7 +74,7 @@ public class StandaloneEventSinkRuntime<B extends EventSinkBindingParams> extend
@Override
public void postDiscard() throws SpRuntimeException {
- for(SpInputCollector spInputCollector : getInputCollectors()) {
+ for (SpInputCollector spInputCollector : getInputCollectors()) {
spInputCollector.disconnect();
}
}
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventProcessorRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventProcessorRuntime.java
index a2177f246..01c01878d 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventProcessorRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventProcessorRuntime.java
@@ -42,7 +42,7 @@ public class StandaloneExternalEventProcessorRuntime<B extends EventProcessorBin
}
@Override
- public void process(Map<String, Object> rawEvent, String sourceInfo) throws SpRuntimeException {
+ public void process(Map<String, Object> rawEvent, String sourceInfo) {
}
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventSinkRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventSinkRuntime.java
index 3fb06e210..418702da1 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventSinkRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneExternalEventSinkRuntime.java
@@ -41,7 +41,7 @@ public class StandaloneExternalEventSinkRuntime<B extends EventSinkBindingParams
}
@Override
- public void process(Map<String, Object> rawEvent, String sourceInfo) throws SpRuntimeException {
+ public void process(Map<String, Object> rawEvent, String sourceInfo) {
}
diff --git a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/routing/RawDataProcessor.java b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/routing/RawDataProcessor.java
index 648b9b5e4..bb4117c26 100644
--- a/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/routing/RawDataProcessor.java
+++ b/streampipes-wrapper/src/main/java/org/apache/streampipes/wrapper/routing/RawDataProcessor.java
@@ -17,11 +17,9 @@
*/
package org.apache.streampipes.wrapper.routing;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-
import java.util.Map;
public interface RawDataProcessor {
- void process(Map<String, Object> rawEvent, String sourceInfo) throws SpRuntimeException;
+ void process(Map<String, Object> rawEvent, String sourceInfo);
}