You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2022/09/20 19:10:36 UTC
[incubator-streampipes] branch dev updated (24f60b733 -> 5d1945b02)
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a change to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
from 24f60b733 Merge branch 'dev' of github.com:apache/incubator-streampipes into dev
new e45f577d4 [STREAMPIPES-585] Improve error handling for runtime exceptions
new 5d1945b02 [STREAMPIPES-585] Minor refactoring in InfluxStore
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../dataexplorer/commons/influx/InfluxStore.java | 284 +++++++++++----------
.../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, 196 insertions(+), 163 deletions(-)
[incubator-streampipes] 02/02: [STREAMPIPES-585] Minor refactoring in InfluxStore
Posted by ri...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 5d1945b02b3fee79c6abf330e74076f57da82490
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Tue Sep 20 16:08:12 2022 +0200
[STREAMPIPES-585] Minor refactoring in InfluxStore
---
.../dataexplorer/commons/influx/InfluxStore.java | 83 ++++++++++++----------
1 file changed, 47 insertions(+), 36 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 fbbe3d7c7..429f62eda 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
@@ -24,6 +24,7 @@ import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.field.PrimitiveField;
import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.svcdiscovery.api.SpConfig;
import org.apache.streampipes.vocabulary.XSD;
import org.influxdb.InfluxDB;
@@ -47,7 +48,7 @@ public class InfluxStore {
private InfluxDB influxDb = null;
DataLakeMeasure measure;
- Map<String, String> targetRuntimeNames = new HashMap<>();
+ Map<String, String> sanitizedRuntimeNames = new HashMap<>();
public InfluxStore(DataLakeMeasure measure,
SpConfig configStore) throws SpRuntimeException {
@@ -58,7 +59,7 @@ public class InfluxStore {
// store sanitized target property runtime names in local variable
measure.getEventSchema()
.getEventProperties()
- .forEach(ep -> targetRuntimeNames.put(ep.getRuntimeName(),
+ .forEach(ep -> sanitizedRuntimeNames.put(ep.getRuntimeName(),
InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())));
connect(settings);
@@ -131,7 +132,7 @@ public class InfluxStore {
}
Long timestampValue = event.getFieldBySelector(measure.getTimestampField()).getAsPrimitive().getAsLong();
- Point.Builder p =
+ Point.Builder point =
Point.measurement(measure.getMeasureName()).time((long) timestampValue, TimeUnit.MILLISECONDS);
for (EventProperty ep : measure.getEventSchema().getEventProperties()) {
@@ -140,46 +141,23 @@ public class InfluxStore {
// timestamp should not be added as a field
if (!measure.getTimestampField().endsWith(runtimeName)) {
- String preparedRuntimeName = targetRuntimeNames.get(runtimeName);
+ String sanitizedRuntimeName = sanitizedRuntimeNames.get(runtimeName);
try {
PrimitiveField eventPropertyPrimitiveField = event.getFieldByRuntimeName(runtimeName).getAsPrimitive();
if (eventPropertyPrimitiveField.getRawValue() == null) {
- LOG.warn("Field value for {} is null, ignoring value.", preparedRuntimeName);
+ LOG.warn("Field value for {} is null, ignoring value.", sanitizedRuntimeName);
} else {
// store property as tag when the field is a dimension property
- if ("DIMENSION_PROPERTY".equals(ep.getPropertyScope())) {
- p.tag(preparedRuntimeName, eventPropertyPrimitiveField.getAsString());
+ if (PropertyScope.DIMENSION_PROPERTY.name().equals(ep.getPropertyScope())) {
+ point.tag(sanitizedRuntimeName, 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);
- }
+ handleMeasurementProperty(
+ point,
+ (EventPropertyPrimitive) ep,
+ sanitizedRuntimeName,
+ eventPropertyPrimitiveField);
}
}
} catch (SpRuntimeException iae) {
@@ -191,7 +169,40 @@ public class InfluxStore {
}
}
- influxDb.write(p.build());
+ influxDb.write(point.build());
+ }
+
+ private void handleMeasurementProperty(Point.Builder p,
+ EventPropertyPrimitive ep,
+ String preparedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField) {
+ try {
+ // Store property according to property type
+ String runtimeType = 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);
+ }
}
/**
[incubator-streampipes] 01/02: [STREAMPIPES-585] Improve error handling for runtime exceptions
Posted by ri...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit e45f577d404a1378a7fba61f57d42cdd9a6588da
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);
}