Merge branch 'master' into ubilabs--rcapp--develop

java/com.sap.sailing.datamining.test/src/com/sap/sailing/datamining/impl/components/TestLeaderboardGroupRetrievalProcessor.java

@@ -46,7 +46,7 @@ public class TestLeaderboardGroupRetrievalProcessor {
         
         Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers = new ArrayList<>();
         resultReceivers.add(receiver);
-        retriever = new LeaderboardGroupRetrievalProcessor(ConcurrencyTestsUtil.getExecutor(), resultReceivers);
+        retriever = new LeaderboardGroupRetrievalProcessor(ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0);
     }
 
     private Map<String, LeaderboardGroup> getGroupsInService() {

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

@@ -2,18 +2,17 @@ package com.sap.sailing.datamining.test.util;
 
 import static org.junit.Assert.fail;
 
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.DataMiningExecutorService;
 
 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 final ExecutorService executor = new DataMiningExecutorService(THREAD_POOL_SIZE);
 
-    public static ThreadPoolExecutor getExecutor() {
+    public static ExecutorService getExecutor() {
         return executor;
     }
 

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

@@ -16,8 +16,8 @@ public class CompetitorOfRaceInLeaderboardRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasRaceResultOfCompetitorContext> {
 
     public CompetitorOfRaceInLeaderboardRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasRaceResultOfCompetitorContext, ?>> resultReceivers) {
-        super(HasLeaderboardContext.class, HasRaceResultOfCompetitorContext.class, executor, resultReceivers);
+            Collection<Processor<HasRaceResultOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasLeaderboardContext.class, HasRaceResultOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -16,8 +16,8 @@ import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
 
 public class GPSFixRetrievalProcessor extends AbstractSimpleRetrievalProcessor<HasTrackedLegOfCompetitorContext, HasGPSFixContext> {
 
-    public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<HasGPSFixContext, ?>> resultReceivers) {
-        super(HasTrackedLegOfCompetitorContext.class, HasGPSFixContext.class, executor, resultReceivers);
+    public GPSFixRetrievalProcessor(ExecutorService executor, Collection<Processor<HasGPSFixContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasTrackedLegOfCompetitorContext.class, HasGPSFixContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -14,19 +14,18 @@ public class LeaderboardGroupRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<RacingEventService, LeaderboardGroupWithContext> {
 
     public LeaderboardGroupRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers) {
-        super(RacingEventService.class, LeaderboardGroupWithContext.class, executor, resultReceivers);
+            Collection<Processor<LeaderboardGroupWithContext, ?>> resultReceivers, int retrievalLevel) {
+        super(RacingEventService.class, LeaderboardGroupWithContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override
     protected Iterable<LeaderboardGroupWithContext> retrieveData(RacingEventService element) {
-		final PolarDataService polarDataService = element.getPolarDataService();
-		return element
-				.getLeaderboardGroups()
-				.values()
-				.stream()
-				.map(lg -> new LeaderboardGroupWithContext(lg, polarDataService))
-				.collect(Collectors.toSet());
-	}
+        final PolarDataService polarDataService = element.getPolarDataService();
+        return element.getLeaderboardGroups()
+                .values()
+                .stream()
+                .map(lg -> new LeaderboardGroupWithContext(lg, polarDataService))
+                .collect(Collectors.toSet());
+    }
 
 }

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

@@ -15,8 +15,8 @@ public class LeaderboardRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<LeaderboardGroupWithContext, HasLeaderboardContext> {
 
     public LeaderboardRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasLeaderboardContext, ?>> resultReceivers) {
-        super(LeaderboardGroupWithContext.class, HasLeaderboardContext.class, executor, resultReceivers);
+            Collection<Processor<HasLeaderboardContext, ?>> resultReceivers, int retrievalLevel) {
+        super(LeaderboardGroupWithContext.class, HasLeaderboardContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -19,8 +19,8 @@ public class MarkPassingRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasMarkPassingContext> {
 
     public MarkPassingRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasMarkPassingContext, ?>> resultReceivers) {
-        super(HasTrackedRaceContext.class, HasMarkPassingContext.class, executor, resultReceivers);
+            Collection<Processor<HasMarkPassingContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasTrackedRaceContext.class, HasMarkPassingContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -15,8 +15,8 @@ public class TrackedLegOfCompetitorRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<HasTrackedLegContext, HasTrackedLegOfCompetitorContext> {
 
     public TrackedLegOfCompetitorRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasTrackedLegOfCompetitorContext, ?>> resultReceivers) {
-        super(HasTrackedLegContext.class, HasTrackedLegOfCompetitorContext.class, executor, resultReceivers);
+            Collection<Processor<HasTrackedLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasTrackedLegContext.class, HasTrackedLegOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -15,8 +15,8 @@ public class TrackedLegRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<HasTrackedRaceContext, HasTrackedLegContext> {
 
     public TrackedLegRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasTrackedLegContext, ?>> resultReceivers) {
-        super(HasTrackedRaceContext.class, HasTrackedLegContext.class, executor, resultReceivers);
+            Collection<Processor<HasTrackedLegContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasTrackedRaceContext.class, HasTrackedLegContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -18,8 +18,8 @@ public class TrackedRaceRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<HasLeaderboardContext, HasTrackedRaceContext> {
 
     public TrackedRaceRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<HasTrackedRaceContext, ?>> resultReceivers) {
-        super(HasLeaderboardContext.class, HasTrackedRaceContext.class, executor, resultReceivers);
+            Collection<Processor<HasTrackedRaceContext, ?>> resultReceivers, int retrievalLevel) {
+        super(HasLeaderboardContext.class, HasTrackedRaceContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/datamining/DataMiningEntryPoint.java

@@ -49,11 +49,11 @@ public class DataMiningEntryPoint extends AbstractSailingEntryPoint {
         SplitLayoutPanel splitPanel = new SplitLayoutPanel();
         rootPanel.add(splitPanel);
 
-        DockLayoutPanel dockPanel = new DockLayoutPanel(Unit.PX);
-        dockPanel.addNorth(createLogoAndTitlePanel(), 68);
+        DockLayoutPanel selectionDockPanel = new DockLayoutPanel(Unit.PX);
+        selectionDockPanel.addNorth(createLogoAndTitlePanel(), 68);
         BufferingQueryDefinitionProviderWithControls queryDefinitionProviderWithControls = new BufferingQueryDefinitionProviderWithControls(session, getStringMessages(), sailingService, dataMiningService, this);
         queryDefinitionProviderWithControls.getEntryWidget().addStyleName("dataMiningPanel");
-        dockPanel.add(queryDefinitionProviderWithControls.getEntryWidget());
+        selectionDockPanel.add(queryDefinitionProviderWithControls.getEntryWidget());
         
         ResultsPresenter<Number> resultsPresenter = new ResultsChart(getStringMessages());
         if (GwtHttpRequestUtils.getBooleanParameter("benchmark", false)) {
@@ -63,7 +63,7 @@ public class DataMiningEntryPoint extends AbstractSailingEntryPoint {
             splitPanel.addSouth(resultsPresenter.getWidget(), 400);
         }
         
-        splitPanel.add(dockPanel);
+        splitPanel.add(selectionDockPanel);
         
         QueryRunner queryRunner = new SimpleQueryRunner(session, getStringMessages(), dataMiningService, this, queryDefinitionProviderWithControls, resultsPresenter);
         queryDefinitionProviderWithControls.addControl(queryRunner.getEntryWidget());

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/datamining/presentation/ResultsChart.java

@@ -22,17 +22,17 @@ import org.moxieapps.gwt.highcharts.client.labels.AxisLabelsData;
 import org.moxieapps.gwt.highcharts.client.labels.AxisLabelsFormatter;
 import org.moxieapps.gwt.highcharts.client.labels.YAxisLabels;
 
+import com.google.gwt.dom.client.Style.Unit;
 import com.google.gwt.event.logical.shared.ValueChangeEvent;
 import com.google.gwt.event.logical.shared.ValueChangeHandler;
 import com.google.gwt.text.shared.AbstractRenderer;
-import com.google.gwt.user.client.ui.FlowPanel;
+import com.google.gwt.user.client.ui.DockLayoutPanel;
 import com.google.gwt.user.client.ui.HTML;
 import com.google.gwt.user.client.ui.HorizontalPanel;
 import com.google.gwt.user.client.ui.Label;
 import com.google.gwt.user.client.ui.ValueListBox;
 import com.google.gwt.user.client.ui.Widget;
 import com.sap.sailing.gwt.ui.client.StringMessages;
-import com.sap.sailing.gwt.ui.client.shared.panels.ResizingFlowPanel;
 import com.sap.sailing.gwt.ui.client.shared.panels.ResizingSimplePanel;
 import com.sap.sailing.gwt.ui.datamining.ResultsPresenter;
 import com.sap.sse.datamining.shared.GroupKey;
@@ -76,7 +76,7 @@ public class ResultsChart implements ResultsPresenter<Number> {
     };
     
     private final StringMessages stringMessages;
-    private final FlowPanel mainPanel;
+    private final DockLayoutPanel mainPanel;
 
     private final HorizontalPanel sortByPanel;
     private final ValueListBox<Comparator<GroupKey>> keyComparatorListBox;
@@ -95,17 +95,12 @@ public class ResultsChart implements ResultsPresenter<Number> {
         super();
         this.stringMessages = stringMessages;
         
-        mainPanel = new ResizingFlowPanel() {
-            @Override
-            public void onResize() {
-                presentationPanel.onResize();
-            }
-        };
+        mainPanel = new DockLayoutPanel(Unit.PX);
         
         sortByPanel = new HorizontalPanel();
         sortByPanel.setSpacing(5);
         sortByPanel.setVisible(false);
-        mainPanel.add(sortByPanel);
+        mainPanel.addNorth(sortByPanel, 50);
         sortByPanel.add(new Label(stringMessages.sortBy()));
         keyComparatorListBox = new ValueListBox<>(new AbstractRenderer<Comparator<?>>() {
             @Override

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/server/DataMiningServiceImpl.java

@@ -129,6 +129,7 @@ public class DataMiningServiceImpl extends RemoteServiceServlet implements DataM
         Iterable<Function<?>> dimensions = functionDTOsAsFunctions(dimensionDTOs);
         Locale locale = ResourceBundleStringMessages.Util.getLocaleFor(localeInfoName);
         Query<Set<Object>> dimensionValuesQuery = getDataMiningServer().createDimensionValuesQuery(retrieverChainDefinition, retrieverLevel, dimensions, locale);
+//        QueryResult<Set<Object>> result = getDataMiningServer().runNewQueryAndAbortPreviousQueries(session, dimensionValuesQuery);
         QueryResult<Set<Object>> result = dimensionValuesQuery.run();
         return result;
     }

java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/AbstractEnrichingProcessor.java

@@ -6,7 +6,7 @@ import java.util.concurrent.ExecutorService;
 import com.sap.sse.datamining.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
-import com.sap.sse.datamining.impl.components.ProcessorInstruction;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 
 public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
         AbstractSimpleParallelProcessor<InputType, ResultType> {
@@ -17,8 +17,8 @@ public abstract class AbstractEnrichingProcessor<InputType, ResultType> extends
     }
 
     @Override
-    protected ProcessorInstruction<ResultType> createInstruction(final InputType element) {
-        return new ProcessorInstruction<ResultType>(this) {
+    protected AbstractProcessorInstruction<ResultType> createInstruction(final InputType element) {
+        return new AbstractProcessorInstruction<ResultType>(this) {
 
             @Override
             public ResultType computeResult() {

java/com.sap.sailing.polars/src/com/sap/sailing/polars/mining/PolarFixFilterCriteria.java

@@ -2,6 +2,7 @@ package com.sap.sailing.polars.mining;
 
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.NavigableSet;
 
 import com.sap.sailing.domain.base.Competitor;
 import com.sap.sailing.domain.base.Course;
@@ -88,20 +89,23 @@ public class PolarFixFilterCriteria implements FilterCriterion<GPSFixMovingWithP
     private static boolean isInLeadingCompetitorsForLiveRace(TrackedRace trackedRace, Competitor competitor,
             double pctOfLeadingCompetitorsToInclude) {
         boolean result = false;
-        Waypoint wayPoint = trackedRace.getMarkPassings(competitor).last().getWaypoint();
-        if (wayPoint != null) {
-            Iterator<MarkPassing> markPassingsAtCompetitorsLastWayPoint = trackedRace.getMarkPassingsInOrder(
-                    wayPoint).iterator();
-            for (int i = 0; i < ((int) Math.max(Math.round(pctOfLeadingCompetitorsToInclude
-                    * getNumberOfCompetitors(trackedRace)), 1)); i++) {
-                if (markPassingsAtCompetitorsLastWayPoint.hasNext()) {
-                    if (markPassingsAtCompetitorsLastWayPoint.next().getCompetitor().equals(competitor)) {
-                        result = true;
+        NavigableSet<MarkPassing> markPassingsOfCompetitor = trackedRace.getMarkPassings(competitor);
+        if (!markPassingsOfCompetitor.isEmpty()) {
+            Waypoint wayPoint = markPassingsOfCompetitor.last().getWaypoint();
+            if (wayPoint != null) {
+                Iterator<MarkPassing> markPassingsAtCompetitorsLastWayPoint = trackedRace.getMarkPassingsInOrder(
+                        wayPoint).iterator();
+                for (int i = 0; i < ((int) Math.max(
+                        Math.round(pctOfLeadingCompetitorsToInclude * getNumberOfCompetitors(trackedRace)), 1)); i++) {
+                    if (markPassingsAtCompetitorsLastWayPoint.hasNext()) {
+                        if (markPassingsAtCompetitorsLastWayPoint.next().getCompetitor().equals(competitor)) {
+                            result = true;
+                            break;
+                        }
+                    } else {
+                        result = false;
                         break;
                     }
-                } else {
-                    result = false;
-                    break;
                 }
             }
         }

java/com.sap.sse.datamining.shared/src/com/sap/sse/datamining/shared/impl/NullAdditionalResultData.java

@@ -3,6 +3,19 @@ package com.sap.sse.datamining.shared.impl;
 import com.sap.sse.datamining.shared.AdditionalResultData;
 import com.sap.sse.datamining.shared.Unit;
 
+/**
+ * Null Object pattern for {@link AdditionalResultData} with the values:
+ * <ul>
+ *   <li><b>Retrieved Data Amount</b>: <code>0</code></li>
+ *   <li><b>Calculation Time</b>: <code>0</code></li>
+ *   <li><b>Result Signifier</b>: <code>""</code></li>
+ *   <li><b>Unit</b>: {@link Unit#None}</li>
+ *   <li><b>Unit Signifier</b>: <code>""</code></li>
+ *   <li><b>Value Decimals</b>: <code>0</code></li>
+ * </ul>
+ * 
+ * @author Lennart Hensler (D054527)
+ */
 public class NullAdditionalResultData implements AdditionalResultData {
     private static final long serialVersionUID = -8129840449690994767L;
 

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

@@ -46,20 +46,15 @@ public class TestDataRetrieverChainCreation {
     @Before
     public void initializeRetrieverChain() {
         dataRetrieverChainDefinition = new SimpleDataRetrieverChainDefinition<>((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_HasLegOfCompetitorContext.class, "TestRetrieverChain");
-        Class<Processor<Collection<Test_Regatta>, Test_Regatta>> regattaRetrieverClass = (Class<Processor<Collection<Test_Regatta>, Test_Regatta>>)(Class<?>) TestRegattaRetrievalProcessor.class;
-        dataRetrieverChainDefinition.startWith(regattaRetrieverClass, Test_Regatta.class, "regatta");
+        dataRetrieverChainDefinition.startWith(TestRegattaRetrievalProcessor.class, Test_Regatta.class, "regatta");
         
-        Class<Processor<Test_Regatta, Test_HasRaceContext>> raceRetrieverClass = 
-                (Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) TestRaceWithContextRetrievalProcessor.class;
-        dataRetrieverChainDefinition.addAfter(regattaRetrieverClass,
-                                               raceRetrieverClass,
-                                               Test_HasRaceContext.class, "race");
+        dataRetrieverChainDefinition.addAfter(TestRegattaRetrievalProcessor.class,
+                                              TestRaceWithContextRetrievalProcessor.class,
+                                              Test_HasRaceContext.class, "race");
         
-        Class<Processor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>> legRetrieverClass = 
-                (Class<Processor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>>)(Class<?>) TestLegOfCompetitorWithContextRetrievalProcessor.class;
-        dataRetrieverChainDefinition.endWith(raceRetrieverClass,
-                                               legRetrieverClass,
-                                               Test_HasLegOfCompetitorContext.class, "legOfCompetitor");
+        dataRetrieverChainDefinition.endWith(TestRaceWithContextRetrievalProcessor.class,
+                                             TestLegOfCompetitorWithContextRetrievalProcessor.class,
+                                             Test_HasLegOfCompetitorContext.class, "legOfCompetitor");
     }
     
     @Before
@@ -179,8 +174,8 @@ public class TestDataRetrieverChainCreation {
         dataRetrieverChainDefinition.startWith(regattaRetrieverClass, Test_Regatta.class, "regatta");
         
         dataRetrieverChainDefinition.addAfter(regattaRetrieverClass,
-                                               (Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) Processor.class,
-                                               Test_HasRaceContext.class, "race");
+                                              (Class<Processor<Test_Regatta, Test_HasRaceContext>>)(Class<?>) Processor.class,
+                                              Test_HasRaceContext.class, "race");
     }
     
     @SuppressWarnings("unchecked")

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

@@ -12,25 +12,28 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import org.junit.Test;
 
 import com.sap.sse.common.Util.Pair;
+import com.sap.sse.datamining.AdditionalQueryData;
 import com.sap.sse.datamining.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.Query;
+import com.sap.sse.datamining.QueryState;
 import com.sap.sse.datamining.components.FilterCriterion;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.factories.ProcessorFactory;
 import com.sap.sse.datamining.functions.Function;
 import com.sap.sse.datamining.functions.ParameterProvider;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
 import com.sap.sse.datamining.impl.components.AbstractSimpleRetrievalProcessor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.ParallelFilteringProcessor;
-import com.sap.sse.datamining.impl.components.ProcessorInstruction;
 import com.sap.sse.datamining.impl.criterias.AbstractFilterCriterion;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.shared.QueryResult;
@@ -63,6 +66,7 @@ public class TestProcessorQuery {
     public void testStandardWorkflow() throws InterruptedException, ExecutionException {
         Collection<Number> dataSource = createDataSource();
         Query<Double> queryWithStandardWorkflow = createQueryWithStandardWorkflow(dataSource);
+        assertThat(queryWithStandardWorkflow.getState(), is(QueryState.NOT_STARTED));
         QueryResult<Double> expectedResult = buildExpectedResult(dataSource);
         verifyResult(queryWithStandardWorkflow.run(), expectedResult);
     }
@@ -106,8 +110,8 @@ public class TestProcessorQuery {
      * their length, extracts the cross sum and aggregates these as sum.
      */
     private Query<Double> createQueryWithStandardWorkflow(Collection<Number> dataSource) {
-        final ThreadPoolExecutor executor = ConcurrencyTestsUtil.getExecutor();
-        ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(dataSource, stringMessages, Locale.ENGLISH) {
+        final ExecutorService executor = ConcurrencyTestsUtil.getExecutor();
+        ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(dataSource, stringMessages, Locale.ENGLISH, AdditionalQueryData.NULL_INSTANCE) {
             @Override
             protected Processor<Iterable<Number>, ?> createFirstProcessor() {
                 Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> sumAggregator = processorFactory.createAggregationProcessor(this, AggregatorType.Sum, Double.class);
@@ -135,7 +139,7 @@ public class TestProcessorQuery {
                 
                 @SuppressWarnings("unchecked")
                 Processor<Iterable<Number>, Number> retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Number>, Number>((Class<Iterable<Number>>)(Class<?>) Iterable.class, Number.class,
-                                                                                                                                                          ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers) {
+                                                                                                                                         ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 0) {
                     @Override
                     protected Iterable<Number> retrieveData(Iterable<Number> element) {
                         return element;
@@ -175,7 +179,7 @@ public class TestProcessorQuery {
         Collection<Number> dataSource = new ArrayList<>();
         dataSource.add(new Number(10));
         ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(
-                dataSource, stringMessages, Locale.ENGLISH) {
+                dataSource, stringMessages, Locale.ENGLISH, AdditionalQueryData.NULL_INSTANCE) {
             @SuppressWarnings("unchecked")
             @Override
             protected Processor<Iterable<Number>, ?> createFirstProcessor() {
@@ -197,6 +201,7 @@ public class TestProcessorQuery {
         } catch (TimeoutException e) {
             // A timeout exception is expected
         }
+        assertThat(query.getState(), is(QueryState.TIMED_OUT));
         
         ConcurrencyTestsUtil.sleepFor(1000); // Wait if a result is received
         assertThat("The processing should be aborted, but received elements", receivedElementOrFinished, is(false));
@@ -211,7 +216,7 @@ public class TestProcessorQuery {
         Collection<Number> dataSource = new ArrayList<>();
         dataSource.add(new Number(10));
         ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(
-                dataSource, stringMessages, Locale.ENGLISH) {
+                dataSource, stringMessages, Locale.ENGLISH, AdditionalQueryData.NULL_INSTANCE) {
             @SuppressWarnings("unchecked")
             @Override
             protected Processor<Iterable<Number>, ?> createFirstProcessor() {
@@ -234,7 +239,9 @@ public class TestProcessorQuery {
             }
         });
         queryRunner.start();
-        ConcurrencyTestsUtil.sleepFor(500);
+        ConcurrencyTestsUtil.sleepFor(250);
+        assertThat(query.getState(), is(QueryState.RUNNING));
+        ConcurrencyTestsUtil.sleepFor(250);
         query.abort();
         
         ConcurrencyTestsUtil.sleepFor(1000); // Wait if a result is received
@@ -275,7 +282,7 @@ public class TestProcessorQuery {
     public void testQueryWithTimeoutAndNonBlockingProcess() throws TimeoutException {
         final String keyValue = "Sum";
         ProcessorQuery<Double, Iterable<Number>> query = new ProcessorQuery<Double, Iterable<Number>>(
-                createDataSource(), stringMessages, Locale.ENGLISH) {
+                createDataSource(), stringMessages, Locale.ENGLISH, AdditionalQueryData.NULL_INSTANCE) {
             @SuppressWarnings("unchecked")
             @Override
             protected Processor<Iterable<Number>, ?> createFirstProcessor() {
@@ -286,8 +293,8 @@ public class TestProcessorQuery {
                                                                                                     ConcurrencyTestsUtil.getExecutor(),
                                                                                                     resultReceivers) {
                     @Override
-                    protected ProcessorInstruction<Map<GroupKey, Double>> createInstruction(final Iterable<Number> element) {
-                        return new ProcessorInstruction<Map<GroupKey,Double>>(this) {
+                    protected AbstractProcessorInstruction<Map<GroupKey, Double>> createInstruction(final Iterable<Number> element) {
+                        return new AbstractProcessorInstruction<Map<GroupKey,Double>>(this) {
                             @Override
                             public Map<GroupKey, Double> computeResult() {
                                 Map<GroupKey, Double> result = new HashMap<>();
@@ -312,5 +319,75 @@ public class TestProcessorQuery {
         expectedResult.put(new GenericGroupKey<String>(keyValue), 10358.0);
         assertThat(query.run(500, TimeUnit.MILLISECONDS).getResults(), is(expectedResult));
     }
+    
+    @Test
+    public void testQueryWithSevereFailure() {
+        Query<Double> query = new ProcessorQuery<Double, Double>(0.0) {
+            @SuppressWarnings("unchecked")
+            @Override
+            protected Processor<Double, ?> createFirstProcessor() {
+                Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers = new ArrayList<>();
+                resultReceivers.add(this.getResultReceiver());
+                return new AbstractSimpleParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
+                                                                                          (Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
+                                                                                          ConcurrencyTestsUtil.getExecutor(),
+                                                                                          resultReceivers) {
+                    @Override
+                    protected AbstractProcessorInstruction<Map<GroupKey, Double>> createInstruction(Double element) {
+                        return new AbstractProcessorInstruction<Map<GroupKey,Double>>(this) {
+                            @Override
+                            protected Map<GroupKey, Double> computeResult() throws Exception {
+                                throw new RejectedExecutionException("This should cause an error during the query processing");
+                            }
+                        };
+                    }
+                    @Override
+                    protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
+                    }
+                };
+            }
+        };
+        try {
+            query.run();
+            fail("The previous line should throw a runtime exception");
+        } catch (RuntimeException e) {
+            // A RuntimeException wiht a RejectedExecitonException is expected
+            if (!(e.getCause() instanceof RejectedExecutionException)) {
+                throw e;
+            }
+        }
+        assertThat(query.getState(), is(QueryState.ERROR));
+    }
+    
+    @Test
+    public void testQueryWithFailure() {
+        Query<Double> query = new ProcessorQuery<Double, Double>(0.0) {
+            @SuppressWarnings("unchecked")
+            @Override
+            protected Processor<Double, ?> createFirstProcessor() {
+                Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers = new ArrayList<>();
+                resultReceivers.add(this.getResultReceiver());
+                return new AbstractSimpleParallelProcessor<Double, Map<GroupKey, Double>>(Double.class,
+                                                                                          (Class<Map<GroupKey, Double>>)(Class<?>) Map.class,
+                                                                                          ConcurrencyTestsUtil.getExecutor(),
+                                                                                          resultReceivers) {
+                    @Override
+                    protected AbstractProcessorInstruction<Map<GroupKey, Double>> createInstruction(Double element) {
+                        return new AbstractProcessorInstruction<Map<GroupKey,Double>>(this) {
+                            @Override
+                            protected Map<GroupKey, Double> computeResult() throws Exception {
+                                throw new NullPointerException("This should cause a failure during the query processing");
+                            }
+                        };
+                    }
+                    @Override
+                    protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {
+                    }
+                };
+            }
+        };
+        query.run();
+        assertThat(query.getState(), is(QueryState.FAILURE));
+    }
 
 }

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

@@ -69,8 +69,8 @@ public class TestAbstractParallelProcessorFinishing {
     private AbstractSimpleParallelProcessor<Integer, Integer> createProcessor(Collection<Processor<Integer, ?>> receivers) {
         return new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
             @Override
-            protected ProcessorInstruction<Integer> createInstruction(Integer partialElement) {
-                return new ProcessorInstruction<Integer>(this) {
+            protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
+                return new AbstractProcessorInstruction<Integer>(this) {
                     @Override
                     public Integer computeResult() {
                         while (instructionIsWorking) {

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

@@ -38,8 +38,8 @@ public class TestAbstractParallelProcessorWithManySimpleInstructions {
         receivers.add(receiver);
         processor = new AbstractSimpleParallelProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers) {
             @Override
-            protected ProcessorInstruction<Integer> createInstruction(final Integer element) {
-                return new ProcessorInstruction<Integer>(this) {
+            protected AbstractProcessorInstruction<Integer> createInstruction(final Integer element) {
+                return new AbstractProcessorInstruction<Integer>(this) {
                     @Override
                     public Integer computeResult() {
                         return element;

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

@@ -27,8 +27,8 @@ public class TestPartitioningOfAbstractSimpleParallelProcessor {
             }
 
             @Override
-            protected ProcessorInstruction<Integer> createInstruction(Integer partialElement) {
-                return new ProcessorInstruction<Integer>(this) {
+            protected AbstractProcessorInstruction<Integer> createInstruction(Integer partialElement) {
+                return new AbstractProcessorInstruction<Integer>(this) {
                     @Override
                     public Integer computeResult() {
                         return 0;

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

@@ -0,0 +1,34 @@
+package com.sap.sse.datamining.impl.components;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import java.util.PriorityQueue;
+import java.util.Queue;
+
+import org.junit.Test;
+
+import com.sap.sse.datamining.test.util.components.NullProcessorInstruction;
+
+public class TestProcessorInstructionOrdering {
+
+    @Test
+    public void testOrderInPriorityQueue() {
+        Queue<NullProcessorInstruction<?>> priorityQueue = new PriorityQueue<>();
+
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.createRetrievalPriority(0)));
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.createRetrievalPriority(1)));
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.Grouping));
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.Extraction));
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.Filtration));
+        priorityQueue.add(new NullProcessorInstruction<>(null, ProcessorInstructionPriority.Aggregation));
+
+        assertThat(priorityQueue.poll().getPriority(), is(0));
+        assertThat(priorityQueue.poll().getPriority(), is(1));
+        assertThat(priorityQueue.poll().getPriority(), is(1));
+        assertThat(priorityQueue.poll().getPriority(), is(2));
+        assertThat(priorityQueue.poll().getPriority(), is(Integer.MAX_VALUE - 1));
+        assertThat(priorityQueue.poll().getPriority(), is(Integer.MAX_VALUE));
+    }
+
+}

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

@@ -72,7 +72,7 @@ public class TestRetrieverFilterProcessorChain {
         resultReceivers.add(retrievalProcessor);
         @SuppressWarnings("unchecked")
         Processor<Iterable<Iterable<Integer>>, Iterable<Integer>> layeredRetrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Iterable<Integer>>, Iterable<Integer>>((Class<Iterable<Iterable<Integer>>>)(Class<?>) Iterable.class, (Class<Iterable<Integer>>)(Class<?>) Iterable.class,
-                                                                                                                                                                                    ConcurrencyTestsUtil.getExecutor(), resultReceivers) {
+                                                                                                                                                                                    ConcurrencyTestsUtil.getExecutor(), resultReceivers, 0) {
             @Override
             protected Iterable<Iterable<Integer>> retrieveData(Iterable<Iterable<Integer>> element) {
                 return element;
@@ -114,7 +114,7 @@ public class TestRetrieverFilterProcessorChain {
         
         Collection<Processor<Integer, ?>> retrievalResultReceivers = new ArrayList<>();
         retrievalResultReceivers.add(filtrationProcessor);
-        retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers) {
+        retrievalProcessor = new AbstractSimpleRetrievalProcessor<Iterable<Integer>, Integer>((Class<Iterable<Integer>>)(Class<?>) Iterable.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), retrievalResultReceivers, 1) {
             @Override
             protected Iterable<Integer> retrieveData(Iterable<Integer> element) {
                 return element;

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

@@ -8,19 +8,18 @@ import static org.junit.Assert.fail;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.DataMiningExecutorService;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public class ConcurrencyTestsUtil extends TestsUtil {
-
+    
     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 final ExecutorService executor = new DataMiningExecutorService(THREAD_POOL_SIZE);
 
-    public static ThreadPoolExecutor getExecutor() {
+    public static ExecutorService getExecutor() {
         return executor;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/BlockingInstructionWrapper.java

@@ -1,31 +0,0 @@
-package com.sap.sse.datamining.test.util.components;
-
-import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
-
-public class BlockingInstructionWrapper implements Runnable {
-    
-    private Runnable innerInstruction;
-
-    private boolean workIsInProgress;
-
-    public BlockingInstructionWrapper(Runnable innerInstruction) {
-        this.innerInstruction = innerInstruction;
-        workIsInProgress = true;
-    }
-
-    @Override
-    public void run() {
-        while (workIsInProgress) {
-            ConcurrencyTestsUtil.sleepFor(100);
-        }
-        innerInstruction.run();
-    }
-    
-    /**
-     * Tells the instruction, that the work is now done. It won't block it's inner instruction anymore and calls the run method of it.
-     */
-    public void workIsNowFinished() {
-        workIsInProgress = false;
-    }
-
-}

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/BlockingProcessor.java

@@ -6,7 +6,7 @@ import java.util.concurrent.ExecutorService;
 import com.sap.sse.datamining.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
-import com.sap.sse.datamining.impl.components.ProcessorInstruction;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 
 public abstract class BlockingProcessor<InputType, ResultType> extends AbstractSimpleParallelProcessor<InputType, ResultType> {
     private final long timeToBlockInMillis;
@@ -19,8 +19,8 @@ public abstract class BlockingProcessor<InputType, ResultType> extends AbstractS
     }
 
     @Override
-    protected ProcessorInstruction<ResultType> createInstruction(InputType element) {
-        return new ProcessorInstruction<ResultType>(this) {
+    protected AbstractProcessorInstruction<ResultType> createInstruction(InputType element) {
+        return new AbstractProcessorInstruction<ResultType>(this) {
             @Override
             public ResultType computeResult() throws Exception {
                 Thread.sleep(timeToBlockInMillis);

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/NullProcessorInstruction.java

@@ -0,0 +1,23 @@
+package com.sap.sse.datamining.test.util.components;
+
+import com.sap.sse.datamining.impl.components.AbstractPartitioningParallelProcessor;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
+import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
+
+public class NullProcessorInstruction<ResultType> extends AbstractProcessorInstruction<ResultType> {
+
+    public NullProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
+            ProcessorInstructionPriority priority) {
+        super(processor, priority);
+    }
+
+    @Override
+    protected ResultType computeResult() throws Exception {
+        return null;
+    }
+    
+    public int getPriority() {
+        return super.getPriority();
+    }
+
+}

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestLegOfCompetitorWithContextRetrievalProcessor.java

@@ -15,8 +15,8 @@ import com.sap.sse.datamining.test.functions.registry.test_contexts.Test_HasRace
 public class TestLegOfCompetitorWithContextRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<Test_HasRaceContext, Test_HasLegOfCompetitorContext>{
 
-    public TestLegOfCompetitorWithContextRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_HasLegOfCompetitorContext, ?>> resultReceivers) {
-        super(Test_HasRaceContext.class, Test_HasLegOfCompetitorContext.class, executor, resultReceivers);
+    public TestLegOfCompetitorWithContextRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_HasLegOfCompetitorContext, ?>> resultReceivers, int retrievalLevel) {
+        super(Test_HasRaceContext.class, Test_HasLegOfCompetitorContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestRaceWithContextRetrievalProcessor.java

@@ -15,8 +15,8 @@ public class TestRaceWithContextRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<Test_Regatta, Test_HasRaceContext> {
 
     public TestRaceWithContextRetrievalProcessor(ExecutorService executor,
-            Collection<Processor<Test_HasRaceContext, ?>> resultReceivers) {
-        super(Test_Regatta.class, Test_HasRaceContext.class, executor, resultReceivers);
+            Collection<Processor<Test_HasRaceContext, ?>> resultReceivers, int retrievalLevel) {
+        super(Test_Regatta.class, Test_HasRaceContext.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/TestRegattaRetrievalProcessor.java

@@ -11,8 +11,8 @@ public class TestRegattaRetrievalProcessor extends
         AbstractSimpleRetrievalProcessor<Collection<Test_Regatta>, Test_Regatta> {
 
     @SuppressWarnings("unchecked")
-    public TestRegattaRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_Regatta, ?>> resultReceivers) {
-        super((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_Regatta.class, executor, resultReceivers);
+    public TestRegattaRetrievalProcessor(ExecutorService executor, Collection<Processor<Test_Regatta, ?>> resultReceivers, int retrievalLevel) {
+        super((Class<Collection<Test_Regatta>>)(Class<?>) Collection.class, Test_Regatta.class, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

java/com.sap.sse.datamining/src/com/sap/sse/datamining/AdditionalQueryData.java

@@ -0,0 +1,16 @@
+package com.sap.sse.datamining;
+
+import java.util.UUID;
+
+import com.sap.sse.datamining.Query.QueryType;
+import com.sap.sse.datamining.impl.NullAdditionalQueryData;
+
+public interface AdditionalQueryData {
+
+    public AdditionalQueryData NULL_INSTANCE = new NullAdditionalQueryData();
+
+    public QueryType getType();
+
+    public UUID getDataRetrieverChainID();
+
+}

java/com.sap.sse.datamining/src/com/sap/sse/datamining/Query.java

@@ -7,8 +7,12 @@ import com.sap.sse.datamining.shared.QueryResult;
 
 public interface Query<AggregatedType> {
     
+    public enum QueryType { STATISTIC, DIMENSION_VALUES, OTHER }
+    
     public QueryState getState();
     
+    public AdditionalQueryData getAdditionalData();
+    
     public QueryResult<AggregatedType> run();
 
     public QueryResult<AggregatedType> run(long timeout, TimeUnit unit) throws TimeoutException;

java/com.sap.sse.datamining/src/com/sap/sse/datamining/factories/QueryFactory.java

@@ -15,11 +15,13 @@ import com.sap.sse.datamining.DataRetrieverChainBuilder;
 import com.sap.sse.datamining.DataRetrieverChainDefinition;
 import com.sap.sse.datamining.Query;
 import com.sap.sse.datamining.QueryDefinition;
+import com.sap.sse.datamining.Query.QueryType;
 import com.sap.sse.datamining.components.FilterCriterion;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.functions.Function;
 import com.sap.sse.datamining.functions.ParameterProvider;
 import com.sap.sse.datamining.impl.ProcessorQuery;
+import com.sap.sse.datamining.impl.SimpleAdditionalQueryData;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.criterias.AndCompoundFilterCriterion;
 import com.sap.sse.datamining.impl.criterias.CompoundFilterCriterion;
@@ -32,7 +34,7 @@ public class QueryFactory {
 
     public <DataSourceType, DataType, ResultType> Query<ResultType> createQuery(DataSourceType dataSource, QueryDefinition<DataSourceType, DataType, ResultType> queryDefinition,
                                                                             ResourceBundleStringMessages stringMessages, ExecutorService executor) {
-        return new ProcessorQuery<ResultType, DataSourceType>(dataSource, stringMessages, queryDefinition.getLocale()) {
+        return new ProcessorQuery<ResultType, DataSourceType>(dataSource, stringMessages, queryDefinition.getLocale(), new SimpleAdditionalQueryData(QueryType.STATISTIC, queryDefinition.getDataRetrieverChainDefinition().getID())) {
             @Override
             protected Processor<DataSourceType, ?> createFirstProcessor() {
                 ProcessorFactory processorFactory = new ProcessorFactory(executor);
@@ -118,7 +120,7 @@ public class QueryFactory {
             DataRetrieverChainDefinition<DataSource, ?> dataRetrieverChainDefinition, int retrieverLevel,
             Iterable<Function<?>> dimensions, Locale locale,
             ResourceBundleStringMessages stringMessages, ExecutorService executor) {
-        return new ProcessorQuery<Set<Object>, DataSource>(dataSource, stringMessages, locale) {
+        return new ProcessorQuery<Set<Object>, DataSource>(dataSource, stringMessages, locale, new SimpleAdditionalQueryData(QueryType.DIMENSION_VALUES, dataRetrieverChainDefinition.getID())) {
             @Override
             protected Processor<DataSource, ?> createFirstProcessor() {
                 ProcessorFactory processorFactory = new ProcessorFactory(executor);

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

@@ -0,0 +1,13 @@
+package com.sap.sse.datamining.impl;
+
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class DataMiningExecutorService extends ThreadPoolExecutor {
+    
+    public DataMiningExecutorService(int corePoolSize) {
+        super(corePoolSize, corePoolSize, 60, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>());
+    }
+
+}

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

@@ -3,10 +3,6 @@ package com.sap.sse.datamining.impl;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 
 import org.osgi.framework.BundleActivator;
@@ -27,8 +23,8 @@ import com.sap.sse.i18n.impl.ResourceBundleStringMessagesImpl;
 
 public class DataMiningFrameworkActivator implements BundleActivator {
     private static final Logger logger = Logger.getLogger(DataMiningFrameworkActivator.class.getName());
-    private static final int THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 3);
     private static final String STRING_MESSAGES_BASE_NAME = "stringmessages/StringMessages";
+    private static final int THREAD_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 3);
     
     private static DataMiningFrameworkActivator INSTANCE;
 
@@ -38,13 +34,7 @@ public class DataMiningFrameworkActivator implements BundleActivator {
     private final ModifiableDataMiningServer dataMiningServer;
     
     public DataMiningFrameworkActivator() {
-        ExecutorService executor = new ThreadPoolExecutor(THREAD_POOL_SIZE, THREAD_POOL_SIZE, 60, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>(), new RejectedExecutionHandler() {
-                    @Override
-                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
-                        r.run();
-                    }
-                });
+        ExecutorService executor = new DataMiningExecutorService(THREAD_POOL_SIZE);
 
         FunctionManager functionManager = new FunctionManager();
         DataRetrieverChainDefinitionRegistry dataRetrieverChainDefinitionRegistry = new SimpleDataRetrieverChainDefinitionRegistry();

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

@@ -47,7 +47,7 @@ public class DataMiningServerImpl implements ModifiableDataMiningServer {
         this.stringMessages = new CompoundResourceBundleStringMessages();
         this.executorService = executorService;
         this.queryFactory = new QueryFactory();
-        dataMiningQueryManager = new ConcurrentDataMiningQueryManager();
+        dataMiningQueryManager = new StrategyPerQueryTypeManager();
         this.functionRegistry = functionRegistry;
         this.functionProvider = functionProvider;
         dataSourceProviderMappedByDataSourceType = new HashMap<>();

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

@@ -0,0 +1,81 @@
+package com.sap.sse.datamining.impl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import com.sap.sse.datamining.DataMiningQueryManager;
+import com.sap.sse.datamining.Query;
+import com.sap.sse.datamining.shared.DataMiningSession;
+import com.sap.sse.datamining.shared.QueryResult;
+
+public class ManyQueriesPerRetrieverChainManager implements DataMiningQueryManager {
+
+    private final ConcurrentMap<DataMiningSession, UUID> currentRetrieverChainMappedBySession;
+    private final Map<DataMiningSession, Collection<Query<?>>> queriesMappedBySession;
+
+    public ManyQueriesPerRetrieverChainManager() {
+        currentRetrieverChainMappedBySession = new ConcurrentHashMap<>();
+        queriesMappedBySession = new HashMap<>();
+    }
+
+    @Override
+    public <ResultType> QueryResult<ResultType> runNewAndAbortPrevious(DataMiningSession session, Query<ResultType> query) {
+        if (session == null || query == null) {
+            throw new NullPointerException();
+        }
+        
+        abortPreviousQueries(session, query);
+        updateCurrentRetrieverChain(session, query);
+        registerNewQuery(session, query);
+        QueryResult<ResultType> result = query.run();
+        unregisterQuery(session, query);
+        return result;
+    }
+    
+    private void abortPreviousQueries(DataMiningSession session, Query<?> newQuery) {
+        final UUID newRetrieverChainID = newQuery.getAdditionalData().getDataRetrieverChainID();
+        if (currentRetrieverChainMappedBySession.containsKey(session) &&
+                !currentRetrieverChainMappedBySession.get(session).equals(newRetrieverChainID)) {
+            Collection<Query<?>> previousQueries = queriesMappedBySession.get(session);
+            for (Query<?> previousQuery : previousQueries) {
+                previousQuery.abort();
+            }
+            queriesMappedBySession.remove(session, previousQueries);
+            
+            currentRetrieverChainMappedBySession.remove(session);
+            currentRetrieverChainMappedBySession.putIfAbsent(session, newRetrieverChainID);
+        }
+    }
+
+    private void updateCurrentRetrieverChain(DataMiningSession session, Query<?> query) {
+        UUID retrieverChainID = query.getAdditionalData().getDataRetrieverChainID();
+        UUID previousRetrieverChainID = currentRetrieverChainMappedBySession.putIfAbsent(session, retrieverChainID);
+        if (previousRetrieverChainID != null && !previousRetrieverChainID.equals(retrieverChainID)) {
+            currentRetrieverChainMappedBySession.replace(session, previousRetrieverChainID, retrieverChainID);
+        }
+    }
+
+    private <ResultType> void registerNewQuery(DataMiningSession session, Query<ResultType> query) {
+        if (!queriesMappedBySession.containsKey(session)) {
+            Set<Query<?>> concurrentSet = Collections.newSetFromMap(new ConcurrentHashMap<Query<?>, Boolean>());
+            queriesMappedBySession.put(session, concurrentSet);
+        }
+        
+        queriesMappedBySession.get(session).add(query);
+    }
+
+    private void unregisterQuery(DataMiningSession session, Query<?> finishedQuery) {
+        UUID retrieverChainID = finishedQuery.getAdditionalData().getDataRetrieverChainID();
+        if (currentRetrieverChainMappedBySession.containsKey(session) &&
+                currentRetrieverChainMappedBySession.get(session).equals(retrieverChainID)) {
+            queriesMappedBySession.get(session).remove(finishedQuery);
+        }
+    }
+
+}

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

@@ -0,0 +1,31 @@
+package com.sap.sse.datamining.impl;
+
+import java.util.UUID;
+
+import com.sap.sse.datamining.AdditionalQueryData;
+import com.sap.sse.datamining.Query.QueryType;
+
+/**
+ * Null Object pattern for {@link AdditionalQueryData} with the values:
+ * <ul>
+ *   <li><b>Type</b>: {@link QueryType#OTHER}</li>
+ *   <li><b>Data Retriever Chain ID</b>: {@link UUID#UUID(long, long) UUID(0, 0)}</li>
+ * </ul>
+ * 
+ * @author Lennart Hensler (D054527)
+ */
+public class NullAdditionalQueryData implements AdditionalQueryData {
+    
+    private final UUID nullUUID = new UUID(0, 0);
+
+    @Override
+    public QueryType getType() {
+        return QueryType.OTHER;
+    }
+
+    @Override
+    public UUID getDataRetrieverChainID() {
+        return nullUUID;
+    }
+
+}

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

@@ -14,14 +14,17 @@ import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import com.sap.sse.datamining.AdditionalQueryData;
 import com.sap.sse.datamining.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.Query;
 import com.sap.sse.datamining.QueryState;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.OverwritingResultDataBuilder;
+import com.sap.sse.datamining.shared.AdditionalResultData;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.shared.QueryResult;
 import com.sap.sse.datamining.shared.QueryResultState;
+import com.sap.sse.datamining.shared.impl.NullAdditionalResultData;
 import com.sap.sse.datamining.shared.impl.QueryResultImpl;
 import com.sap.sse.i18n.ResourceBundleStringMessages;
 
@@ -36,26 +39,33 @@ public abstract class ProcessorQuery<AggregatedType, DataSourceType> implements
     
     private final ResourceBundleStringMessages stringMessages;
     private final Locale locale;
+    private final AdditionalQueryData additionalData;
 
     private final Object monitorObject = new Object();
     private Thread workingThread;
     private QueryState state;
     
     /**
-     * Creates a query that returns a result without any additional data (like the calculation time or the retrieved data amount).<br>
+     * Creates a query
+     * <ul>
+     *   <li> with no {@link AdditionalQueryData} (more exactly with {@link NullAdditionalQueryData} as additional data).</li>
+     *   <li> that returns a result without {@link AdditionalResultData} (more exactly with {@link NullAdditionalResultData} as additional data).</li>
+     * </ul>
+     * 
      * This is useful for non user specific queries, like retrieving the dimension values.
      */
     public ProcessorQuery(DataSourceType dataSource) {
-        this(dataSource, null, null);
+        this(dataSource, null, null, AdditionalQueryData.NULL_INSTANCE);
     }
 
     /**
      * Creates a query that returns a result with additional data.
      */
-    public ProcessorQuery(DataSourceType dataSource, ResourceBundleStringMessages stringMessages, Locale locale) {
+    public ProcessorQuery(DataSourceType dataSource, ResourceBundleStringMessages stringMessages, Locale locale, AdditionalQueryData additionalData) {
         this.dataSource = dataSource;
         this.stringMessages = stringMessages;
         this.locale = locale;
+        this.additionalData = additionalData;
         state = QueryState.NOT_STARTED;
 
         resultReceiver = new ProcessResultReceiver();
@@ -68,6 +78,11 @@ public abstract class ProcessorQuery<AggregatedType, DataSourceType> implements
     public QueryState getState() {
         return state;
     }
+    
+    @Override
+    public AdditionalQueryData getAdditionalData() {
+        return additionalData;
+    }
 
     @Override
     public QueryResult<AggregatedType> run() {
@@ -238,7 +253,7 @@ public abstract class ProcessorQuery<AggregatedType, DataSourceType> implements
         @Override
         public void finish() throws InterruptedException {
             synchronized (monitorObject) {
-                state = QueryState.NORMAL;
+                state = state == QueryState.RUNNING ? QueryState.NORMAL : state;
                 monitorObject.notify();
             }
         }

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

@@ -0,0 +1,28 @@
+package com.sap.sse.datamining.impl;
+
+import java.util.UUID;
+
+import com.sap.sse.datamining.AdditionalQueryData;
+import com.sap.sse.datamining.Query.QueryType;
+
+public class SimpleAdditionalQueryData implements AdditionalQueryData {
+    
+    private final QueryType type;
+    private final UUID dataRetrieverChainID;
+    
+    public SimpleAdditionalQueryData(QueryType type, UUID dataRetrieverChainID) {
+        this.type = type;
+        this.dataRetrieverChainID = dataRetrieverChainID;
+    }
+
+    @Override
+    public QueryType getType() {
+        return type;
+    }
+
+    @Override
+    public UUID getDataRetrieverChainID() {
+        return dataRetrieverChainID;
+    }
+
+}

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

@@ -146,24 +146,24 @@ public class SimpleDataRetrieverChainBuilder<DataSourceType> implements DataRetr
         FilterCriterion<ResultType> filter = (FilterCriterion<ResultType>) filters.get(retrieverTypeIndex);
         
         Class<Processor<?, ResultType>> retrieverType = (Class<Processor<?, ResultType>>)(Class<?>) dataRetrieverTypeWithInformation.getRetrieverType();
-        return createRetriever(retrieverType, retrievedDataType, resultReceivers, filter);
+        return createRetriever(retrieverType, retrievedDataType, resultReceivers, filter, retrieverTypeIndex);
     }
 
     private <ResultType> Processor<?, ResultType> createRetriever(Class<Processor<?, ResultType>> retrieverType, Class<ResultType> retrievedDataType,
-            Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter) {
+            Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter, int retrieverTypeIndex) {
         Constructor<Processor<?, ResultType>> retrieverConstructor = null;
         try {
-            retrieverConstructor = retrieverType.getConstructor(ExecutorService.class, Collection.class);
+            retrieverConstructor = retrieverType.getConstructor(ExecutorService.class, Collection.class, int.class);
         } catch (NoSuchMethodException | SecurityException e) {
             throw new IllegalArgumentException("Couldn't get an usable constructor from the given retrieverType '"
                     + retrieverType.getSimpleName() + "'", e);
         }
         
-        return constructRetriever(retrieverConstructor, retrievedDataType, resultReceivers, filter);
+        return constructRetriever(retrieverConstructor, retrievedDataType, resultReceivers, filter, retrieverTypeIndex);
     }
 
     private <ResultType> Processor<?, ResultType> constructRetriever(Constructor<Processor<?, ResultType>> retrieverConstructor, Class<ResultType> retrievedDataType,
-            Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter) {
+            Collection<Processor<ResultType, ?>> resultReceivers, FilterCriterion<ResultType> filter, int retrieverTypeIndex) {
         try {
             Collection<Processor<ResultType, ?>> retrievalResultReceivers = resultReceivers;
             if (filter != null) {
@@ -173,7 +173,7 @@ public class SimpleDataRetrieverChainBuilder<DataSourceType> implements DataRetr
                 retrievalResultReceivers.add(filteringProcessor);
             }
 
-            return retrieverConstructor.newInstance(executor, retrievalResultReceivers);
+            return retrieverConstructor.newInstance(executor, retrievalResultReceivers, retrieverTypeIndex);
         } catch (InstantiationException | IllegalAccessException |
                  IllegalArgumentException | InvocationTargetException e) {
             throw new UnsupportedOperationException("Couldn't create a data retriever instance with the constructor "

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

@@ -105,7 +105,7 @@ public class SimpleDataRetrieverChainDefinition<DataSourceType, DataType> implem
     private <InputType, ResultType> void checkThatRetrieverHasUsableConstructor(
             Class<? extends Processor<InputType, ResultType>> retrieverType) {
         try {
-            retrieverType.getConstructor(ExecutorService.class, Collection.class);
+            retrieverType.getConstructor(ExecutorService.class, Collection.class, int.class);
         } catch (NoSuchMethodException | SecurityException e) {
             throw new IllegalArgumentException("Couldn't get an usable constructor from the given nextRetrieverType '"
                     + retrieverType.getSimpleName() + "'", e);

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

@@ -8,11 +8,11 @@ import com.sap.sse.datamining.Query;
 import com.sap.sse.datamining.shared.DataMiningSession;
 import com.sap.sse.datamining.shared.QueryResult;
 
-public class ConcurrentDataMiningQueryManager implements DataMiningQueryManager {
+public class SingleQueryPerSessionManager implements DataMiningQueryManager {
 
     private final ConcurrentMap<DataMiningSession, Query<?>> queryMappedBySession;
 
-    public ConcurrentDataMiningQueryManager() {
+    public SingleQueryPerSessionManager() {
         queryMappedBySession = new ConcurrentHashMap<>();
     }
 
@@ -23,15 +23,14 @@ public class ConcurrentDataMiningQueryManager implements DataMiningQueryManager
             throw new NullPointerException();
         }
         
-        abortPreviousQueries(session, query);
+        abortPreviousQueries(session);
         registerNewQuery(session, query);
         QueryResult<ResultType> result = query.run();
         unregisterQuery(session, query);
         return result;
     }
     
-    private void abortPreviousQueries(DataMiningSession session, Query<?> query) {
-        // TODO handle different types of queries (a statistics query doesn't have to be aborted, if a new dimension values query wants to run)
+    private void abortPreviousQueries(DataMiningSession session) {
         if (queryMappedBySession.containsKey(session)) {
             Query<?> previousQuery = queryMappedBySession.get(session);
             previousQuery.abort();
@@ -46,8 +45,8 @@ public class ConcurrentDataMiningQueryManager implements DataMiningQueryManager
         }
     }
 
-    private void unregisterQuery(DataMiningSession session, Query<?> query) {
-        queryMappedBySession.remove(session, query);
+    private void unregisterQuery(DataMiningSession session, Query<?> finishedQuery) {
+        queryMappedBySession.remove(session, finishedQuery);
     }
     
 }

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

@@ -0,0 +1,28 @@
+package com.sap.sse.datamining.impl;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.sap.sse.datamining.DataMiningQueryManager;
+import com.sap.sse.datamining.Query;
+import com.sap.sse.datamining.Query.QueryType;
+import com.sap.sse.datamining.shared.DataMiningSession;
+import com.sap.sse.datamining.shared.QueryResult;
+
+public class StrategyPerQueryTypeManager implements DataMiningQueryManager {
+    
+    private final Map<QueryType, DataMiningQueryManager> managersMappedByQueryType;
+
+    public StrategyPerQueryTypeManager() {
+        managersMappedByQueryType = new HashMap<>();
+        managersMappedByQueryType.put(QueryType.STATISTIC, new SingleQueryPerSessionManager());
+        managersMappedByQueryType.put(QueryType.DIMENSION_VALUES, new ManyQueriesPerRetrieverChainManager());
+        managersMappedByQueryType.put(QueryType.OTHER, new SingleQueryPerSessionManager());
+    }
+
+    @Override
+    public <ResultType> QueryResult<ResultType> runNewAndAbortPrevious(DataMiningSession session, Query<ResultType> query) {
+        return managersMappedByQueryType.get(query.getAdditionalData().getType()).runNewAndAbortPrevious(session, query);
+    }
+
+}

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

@@ -46,8 +46,8 @@ public abstract class AbstractParallelMultiDimensionalNestingGroupingProcessor<D
     }
 
     @Override
-    protected ProcessorInstruction<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
-        return new ProcessorInstruction<GroupedDataEntry<DataType>>(this) {
+    protected AbstractProcessorInstruction<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
+        return new AbstractProcessorInstruction<GroupedDataEntry<DataType>>(this, ProcessorInstructionPriority.Grouping) {
             @Override
             public GroupedDataEntry<DataType> computeResult() {
                 return new GroupedDataEntry<DataType>(createCompoundKeyFor(element,

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

@@ -36,7 +36,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
     public void processElement(InputType element) {
         if (!isFinished && !isAborted) {
             for (WorkingType partialElement : partitionElement(element)) {
-                final ProcessorInstruction<ResultType> instruction = createInstruction(partialElement);
+                final AbstractProcessorInstruction<ResultType> instruction = createInstruction(partialElement);
                 if (isInstructionValid(instruction)) {
                     unfinishedInstructionsCounter.getAndIncrement();
                     try {
@@ -51,7 +51,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
         }
     }
 
-    private boolean isInstructionValid(ProcessorInstruction<ResultType> instruction) {
+    private boolean isInstructionValid(AbstractProcessorInstruction<ResultType> instruction) {
         return instruction != null;
     }
 
@@ -76,7 +76,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
         }
     }
 
-    protected abstract ProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement);
+    protected abstract AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement);
 
     protected abstract Iterable<WorkingType> partitionElement(InputType element);
 

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

@@ -0,0 +1,76 @@
+package com.sap.sse.datamining.impl.components;
+
+public abstract class AbstractProcessorInstruction<ResultType> implements Runnable, Comparable<AbstractProcessorInstruction<?>> {
+    
+    private final AbstractPartitioningParallelProcessor<?, ?, ResultType> processor;
+    private final int priority;
+
+    /**
+     * Creates an instruction with a priority of 0, which is the highest priority.<br />
+     * Use only, if <b>no processor</b> in the chain creates instruction <b>with</b> priorities other than 0 or
+     * if the chain doesn't use priorities at all.
+     * If this isn't the case use
+     * {@link #ProcessorInstruction(AbstractPartitioningParallelProcessor, ProcessorInstructionPriority)}
+     * instead.
+     * 
+     * @param processor The processor, that processes this instruction
+     */
+    public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
+        this(processor, 0);
+    }
+    
+    /**
+     * Creates an instruction with the given {@linkplain ProcessorInstructionPriority priority}.
+     * 
+     * @param processor The processor, that processes this instruction
+     * @param priority The priority of this instruction
+     */
+    public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
+                                ProcessorInstructionPriority priority) {
+        this(processor, priority.asIntValue());
+    }
+
+    /**
+     * Creates an instruction with the given priority as <code>int</code>.<br />
+     * <b>Use only, if you need instructions with custom priorities and if you know what you're doing!</b>
+     * If this isn't the case use
+     * {@link #ProcessorInstruction(AbstractPartitioningParallelProcessor, ProcessorInstructionPriority)}
+     * instead.
+     * 
+     * @param processor The processor, that processes this instruction
+     * @param priority The priority of this instruction. <code>0</code> is the highest priority.
+     */
+    public AbstractProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor,
+                                int priority) {
+        this.processor = processor;
+        this.priority = priority;
+    }
+
+    @Override
+    public void run() {
+        try {
+            ResultType result = computeResult();
+            if (processor.isResultValid(result) && !processor.isAborted()) {
+                processor.forwardResultToReceivers(result);
+            }
+        } catch (Exception e) {
+            if (!processor.isAborted() || !(e instanceof InterruptedException)) {
+                processor.onFailure(e);
+            }
+        } finally {
+            processor.getUnfinishedInstructionsCounter().getAndDecrement();
+        }
+    }
+
+    protected abstract ResultType computeResult() throws Exception;
+    
+    protected int getPriority() {
+        return priority;
+    }
+    
+    @Override
+    public int compareTo(AbstractProcessorInstruction<?> instruction) {
+        return Integer.compare(priority, instruction.priority);
+    }
+
+}

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

@@ -11,17 +11,29 @@ import com.sap.sse.datamining.components.Processor;
 public abstract class AbstractRetrievalProcessor<InputType, WorkingType, ResultType> extends
         AbstractPartitioningParallelProcessor<InputType, WorkingType, ResultType> {
 
+    private final int retrievalLevel;
+    
     private final Lock retrievedDataAmountLock;
     private int retrievedDataAmount;
 
-    public AbstractRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
+    /**
+     * 
+     * @param inputType
+     * @param resultType
+     * @param executor
+     * @param resultReceivers
+     * @param retrievalLevel The position of this retriever in it's chain. <code>0</code> represents the first.
+     */
+    public AbstractRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType,
+            ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers, int retrievalLevel) {
         super(inputType, resultType, executor, resultReceivers);
+        this.retrievalLevel = retrievalLevel;
         retrievedDataAmountLock = new ReentrantLock(); 
     }
 
     @Override
-    protected ProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
-        return new ProcessorInstruction<ResultType>(this) {
+    protected AbstractProcessorInstruction<ResultType> createInstruction(final WorkingType partialElement) {
+        return new AbstractProcessorInstruction<ResultType>(this, ProcessorInstructionPriority.createRetrievalPriority(retrievalLevel)) {
             @Override
             public ResultType computeResult() {
                 incrementRetrievedDataAmount();

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

@@ -21,6 +21,6 @@ public abstract class AbstractSimpleParallelProcessor<InputType, ResultType>
     //Redefinition of the method to set the parameter name to element instead of partial element.
     //This makes the implementation of sub classes more fluent.
     @Override
-    protected abstract ProcessorInstruction<ResultType> createInstruction(InputType element);
+    protected abstract AbstractProcessorInstruction<ResultType> createInstruction(InputType element);
 
 }

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

@@ -8,8 +8,17 @@ import com.sap.sse.datamining.components.Processor;
 public abstract class AbstractSimpleRetrievalProcessor<InputType, ResultType> extends
         AbstractRetrievalProcessor<InputType, ResultType, ResultType> {
 
-    public AbstractSimpleRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType, ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers) {
-        super(inputType, resultType, executor, resultReceivers);
+    /**
+     * 
+     * @param inputType
+     * @param resultType
+     * @param executor
+     * @param resultReceivers
+     * @param retrievalLevel The position of this retriever in it's chain. <code>0</code> represents the first.
+     */
+    public AbstractSimpleRetrievalProcessor(Class<InputType> inputType, Class<ResultType> resultType,
+            ExecutorService executor, Collection<Processor<ResultType, ?>> resultReceivers, int retrievalLevel) {
+        super(inputType, resultType, executor, resultReceivers, retrievalLevel);
     }
 
     @Override

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

@@ -23,8 +23,8 @@ public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParalle
     }
 
     @Override
-    protected ProcessorInstruction<InputType> createInstruction(final InputType element) {
-        return new ProcessorInstruction<InputType>(this) {
+    protected AbstractProcessorInstruction<InputType> createInstruction(final InputType element) {
+        return new AbstractProcessorInstruction<InputType>(this, ProcessorInstructionPriority.Filtration) {
             @Override
             public InputType computeResult() {
                 if (filterCriterion.matches(element)) {

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

@@ -30,8 +30,8 @@ public class ParallelGroupedElementsValueExtractionProcessor<DataType, FunctionR
     }
 
     @Override
-    protected ProcessorInstruction<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<DataType> element) {
-        return new ProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this) {
+    protected AbstractProcessorInstruction<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<DataType> element) {
+        return new AbstractProcessorInstruction<GroupedDataEntry<FunctionReturnType>>(this, ProcessorInstructionPriority.Extraction) {
             @Override
             public GroupedDataEntry<FunctionReturnType> computeResult() {
                 FunctionReturnType value = extractionFunction.tryToInvoke(element.getDataEntry(), parameterProvider);

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

@@ -1,29 +0,0 @@
-package com.sap.sse.datamining.impl.components;
-
-public abstract class ProcessorInstruction<ResultType> implements Runnable {
-    
-    private final AbstractPartitioningParallelProcessor<?, ?, ResultType> processor;
-
-    public ProcessorInstruction(AbstractPartitioningParallelProcessor<?, ?, ResultType> processor) {
-        this.processor = processor;
-    }
-
-    @Override
-    public void run() {
-        try {
-            ResultType result = computeResult();
-            if (processor.isResultValid(result) && !processor.isAborted()) {
-                processor.forwardResultToReceivers(result);
-            }
-        } catch (Exception e) {
-            if (!processor.isAborted() || !(e instanceof InterruptedException)) {
-                processor.onFailure(e);
-            }
-        } finally {
-            processor.getUnfinishedInstructionsCounter().getAndDecrement();
-        }
-    }
-
-    protected abstract ResultType computeResult() throws Exception;
-
-}

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

@@ -0,0 +1,33 @@
+package com.sap.sse.datamining.impl.components;
+
+public class ProcessorInstructionPriority {
+
+    public static final ProcessorInstructionPriority Aggregation = new ProcessorInstructionPriority(0);
+    public static final ProcessorInstructionPriority Extraction = new ProcessorInstructionPriority(1);
+    public static final ProcessorInstructionPriority Filtration = new ProcessorInstructionPriority(1);
+    public static final ProcessorInstructionPriority Grouping = new ProcessorInstructionPriority(2);
+    
+    private static final int retrievalBasePriority = Integer.MAX_VALUE;
+    
+    /**
+     * Creates a priority for the given <code>retrieverLevel</code>, where the level <code>0</code>
+     * represents the first retriever in the chain.
+     * 
+     * @param retrievalLevel The position of the retriever in its chain
+     * @return A priority for the given <code>retrieverLevel</code>
+     */
+    public static ProcessorInstructionPriority createRetrievalPriority(int retrievalLevel) {
+        return new ProcessorInstructionPriority(retrievalBasePriority - retrievalLevel);
+    }
+
+    private final int intValue;
+
+    private ProcessorInstructionPriority(int intValue) {
+        this.intValue = intValue;
+    }
+
+    public int asIntValue() {
+        return intValue;
+    }
+
+}

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

@@ -8,7 +8,8 @@ import java.util.concurrent.locks.ReentrantLock;
 import com.sap.sse.datamining.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
-import com.sap.sse.datamining.impl.components.ProcessorInstruction;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
+import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
 
 public abstract class AbstractParallelStoringAggregationProcessor<InputType, AggregatedType> 
                       extends AbstractSimpleParallelProcessor<InputType, AggregatedType> {
@@ -27,8 +28,8 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
     }
 
     @Override
-    protected ProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
-        return new ProcessorInstruction<AggregatedType>(this) {
+    protected AbstractProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
+        return new AbstractProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
             @Override
             public AggregatedType computeResult() {
                 storeLock.lock();