Implemented an aborting mechanism for processors and timeouting in the

query

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java

@@ -2,7 +2,7 @@ package com.sap.sailing.datamining.impl.components;
 
 import java.util.Collection;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sailing.datamining.data.GPSFixWithContext;
 import com.sap.sailing.domain.leaderboard.LeaderboardGroup;
@@ -13,7 +13,7 @@ import com.sap.sse.datamining.impl.components.AbstractPartitioningParallelProces
 public class GPSFixRetrievalProcessor extends
         AbstractPartitioningParallelProcessor<RacingEventService, LeaderboardGroup, GPSFixWithContext> {
 
-    public GPSFixRetrievalProcessor(Executor executor, Collection<Processor<GPSFixWithContext>> resultReceivers) {
+    public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<GPSFixWithContext>> resultReceivers) {
         super(executor, resultReceivers);
     }
 

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

@@ -54,6 +54,8 @@ public class TestAbstractParallelProcessorFinishing {
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }
+            @Override
+            public void abort() { }
         };
     }
 

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

@@ -31,6 +31,9 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }
+            @Override
+            public void abort() {
+            }
         };
         
         Collection<Processor<Integer>> receivers = new ArrayList<>();

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

@@ -26,6 +26,9 @@ public class TestAbstractProcessor {
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }
+            @Override
+            public void abort() {
+            }
         };
         
         receivers = new HashSet<>();
@@ -39,6 +42,9 @@ public class TestAbstractProcessor {
             protected Integer processElement(Integer element) {
                 return null;
             }
+            @Override
+            public void abort() {
+            }
         };
         processor.finish();
         assertThat(receiverWasToldToFinish, is(true));

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

@@ -98,6 +98,9 @@ public class TestFilteringProcessors {
             @Override
             public void finish() throws InterruptedException {
             }
+            @Override
+            public void abort() {
+            }
         };
         
         receivers = new HashSet<>();

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

@@ -46,6 +46,9 @@ public class TestParallelExtractionProcessor {
             @Override
             public void finish() throws InterruptedException {
             }
+            @Override
+            public void abort() {
+            }
         };
         
         receivers = new HashSet<>();

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

@@ -38,6 +38,8 @@ public class TestParallelMultiDimensionalGroupingProcessor {
             }
             @Override
             public void finish() throws InterruptedException { }
+            @Override
+            public void abort() { }
         };
         
         receivers = new ArrayList<>();

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

@@ -34,8 +34,8 @@ import com.sap.sse.datamining.test.util.FunctionTestsUtil;
 
 public class TestProcessorQuery {
 
-    @SuppressWarnings("unused")
     private boolean receivedElementOrFinished;
+    private boolean receivedAbort;
 
     @Test
     public void testStandardWorkflow() throws InterruptedException, ExecutionException {
@@ -143,6 +143,10 @@ public class TestProcessorQuery {
             public void finish() throws InterruptedException {
                 receivedElementOrFinished = true;
             }
+            @Override
+            public void abort() {
+                receivedAbort = true;
+            }
         };
         query.setFirstProcessor(createBlockingProcessor(1000, resultReceiver));
         
@@ -153,8 +157,9 @@ public class TestProcessorQuery {
             // A timeout exception is expected
         }
         
-//        ConcurrencyTestsUtil.sleepFor(1000); // Wait if a result is received
-//        assertThat("The processing should be aborted", receivedElementOrFinished, is(false));
+        ConcurrencyTestsUtil.sleepFor(1000); // Wait if a result is received
+        assertThat("The processing should be aborted", receivedElementOrFinished, is(false));
+        assertThat("The processing should be aborted", receivedAbort, is(true));
     }
 
     private Processor<Iterable<Number>> createBlockingProcessor(final long timeToBlockInMillis, Processor<Double> resultReceiver) {

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

@@ -32,6 +32,9 @@ public class TestAbstractStoringParallelAggregationProcessor {
             public void finish() throws InterruptedException {
                 receiverWasToldToFinish = true;
             }
+            @Override
+            public void abort() {
+            }
         };
         
         receivers = new HashSet<>();

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

@@ -185,6 +185,9 @@ public class TestParallelDoubleAggregationProcessors {
             @Override
             public void finish() throws InterruptedException {
             }
+            @Override
+            public void abort() {
+            }
         };
         
         receivers = new ArrayList<>();

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/ConcurrencyTestsUtil.java

@@ -11,11 +11,18 @@ import com.sap.sse.datamining.components.Processor;
 public class ConcurrencyTestsUtil {
 
     private static final int THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 3);
-    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
+    private static ThreadPoolExecutor executor = createExecutor();
 
     public static ThreadPoolExecutor getExecutor() {
+        if (executor.isShutdown()) {
+            executor = createExecutor();
+        }
         return executor;
     }
+
+    private static ThreadPoolExecutor createExecutor() {
+        return new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
+    }
     
     public static void sleepFor(long milliseconds) {
         try {

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

@@ -14,5 +14,9 @@ public class NonFilteringProcessor<InputType> extends AbstractProcessor<InputTyp
     protected InputType processElement(InputType element) {
         return element;
     }
+    
+    @Override
+    public void abort() {
+    }
 
 }

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

@@ -1,14 +1,17 @@
 package com.sap.sse.datamining.components;
 
 public interface Processor<InputType> {
-	
-	public void onElement(InputType element);
-	
-	/**
-	 * 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.
-	 * @throws InterruptedException 
-	 */
-	public void finish() throws InterruptedException;
+
+    public void onElement(InputType element);
+
+    /**
+     * 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.
+     * 
+     * @throws InterruptedException
+     */
+    public void finish() throws InterruptedException;
+
+    public void abort();
 
 }

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

@@ -3,7 +3,7 @@ package com.sap.sse.datamining.impl.components;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.FilterCriteria;
 import com.sap.sse.datamining.components.Processor;
@@ -13,7 +13,7 @@ public abstract class AbstractFilteringRetrievalProcessor<InputType, WorkingType
 
     private final FilterCriteria<WorkingType> criteria;
 
-    public AbstractFilteringRetrievalProcessor(Executor executor, Collection<Processor<ResultType>> resultReceivers, FilterCriteria<WorkingType> criteria) {
+    public AbstractFilteringRetrievalProcessor(ExecutorService executor, Collection<Processor<ResultType>> resultReceivers, FilterCriteria<WorkingType> criteria) {
         super(executor, resultReceivers);
         this.criteria = criteria;
     }

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

@@ -5,7 +5,7 @@ import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -21,10 +21,10 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
     private static final int SLEEP_TIME_DURING_FINISHING = 100;
 
     private final Set<Processor<ResultType>> resultReceivers;
-    private final Executor executor;
+    private final ExecutorService executor;
     private final UnfinishedInstructionsCounter unfinishedInstructionsCounter;
 
-    public AbstractPartitioningParallelProcessor(Executor executor, Collection<Processor<ResultType>> resultReceivers) {
+    public AbstractPartitioningParallelProcessor(ExecutorService executor, Collection<Processor<ResultType>> resultReceivers) {
         this.executor = executor;
         this.resultReceivers = new HashSet<Processor<ResultType>>(resultReceivers);
         unfinishedInstructionsCounter = new UnfinishedInstructionsCounter();
@@ -90,7 +90,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
     @Override
     public void finish() throws InterruptedException {
         sleepUntilAllInstructionsFinished();
-        notifyResultReceiversToFinish();
+        tellResultReceiversToFinish();
     }
 
     protected void sleepUntilAllInstructionsFinished() throws InterruptedException {
@@ -103,7 +103,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
         return unfinishedInstructionsCounter.getUnfinishedInstructionsAmount() > 0;
     }
 
-    protected void notifyResultReceiversToFinish() {
+    protected void tellResultReceiversToFinish() {
         for (Processor<ResultType> resultReceiver : getResultReceivers()) {
             try {
                 resultReceiver.finish();
@@ -113,6 +113,18 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
         }
     }
     
+    @Override
+    public void abort() {
+        executor.shutdownNow();
+        tellResultReceiversToAbort();
+    }
+
+    private void tellResultReceiversToAbort() {
+        for (Processor<ResultType> resultReceiver : getResultReceivers()) {
+            resultReceiver.abort();
+        }
+    }
+    
     /**
      * Thread safe class to manage, if there are unfinished instructions. 
      */

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

@@ -3,14 +3,14 @@ package com.sap.sse.datamining.impl.components;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 
 public abstract class AbstractSimpleParallelProcessor<InputType, ResultType>
                       extends AbstractPartitioningParallelProcessor<InputType, InputType, ResultType> {
 
-    public AbstractSimpleParallelProcessor(Executor executor, Collection<Processor<ResultType>> resultReceivers) {
+    public AbstractSimpleParallelProcessor(ExecutorService executor, Collection<Processor<ResultType>> resultReceivers) {
         super(executor, resultReceivers);
     }
 

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

@@ -2,7 +2,7 @@ package com.sap.sse.datamining.impl.components;
 
 import java.util.Collection;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.FilterCriteria;
 import com.sap.sse.datamining.components.Processor;
@@ -11,7 +11,7 @@ public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParalle
 
     private final FilterCriteria<InputType> filterCriteria;
 
-    public ParallelFilteringProcessor(Executor executor, Collection<Processor<InputType>> resultReceivers, FilterCriteria<InputType> filterCriteria) {
+    public ParallelFilteringProcessor(ExecutorService executor, Collection<Processor<InputType>> resultReceivers, FilterCriteria<InputType> filterCriteria) {
         super(executor, resultReceivers);
         this.filterCriteria = filterCriteria;
     }

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

@@ -3,7 +3,7 @@ package com.sap.sse.datamining.impl.components;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.functions.Function;
@@ -16,7 +16,7 @@ public class ParallelMultiDimensionalGroupingProcessor<DataType>
 
     private Iterable<Function<?>> dimensions;
 
-    public ParallelMultiDimensionalGroupingProcessor(Executor executor, Collection<Processor<GroupedDataEntry<DataType>>> resultReceivers,
+    public ParallelMultiDimensionalGroupingProcessor(ExecutorService executor, Collection<Processor<GroupedDataEntry<DataType>>> resultReceivers,
                                                      Iterable<Function<?>> dimensions) {
         super(executor, resultReceivers);
         verifyThatDimensionsAreDimensions(dimensions);

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

@@ -85,8 +85,8 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
         synchronized (monitorObject) {
             while (!workIsDone) {
                 monitorObject.wait();
-                if (processorTimedOut) {
-                    // TODO Abort the processing
+                if (processorTimedOut && !workIsDone) {
+                    firstProcessor.abort();
                     throw new TimeoutException("The query processing timed out");
                 }
             }
@@ -139,6 +139,11 @@ public class ProcessorQuery<AggregatedType, DataSourceType> implements Query<Agg
             }
         }
         
+        @Override
+        public void abort() {
+            result = null;
+        }
+        
         public QueryResult<AggregatedType> getResult() {
             return result;
         }

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java

@@ -2,7 +2,7 @@ package com.sap.sse.datamining.impl.components.aggregators;
 
 import java.util.Collection;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import com.sap.sse.datamining.components.Processor;
@@ -13,7 +13,7 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
 
     private final ReentrantReadWriteLock storeLock;
 
-    public AbstractParallelStoringAggregationProcessor(Executor executor, Collection<Processor<AggregatedType>> resultReceivers) {
+    public AbstractParallelStoringAggregationProcessor(ExecutorService executor, Collection<Processor<AggregatedType>> resultReceivers) {
         super(executor, resultReceivers);
         storeLock = new ReentrantReadWriteLock();
     }
@@ -44,7 +44,7 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
     public void finish() throws InterruptedException {
         super.sleepUntilAllInstructionsFinished();
         super.forwardResultToReceivers(aggregateResult());
-        super.notifyResultReceiversToFinish();
+        super.tellResultReceiversToFinish();
     }
 
     protected abstract AggregatedType aggregateResult();

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataAverageAggregationProcessor.java

@@ -4,7 +4,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
@@ -16,7 +16,7 @@ public class ParallelGroupedDoubleDataAverageAggregationProcessor extends
     private final AbstractParallelStoringAggregationProcessor<GroupedDataEntry<Double>, Map<GroupKey, Double>> sumAggregationProcessor;
     private final Map<GroupKey, Integer> elementAmountPerKey;
 
-    public ParallelGroupedDoubleDataAverageAggregationProcessor(Executor executor,
+    public ParallelGroupedDoubleDataAverageAggregationProcessor(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
         super(executor, resultReceivers);
         elementAmountPerKey = new HashMap<>();

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMedianAggregationProcessor.java

@@ -7,7 +7,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
@@ -18,7 +18,7 @@ public class ParallelGroupedDoubleDataMedianAggregationProcessor
 
     private Map<GroupKey, List<Double>> groupedValues;
 
-    public ParallelGroupedDoubleDataMedianAggregationProcessor(Executor executor,
+    public ParallelGroupedDoubleDataMedianAggregationProcessor(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
         super(executor, resultReceivers);
         groupedValues = new HashMap<>();

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataSumAggregationProcessor.java

@@ -4,7 +4,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
@@ -15,7 +15,7 @@ public class ParallelGroupedDoubleDataSumAggregationProcessor
 
     private Map<GroupedDataEntry<Double>, Integer> elementAmountMap;
 
-    public ParallelGroupedDoubleDataSumAggregationProcessor(Executor executor,
+    public ParallelGroupedDoubleDataSumAggregationProcessor(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>>> resultReceivers) {
         super(executor, resultReceivers);
         elementAmountMap = new HashMap<>();