Make handleEntity throw exceptions instead of returning empty optional

src/main/java/edu/ie3/datamodel/io/processor/EntityProcessor.java

@@ -51,7 +51,7 @@ protected EntityProcessor(Class<? extends T> registeredClass) {
    * @return an optional Map with fieldName to fieldValue or an empty optional if an error occurred
    *     during processing
    */
-  public Optional<LinkedHashMap<String, String>> handleEntity(T entity) {
+  public LinkedHashMap<String, String> handleEntity(T entity) {
     if (!registeredClass.equals(entity.getClass()))
       throw new EntityProcessorException(
           "Cannot process "
@@ -63,10 +63,10 @@ public Optional<LinkedHashMap<String, String>> handleEntity(T entity) {
               + ".class!");
 
     try {
-      return Optional.of(processObject(entity, fieldNameToMethod));
+      return processObject(entity, fieldNameToMethod);
     } catch (EntityProcessorException e) {
       logger.error("Cannot process the entity{}.", entity, e);
-      return Optional.empty();
+      throw new EntityProcessorException("Entity " + entity + " cannot be processed.", e);
     }
   }
 

src/main/java/edu/ie3/datamodel/io/processor/ProcessorProvider.java

@@ -70,13 +70,14 @@ public ProcessorProvider(
     this.timeSeriesProcessors = timeSeriesProcessors;
   }
 
-  public <T extends UniqueEntity> Optional<LinkedHashMap<String, String>> handleEntity(T entity) {
+  public <T extends UniqueEntity> LinkedHashMap<String, String> handleEntity(T entity)
+      throws ProcessorProviderException {
     try {
       EntityProcessor<? extends UniqueEntity> processor = getEntityProcessor(entity.getClass());
       return castProcessor(processor).handleEntity(entity);
     } catch (ProcessorProviderException e) {
       log.error("Exception occurred during entity handling.", e);
-      return Optional.empty();
+      throw e;
     }
   }
 
@@ -112,17 +113,19 @@ private EntityProcessor<? extends UniqueEntity> getEntityProcessor(
    * @return A set of mappings from field name to value
    */
   public <T extends TimeSeries<E, V>, E extends TimeSeriesEntry<V>, V extends Value>
-      Optional<Set<LinkedHashMap<String, String>>> handleTimeSeries(T timeSeries) {
+      Set<LinkedHashMap<String, String>> handleTimeSeries(T timeSeries)
+          throws ProcessorProviderException {
     TimeSeriesProcessorKey key = new TimeSeriesProcessorKey(timeSeries);
     try {
       TimeSeriesProcessor<T, E, V> processor = getTimeSeriesProcessor(key);
-      return Optional.of(processor.handleTimeSeries(timeSeries));
+      return processor.handleTimeSeries(timeSeries);
     } catch (ProcessorProviderException e) {
       log.error("Cannot handle the time series '{}'.", timeSeries, e);
-      return Optional.empty();
+      throw new ProcessorProviderException(
+          "Cannot handle the time series {" + timeSeries + "}.", e);
     } catch (EntityProcessorException e) {
       log.error("Error during processing of time series.", e);
-      return Optional.empty();
+      throw new EntityProcessorException("Error during processing of time series.", e);
     }
   }
 

src/main/java/edu/ie3/datamodel/io/processor/timeseries/TimeSeriesProcessor.java

@@ -176,7 +176,7 @@ private SortedMap<String, FieldSourceToMethod> buildFieldToSource(
   }
 
   @Override
-  public Optional<LinkedHashMap<String, String>> handleEntity(TimeSeries entity) {
+  public LinkedHashMap<String, String> handleEntity(TimeSeries entity) {
     throw new UnsupportedOperationException(
         "Don't invoke this simple method, but TimeSeriesProcessor#handleTimeSeries(TimeSeries).");
   }

src/main/java/edu/ie3/datamodel/io/sink/CsvFileSink.java

@@ -264,23 +264,10 @@ public <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
   }
 
   private <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
-      TimeSeries<E, V> timeSeries, BufferedCsvWriter writer) {
-    TimeSeriesProcessorKey key = new TimeSeriesProcessorKey(timeSeries);
-
+      TimeSeries<E, V> timeSeries, BufferedCsvWriter writer) throws ProcessorProviderException {
     try {
       Set<LinkedHashMap<String, String>> entityFieldData =
-          processorProvider
-              .handleTimeSeries(timeSeries)
-              .orElseThrow(
-                  () ->
-                      new SinkException(
-                          "Cannot persist time series of combination '"
-                              + key
-                              + "'. This sink can only process the following combinations: ["
-                              + processorProvider.getRegisteredTimeSeriesCombinations().stream()
-                                  .map(TimeSeriesProcessorKey::toString)
-                                  .collect(Collectors.joining(","))
-                              + "]"));
+          processorProvider.handleTimeSeries(timeSeries);
       entityFieldData.forEach(
           data -> {
             try {
@@ -291,8 +278,9 @@ private <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
               log.error("Exception occurred during processing the provided data fields: ", e);
             }
           });
-    } catch (SinkException e) {
+    } catch (ProcessorProviderException e) {
       log.error("Exception occurred during processor request: ", e);
+      throw new ProcessorProviderException("Exception occurred during processor request: ", e);
     }
   }
 
@@ -306,21 +294,7 @@ private <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
   private <C extends UniqueEntity> void write(C entity) {
     LinkedHashMap<String, String> entityFieldData;
     try {
-      entityFieldData =
-          processorProvider
-              .handleEntity(entity)
-              .map(this::csvEntityFieldData)
-              .orElseThrow(
-                  () ->
-                      new SinkException(
-                          "Cannot persist entity of type '"
-                              + entity.getClass().getSimpleName()
-                              + "'. This sink can only process the following entities: ["
-                              + processorProvider.getRegisteredClasses().stream()
-                                  .map(Class::getSimpleName)
-                                  .collect(Collectors.joining(","))
-                              + "]"));
-
+      entityFieldData = csvEntityFieldData(processorProvider.handleEntity(entity));
       String[] headerElements = processorProvider.getHeaderElements(entity.getClass());
       BufferedCsvWriter writer =
           connector.getOrInitWriter(entity.getClass(), headerElements, csvSep);

src/main/java/edu/ie3/datamodel/io/sink/DataSink.java

@@ -5,6 +5,7 @@
 */
 package edu.ie3.datamodel.io.sink;
 
+import edu.ie3.datamodel.exceptions.ProcessorProviderException;
 import edu.ie3.datamodel.io.connectors.DataConnector;
 import edu.ie3.datamodel.io.processor.EntityProcessor;
 import edu.ie3.datamodel.models.UniqueEntity;
@@ -37,7 +38,7 @@ public interface DataSink {
    * @param <C> bounded to be all unique entities. Handling of specific entities is normally then
    *     executed by a specific {@link EntityProcessor}
    */
-  <C extends UniqueEntity> void persist(C entity);
+  <C extends UniqueEntity> void persist(C entity) throws ProcessorProviderException;
 
   /**
    * Should implement the entry point of a data sink to persist multiple entities in a collection.
@@ -51,7 +52,8 @@ public interface DataSink {
    * @param <C> bounded to be all unique entities. Handling of specific entities is normally then
    *     executed by a specific {@link EntityProcessor}
    */
-  <C extends UniqueEntity> void persistAll(Collection<C> entities);
+  <C extends UniqueEntity> void persistAll(Collection<C> entities)
+      throws ProcessorProviderException;
 
   /**
    * Should implement the handling of a whole time series. Therefore the single entries have to be
@@ -62,5 +64,5 @@ public interface DataSink {
    * @param <V> Type of actual value, that is inside the entry
    */
   <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
-      TimeSeries<E, V> timeSeries);
+      TimeSeries<E, V> timeSeries) throws ProcessorProviderException;
 }

src/main/java/edu/ie3/datamodel/io/sink/InfluxDbSink.java

@@ -5,11 +5,10 @@
 */
 package edu.ie3.datamodel.io.sink;
 
-import edu.ie3.datamodel.exceptions.SinkException;
+import edu.ie3.datamodel.exceptions.ProcessorProviderException;
 import edu.ie3.datamodel.io.connectors.InfluxDbConnector;
 import edu.ie3.datamodel.io.naming.EntityPersistenceNamingStrategy;
 import edu.ie3.datamodel.io.processor.ProcessorProvider;
-import edu.ie3.datamodel.io.processor.timeseries.TimeSeriesProcessorKey;
 import edu.ie3.datamodel.models.UniqueEntity;
 import edu.ie3.datamodel.models.result.ResultEntity;
 import edu.ie3.datamodel.models.timeseries.TimeSeries;
@@ -18,7 +17,6 @@
 import java.time.ZonedDateTime;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
 import org.slf4j.Logger;
@@ -32,6 +30,8 @@ public class InfluxDbSink implements OutputDataSink {
   /** Field name for input model uuid field in result entities */
   private static final String FIELD_NAME_INPUT = "inputModel";
 
+  private static final String ERROR_MESSAGE = "Cannot persist provided entity '{}'. Exception: {}";
+
   private final InfluxDbConnector connector;
   private final EntityPersistenceNamingStrategy entityPersistenceNamingStrategy;
   private final ProcessorProvider processorProvider;
@@ -68,15 +68,16 @@ public void shutdown() {
   }
 
   @Override
-  public <C extends UniqueEntity> void persist(C entity) {
+  public <C extends UniqueEntity> void persist(C entity) throws ProcessorProviderException {
     Set<Point> points = extractPoints(entity);
     // writes only the exact one point instead of unnecessarily wrapping it in BatchPoints
     if (points.size() == 1) write(points.iterator().next());
     else writeAll(points);
   }
 
   @Override
-  public <C extends UniqueEntity> void persistAll(Collection<C> entities) {
+  public <C extends UniqueEntity> void persistAll(Collection<C> entities)
+      throws ProcessorProviderException {
     Set<Point> points = new HashSet<>();
     for (C entity : entities) {
       points.addAll(extractPoints(entity));
@@ -86,7 +87,7 @@ public <C extends UniqueEntity> void persistAll(Collection<C> entities) {
 
   @Override
   public <E extends TimeSeriesEntry<V>, V extends Value> void persistTimeSeries(
-      TimeSeries<E, V> timeSeries) {
+      TimeSeries<E, V> timeSeries) throws ProcessorProviderException {
     Set<Point> points = transformToPoints(timeSeries);
     writeAll(points);
   }
@@ -107,7 +108,7 @@ public void flush() {
    *
    * @param entity the entity to transform
    */
-  private Optional<Point> transformToPoint(ResultEntity entity) {
+  private Point transformToPoint(ResultEntity entity) throws ProcessorProviderException {
     Optional<String> measurementName =
         entityPersistenceNamingStrategy.getResultEntityName(entity.getClass());
     if (measurementName.isEmpty())
@@ -124,37 +125,23 @@ private Optional<Point> transformToPoint(ResultEntity entity) {
    * @param entity the entity to transform
    * @param measurementName equivalent to the name of a relational table
    */
-  private Optional<Point> transformToPoint(ResultEntity entity, String measurementName) {
+  private Point transformToPoint(ResultEntity entity, String measurementName)
+      throws ProcessorProviderException {
     LinkedHashMap<String, String> entityFieldData;
     try {
-      entityFieldData =
-          processorProvider
-              .handleEntity(entity)
-              .orElseThrow(
-                  () ->
-                      new SinkException(
-                          "Cannot persist entity of type '"
-                              + entity.getClass().getSimpleName()
-                              + "'. This sink can only process the following entities: ["
-                              + processorProvider.getRegisteredClasses().stream()
-                                  .map(Class::getSimpleName)
-                                  .collect(Collectors.joining(","))
-                              + "]"));
+      entityFieldData = processorProvider.handleEntity(entity);
       entityFieldData.remove(FIELD_NAME_TIME);
-      return Optional.of(
-          Point.measurement(transformToMeasurementName(measurementName))
-              .time(entity.getTime().toInstant().toEpochMilli(), TimeUnit.MILLISECONDS)
-              .tag("input_model", entityFieldData.remove(FIELD_NAME_INPUT))
-              .tag("scenario", connector.getScenarioName())
-              .fields(Collections.unmodifiableMap(entityFieldData))
-              .build());
-    } catch (SinkException e) {
-      log.error(
-          "Cannot persist provided entity '{}'. Exception: {}",
-          entity.getClass().getSimpleName(),
-          e);
+      return Point.measurement(transformToMeasurementName(measurementName))
+          .time(entity.getTime().toInstant().toEpochMilli(), TimeUnit.MILLISECONDS)
+          .tag("input_model", entityFieldData.remove(FIELD_NAME_INPUT))
+          .tag("scenario", connector.getScenarioName())
+          .fields(Collections.unmodifiableMap(entityFieldData))
+          .build();
+    } catch (ProcessorProviderException e) {
+      log.error(ERROR_MESSAGE, entity.getClass().getSimpleName(), e);
+
+      throw new ProcessorProviderException(e);
     }
-    return Optional.empty();
   }
 
   /**
@@ -166,18 +153,25 @@ private Optional<Point> transformToPoint(ResultEntity entity, String measurement
    * @param timeSeries the time series to transform
    */
   private <E extends TimeSeriesEntry<V>, V extends Value> Set<Point> transformToPoints(
-      TimeSeries<E, V> timeSeries) {
+      TimeSeries<E, V> timeSeries) throws ProcessorProviderException {
     if (timeSeries.getEntries().isEmpty()) return Collections.emptySet();
-    Optional<String> measurementName = entityPersistenceNamingStrategy.getEntityName(timeSeries);
-    if (measurementName.isEmpty()) {
-      String valueClassName =
-          timeSeries.getEntries().iterator().next().getValue().getClass().getSimpleName();
-      log.warn(
-          "I could not get a measurement name for TimeSeries value class {}. I am using it's value's simple name instead.",
-          valueClassName);
-      return transformToPoints(timeSeries, valueClassName);
+
+    try {
+      Optional<String> measurementName = entityPersistenceNamingStrategy.getEntityName(timeSeries);
+      if (measurementName.isEmpty()) {
+        String valueClassName =
+            timeSeries.getEntries().iterator().next().getValue().getClass().getSimpleName();
+        log.warn(
+            "I could not get a measurement name for TimeSeries value class {}. I am using it's value's simple name instead.",
+            valueClassName);
+        return transformToPoints(timeSeries, valueClassName);
+      }
+      return transformToPoints(timeSeries, measurementName.get());
+    } catch (ProcessorProviderException e) {
+      log.error(ERROR_MESSAGE, timeSeries.getClass().getSimpleName(), e);
+      throw new ProcessorProviderException(
+          "Cannot persist provided time series {" + timeSeries.getClass().getSimpleName() + "}", e);
     }
-    return transformToPoints(timeSeries, measurementName.get());
   }
 
   /**
@@ -188,23 +182,11 @@ private <E extends TimeSeriesEntry<V>, V extends Value> Set<Point> transformToPo
    * @param measurementName equivalent to the name of a relational table
    */
   private <E extends TimeSeriesEntry<V>, V extends Value> Set<Point> transformToPoints(
-      TimeSeries<E, V> timeSeries, String measurementName) {
-    TimeSeriesProcessorKey key = new TimeSeriesProcessorKey(timeSeries);
+      TimeSeries<E, V> timeSeries, String measurementName) throws ProcessorProviderException {
     Set<Point> points = new HashSet<>();
     try {
       Set<LinkedHashMap<String, String>> entityFieldData =
-          processorProvider
-              .handleTimeSeries(timeSeries)
-              .orElseThrow(
-                  () ->
-                      new SinkException(
-                          "Cannot persist time series of combination '"
-                              + key
-                              + "'. This sink can only process the following combinations: ["
-                              + processorProvider.getRegisteredTimeSeriesCombinations().stream()
-                                  .map(TimeSeriesProcessorKey::toString)
-                                  .collect(Collectors.joining(","))
-                              + "]"));
+          processorProvider.handleTimeSeries(timeSeries);
 
       for (LinkedHashMap<String, String> dataMapping : entityFieldData) {
         String timeString = dataMapping.remove(FIELD_NAME_TIME);
@@ -217,8 +199,10 @@ private <E extends TimeSeriesEntry<V>, V extends Value> Set<Point> transformToPo
                 .build();
         points.add(point);
       }
-    } catch (SinkException e) {
-      log.error("Cannot persist provided time series '{}'. Exception: {}", key, e);
+    } catch (ProcessorProviderException e) {
+      log.error(ERROR_MESSAGE, timeSeries.getClass().getSimpleName(), e);
+      throw new ProcessorProviderException(
+          "Cannot persist provided time series {" + timeSeries.getClass().getSimpleName() + "}", e);
     }
     return points;
   }
@@ -233,19 +217,15 @@ private <E extends TimeSeriesEntry<V>, V extends Value> Set<Point> transformToPo
    * @param <C> bounded to be all unique entities, but logs an error and returns an empty Set if it
    *     does not extend {@link ResultEntity} or {@link TimeSeries}
    */
-  private <C extends UniqueEntity> Set<Point> extractPoints(C entity) {
+  private <C extends UniqueEntity> Set<Point> extractPoints(C entity)
+      throws ProcessorProviderException {
     Set<Point> points = new HashSet<>();
     /* Distinguish between result models and time series */
     if (entity instanceof ResultEntity resultEntity) {
       try {
-        points.add(
-            transformToPoint(resultEntity)
-                .orElseThrow(() -> new SinkException("Could not transform entity")));
-      } catch (SinkException e) {
-        log.error(
-            "Cannot persist provided entity '{}'. Exception: {}",
-            entity.getClass().getSimpleName(),
-            e);
+        points.add(transformToPoint(resultEntity));
+      } catch (ProcessorProviderException e) {
+        log.error(ERROR_MESSAGE, entity.getClass().getSimpleName(), e);
       }
     } else if (entity instanceof TimeSeries<?, ?> timeSeries) {
       points.addAll(transformToPoints(timeSeries));