Errors during the data mining are now forwarded and logged

This improves the visibilities of failures, because they aren't hided by
the multithreading anymore.

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java

@@ -166,6 +166,9 @@ public class TestProcessorQuery {
                 receivedElementOrFinished = true;
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
                 receivedElementOrFinished = true;
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java

@@ -57,6 +57,9 @@ public class TestAbstractFilteringRetrievalProcessor {
                 }
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
             }
             @Override

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorFinishing.java

@@ -52,6 +52,8 @@ public class TestAbstractParallelProcessorFinishing {
             @Override
             public void onElement(Integer element) { }
             @Override
+            public void onFailure(Throwable failure) { }
+            @Override
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractParallelProcessorWithManySimpleInstructions.java

@@ -29,6 +29,9 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
                 incrementSum(element);
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractProcessor.java

@@ -24,6 +24,9 @@ public class TestAbstractProcessor {
             public void onElement(Integer element) {
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestFilteringProcessors.java

@@ -97,6 +97,9 @@ public class TestFilteringProcessors {
                 receivedElements.put(element, elementAmount);
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
             }
             @Override

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelExtractionProcessor.java

@@ -45,6 +45,9 @@ public class TestParallelExtractionProcessor {
                 receivedValues.put(element.getKey(), element.getDataEntry());
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
             }
             @Override

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java

@@ -37,6 +37,8 @@ public class TestParallelMultiDimensionalGroupingProcessor {
                 groupedElement = element;
             }
             @Override
+            public void onFailure(Throwable failure) { }
+            @Override
             public void finish() throws InterruptedException { }
             @Override
             public void abort() { }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java

@@ -31,6 +31,9 @@ public class TestAbstractStoringParallelAggregationProcessor {
                 receivedElement = element;
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelDoubleAggregationProcessors.java

@@ -184,6 +184,9 @@ public class TestParallelDoubleAggregationProcessors {
                 receivedAggregations = element;
             }
             @Override
+            public void onFailure(Throwable failure) {
+            }
+            @Override
             public void finish() throws InterruptedException {
             }
             @Override

java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java

@@ -6,6 +6,16 @@ public interface Processor<InputType> {
 
     public void onElement(InputType element);
 
+    /**
+     * This method handles the throwing of Throwables, so they don't get lost because of the 
+     * multi-threading.<br>
+     * The standard implementation would be a forwarding to the last processor,
+     * that collects the failures, until the processing is finished. Than the failures will be
+     * handled.
+     * @param failure
+     */
+    void onFailure(Throwable failure);
+
     /**
      * Tells this Processor, that there will be no incoming data.<br />
      * The called Processor will finish his work and call <code>finish()</code> on all subsequent processors.

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/ProcessorQuery.java

@@ -1,6 +1,8 @@
 package com.sap.sse.datamining.impl;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Timer;
@@ -93,6 +95,8 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
         });
         waitTillWorkIsDone(timeoutInMillis);
         final long endTime = System.nanoTime();
+        
+        logOccuredFailures();
 
         long calculationTimeInNanos = endTime - startTime;
         AdditionalResultDataBuilder additionalDataBuilder = new SumBuildingAndOverwritingResultDataBuilder();
@@ -101,6 +105,12 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
         return new QueryResultImpl<>(results, additionalDataBuilder.build(calculationTimeInNanos, stringMessages, locale));
     }
 
+    private void logOccuredFailures() {
+        for (Throwable failure : resultReceiver.getOccuredFailures()) {
+            LOGGER.log(Level.SEVERE, "An error occured during the processing of an instruction: ", failure);
+        }
+    }
+
     private void waitTillWorkIsDone(long timeoutInMillis) throws InterruptedException, TimeoutException {
         setUpTimeoutTimer(timeoutInMillis);
         synchronized (monitorObject) {
@@ -138,10 +148,12 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
         
         private final ReentrantLock resultsLock;
         private Map<GroupKey, AggregatedType> results;
+        private List<Throwable> occuredFailures;
         
         public ProcessResultReceiver() {
             resultsLock = new ReentrantLock();
             results = new HashMap<>();
+            occuredFailures = new ArrayList<>();
         }
 
         @Override
@@ -153,6 +165,11 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
                 resultsLock.unlock();
             }
         }
+        
+        @Override
+        public void onFailure(Throwable failure) {
+            occuredFailures.add(failure);
+        }
 
         @Override
         public void finish() throws InterruptedException {
@@ -164,12 +181,17 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
         
         @Override
         public void abort() {
-            results = null;
+            results = new HashMap<>();
+            occuredFailures = new ArrayList<>();
         }
         
         public Map<GroupKey, AggregatedType> getResult() {
             return results;
         }
+        
+        public List<Throwable> getOccuredFailures() {
+            return occuredFailures;
+        }
 
         @Override
         public AdditionalResultDataBuilder getAdditionalResultData(AdditionalResultDataBuilder additionalDataBuilder) {

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java

@@ -46,7 +46,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
                             }
                         } catch (Exception e) {
                             if (!gotAborted || !(e instanceof InterruptedException)) {
-                                LOGGER.log(Level.FINEST, "An error occured during the processing of an instruction: ", e);
+                                onFailure(e);
                             }
                         } finally {
                             AbstractPartitioningParallelProcessor.this.unfinishedInstructionsCounter.decrement();
@@ -88,6 +88,13 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
 
     protected abstract Iterable<WorkingType> partitionElement(InputType element);
 
+    @Override
+    public void onFailure(Throwable failure) {
+        for (Processor<ResultType> resultReceiver : resultReceivers) {
+            resultReceiver.onFailure(failure);
+        }
+    }
+
     @Override
     public void finish() throws InterruptedException {
         sleepUntilAllInstructionsFinished();

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessor.java

@@ -20,6 +20,13 @@ public abstract class AbstractProcessor<InputType, ResultType> implements Proces
         ResultType result = processElement(element);
         forwardResultToTheReceivers(result);
     }
+    
+    @Override
+    public void onFailure(Throwable failure) {
+        for (Processor<ResultType> resultReceiver : resultReceivers) {
+            resultReceiver.onFailure(failure);
+        }
+    }
 
     private void forwardResultToTheReceivers(ResultType result) {
         for (Processor<ResultType> resultReceiver : resultReceivers) {

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/functions/SimpleFunctionRegistry.java

@@ -1,10 +1,12 @@
 package com.sap.sse.datamining.impl.functions;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -129,9 +131,12 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
     private class FunctionsReceiver implements Processor<Collection<Function<?>>> {
         
         private final Lock functionsRegistrationLock;
+        
+        private final List<Throwable> occuredFailures;
 
         public FunctionsReceiver() {
             functionsRegistrationLock = new ReentrantLock();
+            occuredFailures = new ArrayList<>();
         }
 
         @Override
@@ -143,9 +148,17 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
                 functionsRegistrationLock.unlock();
             }
         }
+        
+        @Override
+        public void onFailure(Throwable failure) {
+            occuredFailures.add(failure);
+        }
 
         @Override
         public void finish() throws InterruptedException {
+            for (Throwable failure : occuredFailures) {
+                LOGGER.log(Level.SEVERE, "An error occured during the processing of an instruction: ", failure);
+            }
         }
 
         @Override