Finished the integration test to check the behavior when a query with heavy load instructions is aborted

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

@@ -1,14 +1,20 @@
 package com.sap.sse.datamining.impl;
 
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.lang.reflect.Method;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -17,6 +23,7 @@ import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -37,25 +44,63 @@ import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.shared.data.QueryResultState;
 import com.sap.sse.datamining.shared.impl.GenericGroupKey;
 import com.sap.sse.datamining.test.util.components.StatefulBlockingInstruction;
+import com.sap.sse.datamining.test.util.components.StatefulProcessorInstruction;
 
+/**
+ * Integration test aborting a query with simulated heavy load instructions. Uses a highly customized processor chain
+ * (based on the processors/instructions used in production) that allows detailed introspection and logging of the
+ * execution process. The test is used to verify the behavior of processors, if such a heavy load query is aborted
+ * during the process. For example that no instructions are scheduled after the query was aborted and the behavior of
+ * already scheduled instructions that may or may not have been started.<br>
+ * <br>
+ * The test is configurable regarding the number of threads for the executor, the number of data elements/heavy load
+ * instructions, the duration of heavy load instructions and the time to wait before the query is aborted. Note that
+ * the test can fail for some configurations, due to race conditions that would be very hard to check for.<br>
+ * <br>
+ * The test optionally records the process in order of the execution (using a concurrent message queue), which is
+ * printed to the console before assertions are performed. This isn't used by the assertions, but can be used to get
+ * a better understanding of what happens during the execution.<br>
+ * <br>
+ * See {@link #initialize()} for more details about the processor chain.
+ * 
+ * @author Lennart Hensler
+ */
 public class TestAbortingHeavyLoadQuery {
     
-    private static final int ExecutorPoolSize = Math.min(3, Runtime.getRuntime().availableProcessors() - 2);
+    // Test Configuration ----------------------------------------------------------------------------------------
+    /** 
+     * Number of threads in the executor. Determines the number of elements (and thus heavy load instructions)
+     * retrieved for each group. This means that each group should have a runtime of {@value #HeavyLoadInstructionDuration}ms,
+     * since the single instructions are executed concurrently.
+     */
+    private static final int ExecutorPoolSize = Math.max(3, Runtime.getRuntime().availableProcessors());
+    /** The number of groups contained in the initial data source. */
     private static final int DataSourceSize = 2000;
     private static final String GroupKeyPrefix = "G";
+    /** The time a heavy load instruction blocks the executing thread using {@link Thread#sleep(long)}. */
     private static final long HeavyLoadInstructionDuration = 500;
-    private static final long AbortQueryDelay = (long) (HeavyLoadInstructionDuration * 1.1);
+    /** The number of milliseconds to wait before {@link Query#abort()} is called. */
+    private static final long AbortQueryDelay = (long) (HeavyLoadInstructionDuration * 5.45);
+    //------------------------------------------------------------------------------------------------------------
     
+    // Execution Recording Configuration -------------------------------------------------------------------------
+    /** Enables concurrent logging of the query execution and prints the current record before assertions. */
     private static final boolean RecordExecution = true;
+    private static final SimpleDateFormat DateFormatter = new SimpleDateFormat("HH:mm:ss.SSS");
     private ConcurrentLinkedQueue<String> executionRecord;
+    //------------------------------------------------------------------------------------------------------------
     
+    // Test State - Initialized in initialize() ------------------------------------------------------------------
     private ExecutorService executor;
     private Query<HashSet<Integer>> query;
     private Collection<Processor<?, ?>> processors;
+    private Set<StatefulProcessorInstruction<?>> unfinishedInstructions;
+    //------------------------------------------------------------------------------------------------------------
     
     @Test
     public void testAbortingHeavyLoadQuery() throws InterruptedException, ExecutionException {
-        RunnableFuture<QueryResult<HashSet<Integer>>> queryTask = new FutureTask<>(() -> { 
+        // Executing query in separate thread
+        RunnableFuture<QueryResult<HashSet<Integer>>> queryTask = new FutureTask<>(() -> {
             logExecution("Starting query execution");
             long start = System.currentTimeMillis();
             QueryResult<HashSet<Integer>> result = query.run();
@@ -69,46 +114,133 @@ public class TestAbortingHeavyLoadQuery {
             Thread.sleep(10);
         } while (query.getState() != QueryState.RUNNING);
         
+        // Aborting query and waiting for completion
         Thread.sleep(AbortQueryDelay);
         logExecution("Aborting query");
         query.abort();
         QueryResult<HashSet<Integer>> result = queryTask.get();
+        
+        // The query execution returned, so all processors are aborted (checked below).
+        // The number of unfinished instructions mustn't change after this point
+        int unfinishedInstructionsCount = unfinishedInstructions.size();
+        boolean unfinishedInstructionWasFinished = false;
+        Set<StatefulProcessorInstruction<?>> runningInstructions = new HashSet<>();
+        Set<StatefulProcessorInstruction<?>> notStartedInstructions = new HashSet<>();
+        for (StatefulProcessorInstruction<?> instruction : unfinishedInstructions) {
+            if (instruction.runWasCalled()) {
+                runningInstructions.add(instruction);
+            } else {
+                notStartedInstructions.add(instruction);
+            }
+            unfinishedInstructionWasFinished |= instruction.computeResultWasFinished();
+        }
+        logExecution(unfinishedInstructionsCount + " unfinished instructions left - " +
+                     runningInstructions.size() + " running, " + notStartedInstructions.size() + " not started");
+        printExecutionRecord();
+        assertThat("Number of unfinished instructions changed", runningInstructions.size() + notStartedInstructions.size(), is(unfinishedInstructionsCount));
+        assertFalse("Unfinished instructions expected, but at least one was completed", unfinishedInstructionWasFinished);
+        
+        // Checking query, result and processor chain state
+        assertThat(query.getState(), is(QueryState.ABORTED));
         assertThat(result.getState(), is(QueryResultState.ABORTED));
         assertTrue("The result is not empty", result.isEmpty());
         for (Processor<?, ?> processor : processors) {
             assertTrue("Processor wasn't aborted", processor.isAborted());
         }
 
-        // TODO Ensure that no instructions are scheduled after the query has been aborted 
-        // TODO More detailed assessment of the leftover instructions
-        
+        // Execution of unfinished instructions
         executor.shutdown();
         boolean terminated = executor.awaitTermination((long) (HeavyLoadInstructionDuration * 1.5), TimeUnit.MILLISECONDS);
-        
-        if (RecordExecution) {
-            for (String string : executionRecord) {
-                System.out.println(string);
-            } 
-        }
+        printExecutionRecord();
         assertTrue("The executor didn't terminate in the given time", terminated);
+        
+        // Verify that the collection of unfinished instructions didn't change
+        assertThat("Number of unfinished instructions changed", unfinishedInstructions.size(), is(unfinishedInstructionsCount));
+        for (StatefulProcessorInstruction<?> instruction : unfinishedInstructions) {
+            boolean previouslyContained = runningInstructions.contains(instruction) || notStartedInstructions.contains(instruction);
+            assertTrue("A new instruction has been scheduled after aborting the query", previouslyContained);
+        }
+        for (StatefulProcessorInstruction<?> instruction : runningInstructions) {
+            assertTrue("A previously running instruction was removed from unfinished instructions", unfinishedInstructions.contains(instruction));
+        }
+        for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
+            assertTrue("A previously unstarted instruction was removed from unfinished instructions", unfinishedInstructions.contains(instruction));
+        }
+
+        // Checking state of unfinished instructions
+        for (StatefulProcessorInstruction<?> instruction : runningInstructions) {
+            assertTrue("computeResult() of a running unfinished instruction wasn't called", instruction.computeResultWasCalled());
+            assertTrue("computeResult() of a running unfinished instruction didn't finish", instruction.computeResultWasFinished());
+        }
+        for (StatefulProcessorInstruction<?> instruction : notStartedInstructions) {
+            assertTrue("run() of an unstarted unfinished instruction wasn't called", instruction.runWasCalled());
+            
+            assertFalse("computeResult() of an unstarted unfinished instruction was called", instruction.computeResultWasCalled());
+            assertFalse("computeResult() of an unstarted unfinished instruction was finished", instruction.computeResultWasFinished());
+        }
+    }
+
+    private void printExecutionRecord() {
+        if (RecordExecution) {
+            Collection<String> snapshot = new ArrayList<>(executionRecord);
+            executionRecord.clear();
+            for (String string : snapshot) {
+                System.out.println(string);
+            }
+        }
     }
 
     private void logExecution(String message) {
         if (RecordExecution) {
-            executionRecord.add(Thread.currentThread().getName() + ": " + message);
+            String timeString = DateFormatter.format(new Date());
+            executionRecord.add(timeString + " - " + Thread.currentThread().getName() + ": " + message);
         }
     }
-    
+
+    /**
+     * Initializes the {@link ExecutorService}, the execution record, helper collections (e.g. the processors in the
+     * chain or the unfinished instructions) and most importantly the processor chain for the query, which works as
+     * follows:
+     * <ol>
+     *   <li>
+     *     The initial data source is a collection of strings from {@value #GroupKeyPrefix}<code>n</code> to
+     *     {@value #GroupKeyPrefix}<code>n-1</code>, denoting a group. <code>n</code> is specified in {@link #DataSourceSize}.
+     *   </li>
+     *   <li>
+     *     <code>x</code> (specified by {@link #ExecutorPoolSize}) {@link Element elements} are retrieved for each
+     *     group, where the elements name is set to the received string and the elements value is the current value
+     *     of <code>x</code>.
+     *   </li>
+     *   <li>
+     *     A heavy load instruction for each element is scheduled, which blocks the running thread for
+     *     {@value #HeavyLoadInstructionDuration}ms.
+     *   </li>
+     *   <li>
+     *     Each element is grouped by its name and its value is used as value for the {@link GroupedDataEntry}.
+     *   </li>
+     *   <li>The grouped data is collected as set.</li>
+     * </ol>
+     * This results in <code>n * x</code> data elements (and heavy load instructions), with <code>x</code> instructions
+     * executed concurrently.<br>
+     * <br>
+     * Each processor creates {@link StatefulProcessorInstruction} or uses {@link StatefulInstructionWrapper}, which allows
+     * to verify if <code>instruction.run()</code> was called, its computation started and its computation finished.<br>
+     * <br>
+     * A concurrent set ({@link ConcurrentHashMap#newKeySet()}) is used to track the unfinished instructions. An instruction
+     * is added to the set upon its construction and is removed when the instruction finished callback method is called.
+     */
     @Before
     @SuppressWarnings("unchecked")
     public void initialize() {
-        executor = new DataMiningExecutorService(ExecutorPoolSize);
-//        executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>());
-//        executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
         if (RecordExecution) {
             executionRecord = new ConcurrentLinkedQueue<>();
         }
+        
+        executor = new DataMiningExecutorService(ExecutorPoolSize);
+//        executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>());
+//        executor = new ThreadPoolExecutor(ExecutorPoolSize, ExecutorPoolSize, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
         processors = new ArrayList<>();
+        unfinishedInstructions = ConcurrentHashMap.newKeySet();
         
         Class<Iterable<String>> dataSourceType = (Class<Iterable<String>>)(Class<?>) Iterable.class;
         Class<GroupedDataEntry<Integer>> groupedType = (Class<GroupedDataEntry<Integer>>)(Class<?>) GroupedDataEntry.class;
@@ -118,29 +250,51 @@ public class TestAbortingHeavyLoadQuery {
         for (int i = 0; i < DataSourceSize; i++) {
             dataSource.add(GroupKeyPrefix + i);
         }
-        
         query = new ProcessorQuery<HashSet<Integer>, Iterable<String>>(dataSource, resultType) {
             @Override
             protected Processor<Iterable<String>, ?> createChainAndReturnFirstProcessor(Processor<Map<GroupKey, HashSet<Integer>>, Void> resultReceiver) {
                 Processor<GroupedDataEntry<Integer>, Map<GroupKey, HashSet<Integer>>> aggregator = new ParallelGroupedDataCollectingAsSetProcessor<Integer>(executor, Collections.singleton(resultReceiver)) {
+                    @Override
+                    protected ProcessorInstruction<Map<GroupKey, HashSet<Integer>>> createInstruction(GroupedDataEntry<Integer> element) {
+                        AbstractProcessorInstruction<Map<GroupKey, HashSet<Integer>>> instruction = (AbstractProcessorInstruction<Map<GroupKey, HashSet<Integer>>>) super.createInstruction(element);
+                        StatefulProcessorInstruction<Map<GroupKey, HashSet<Integer>>> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
+                        unfinishedInstructions.add(statefulInstruction);
+                        return statefulInstruction;
+                    }
+                    
                     @Override
                     protected void storeElement(GroupedDataEntry<Integer> element) {
                         logExecution("Storing " + element);
                         super.storeElement(element);
                     }
+                    
+                    @Override
+                    public void afterInstructionFinished(ProcessorInstruction<Map<GroupKey, HashSet<Integer>>> instruction) {
+                        super.afterInstructionFinished(instruction);
+                        if(canProcessElements()) unfinishedInstructions.remove(instruction);
+                    }
                 };
                 
                 Processor<Element, GroupedDataEntry<Integer>> grouper = new AbstractParallelProcessor<Element, GroupedDataEntry<Integer>>(Element.class, groupedType, executor, Collections.singleton(aggregator)) {
                     @Override
                     protected ProcessorInstruction<GroupedDataEntry<Integer>> createInstruction(Element element) {
-                        return new AbstractProcessorInstruction<GroupedDataEntry<Integer>>(this, ProcessorInstructionPriority.Grouping) {
+                        StatefulProcessorInstruction<GroupedDataEntry<Integer>> instruction = new StatefulProcessorInstruction<GroupedDataEntry<Integer>>(this, ProcessorInstructionPriority.Grouping) {
                             @Override
-                            protected GroupedDataEntry<Integer> computeResult() throws Exception {
+                            protected GroupedDataEntry<Integer> internalComputeResult() throws Exception {
                                 logExecution("Grouping " + element);
                                 return new GroupedDataEntry<>(new GenericGroupKey<>(element.getName()), element.getValue());
                             }
                         };
+                        unfinishedInstructions.add(instruction);
+                        return instruction;
                     }
+                    
+                    @Override
+                    public void afterInstructionFinished(ProcessorInstruction<GroupedDataEntry<Integer>> instruction) {
+                        super.afterInstructionFinished(instruction);
+                        if(canProcessElements()) unfinishedInstructions.remove(instruction);
+                    }
+                    
                     @Override
                     protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { }
                 };
@@ -148,20 +302,35 @@ public class TestAbortingHeavyLoadQuery {
                 Processor<Element, Element> heavyLoadProcessor = new AbstractParallelProcessor<Element, Element>(Element.class, Element.class, executor, Collections.singleton(grouper)) {
                     @Override
                     protected ProcessorInstruction<Element> createInstruction(Element element) {
-                        ProcessorInstruction<Element> instruction = new RecordingBlockingInstruction(this,
+                        StatefulProcessorInstruction<Element> instruction = new HeavyLoadInstruction(this,
                                 ProcessorInstructionPriority.Extraction, HeavyLoadInstructionDuration, element,
                                 TestAbortingHeavyLoadQuery.this::logExecution);
+                        unfinishedInstructions.add(instruction);
                         return instruction;
                     }
+                    
+                    @Override
+                    public void afterInstructionFinished(ProcessorInstruction<Element> instruction) {
+                        super.afterInstructionFinished(instruction);
+                        if(canProcessElements()) unfinishedInstructions.remove(instruction);
+                    }
+                    
                     @Override
                     protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { }
-                    
                 };
                 
                 Processor<String, Element> retriever1 = new AbstractRetrievalProcessor<String, Element>(String.class, Element.class, executor, Collections.singleton(heavyLoadProcessor), 1) {
+                    @Override
+                    protected ProcessorInstruction<Element> createInstruction(String element) {
+                        AbstractProcessorInstruction<Element> instruction = (AbstractProcessorInstruction<Element>) super.createInstruction(element);
+                        StatefulProcessorInstruction<Element> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
+                        unfinishedInstructions.add(statefulInstruction);
+                        return statefulInstruction;
+                    }
+                    
                     @Override
                     protected Iterable<Element> retrieveData(String element) {
-                        logExecution("Retrieving " + ExecutorPoolSize + " elements from " + element);
+                        logExecution("Retrieving " + ExecutorPoolSize + " elements for " + element);
                         Collection<Element> data = new ArrayList<>();
                         for (int i = 0; i < ExecutorPoolSize; i++) {
                             data.add(new Element(element, i));
@@ -169,13 +338,44 @@ public class TestAbortingHeavyLoadQuery {
                         return data;
                     }
                     
+                    @Override
+                    public void instructionSucceeded(Element result) {
+                        logExecution("Group retrieval finished");
+                        super.instructionSucceeded(result);
+                    }
+                    
+                    @Override
+                    public void afterInstructionFinished(ProcessorInstruction<Element> instruction) {
+                        super.afterInstructionFinished(instruction);
+                        if(canProcessElements()) unfinishedInstructions.remove(instruction);
+                    }
                 };
                 Processor<Iterable<String>, String> retriever0 = new AbstractRetrievalProcessor<Iterable<String>, String>(dataSourceType, String.class, executor, Collections.singleton(retriever1), 0) {
+                    @Override
+                    protected ProcessorInstruction<String> createInstruction(Iterable<String> element) {
+                        AbstractProcessorInstruction<String> instruction = (AbstractProcessorInstruction<String>) super.createInstruction(element);
+                        StatefulProcessorInstruction<String> statefulInstruction = new StatefulInstructionWrapper<>(instruction);
+                        unfinishedInstructions.add(statefulInstruction);
+                        return statefulInstruction;
+                    }
+                    
                     @Override
                     protected Iterable<String> retrieveData(Iterable<String> element) {
                         logExecution("Retrieving data from data source");
                         return element;
                     }
+                    
+                    @Override
+                    public void instructionSucceeded(String result) {
+                        logExecution("Data source retrieval finished");
+                        super.instructionSucceeded(result);
+                    }
+                    
+                    @Override
+                    public void afterInstructionFinished(ProcessorInstruction<String> instruction) {
+                        super.afterInstructionFinished(instruction);
+                        if(canProcessElements()) unfinishedInstructions.remove(instruction);
+                    }
                 };
 
                 processors.add(retriever0);
@@ -189,11 +389,11 @@ public class TestAbortingHeavyLoadQuery {
         };
     }
     
-    private static class RecordingBlockingInstruction extends StatefulBlockingInstruction<Element> {
+    private static class HeavyLoadInstruction extends StatefulBlockingInstruction<Element> {
 
         private final Consumer<String> recorder;
 
-        public RecordingBlockingInstruction(ProcessorInstructionHandler<Element> handler,
+        public HeavyLoadInstruction(ProcessorInstructionHandler<Element> handler,
                 ProcessorInstructionPriority priority, long blockDuration, Element result, Consumer<String> recorder) {
             super(handler, priority, blockDuration, result);
             this.recorder = recorder;
@@ -217,6 +417,37 @@ public class TestAbortingHeavyLoadQuery {
         
     }
     
+    private static class StatefulInstructionWrapper<ResultType> extends StatefulProcessorInstruction<ResultType> {
+        
+        private static Method computeResult;
+        
+        private final AbstractProcessorInstruction<ResultType> instruction;
+
+        public StatefulInstructionWrapper(AbstractProcessorInstruction<ResultType> instruction) {
+            super(instruction.getHandler(), instruction.getPriority());
+            this.instruction = instruction;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        protected ResultType internalComputeResult() throws Exception {
+            if (computeResult == null) {
+                computeResult = AbstractProcessorInstruction.class.getDeclaredMethod("computeResult");
+                computeResult.setAccessible(true);
+            }
+            return (ResultType) computeResult.invoke(instruction);
+        }
+
+    }
+    
+    @After
+    public void resetComputeResultAccessibility() throws SecurityException, NoSuchMethodException {
+        AbstractProcessorInstruction.class.getDeclaredMethod("computeResult").setAccessible(false);
+    }
+    
+    /**
+     * Simple data type consisting of a name and value.
+     */
     private static class Element {
         
         private final String name;

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

@@ -1,17 +1,12 @@
 package com.sap.sse.datamining.test.util.components;
 
 import com.sap.sse.datamining.components.ProcessorInstructionHandler;
-import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
 
-public class StatefulBlockingInstruction<ResultType> extends AbstractProcessorInstruction<ResultType> {
+public class StatefulBlockingInstruction<ResultType> extends StatefulProcessorInstruction<ResultType> {
     
     protected final long blockDuration;
     protected final ResultType result;
-    
-    private boolean runWasCalled;
-    private boolean computeResultWasCalled;
-    private boolean computeResultWasFinished;
 
     public StatefulBlockingInstruction(ProcessorInstructionHandler<ResultType> handler, long blockDuration) {
         this(handler, 0, blockDuration, null);
@@ -26,22 +21,14 @@ public class StatefulBlockingInstruction<ResultType> extends AbstractProcessorIn
         this.blockDuration = blockDuration;
         this.result = result;
     }
-    
-    @Override
-    public void run() {
-        runWasCalled = true;
-        super.run();
-    }
 
     @Override
-    protected ResultType computeResult() throws Exception {
-        computeResultWasCalled = true;
+    protected ResultType internalComputeResult() throws Exception {
         actionBeforeBlock();
         if (blockDuration > 0) {
             Thread.sleep(blockDuration);
         }
         actionAfterBlock();
-        computeResultWasFinished = true;
         return result;
     }
     
@@ -51,17 +38,5 @@ public class StatefulBlockingInstruction<ResultType> extends AbstractProcessorIn
     public long getBlockDuration() {
         return blockDuration;
     }
-
-    public boolean runWasCalled() {
-        return runWasCalled;
-    }
-
-    public boolean computeResultWasCalled() {
-        return computeResultWasCalled;
-    }
-
-    public boolean computeResultWasFinished() {
-        return computeResultWasFinished;
-    }
     
 }

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

@@ -0,0 +1,54 @@
+package com.sap.sse.datamining.test.util.components;
+
+import com.sap.sse.datamining.components.ProcessorInstructionHandler;
+import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
+import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
+
+public abstract class StatefulProcessorInstruction<ResultType> extends AbstractProcessorInstruction<ResultType> {
+    
+    private boolean runWasCalled;
+    private boolean computeResultWasCalled;
+    private boolean computeResultWasFinished;
+
+    public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> processor, int priority) {
+        super(processor, priority);
+    }
+
+    public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> handler,
+            ProcessorInstructionPriority priority) {
+        super(handler, priority);
+    }
+
+    public StatefulProcessorInstruction(ProcessorInstructionHandler<ResultType> handler) {
+        super(handler);
+    }
+    
+    @Override
+    public void run() {
+        runWasCalled = true;
+        super.run();
+    }
+
+    @Override
+    protected ResultType computeResult() throws Exception {
+        computeResultWasCalled = true;
+        ResultType result = internalComputeResult();
+        computeResultWasFinished = true;
+        return result;
+    }
+
+    protected abstract ResultType internalComputeResult() throws Exception;
+
+    public boolean runWasCalled() {
+        return runWasCalled;
+    }
+
+    public boolean computeResultWasCalled() {
+        return computeResultWasCalled;
+    }
+
+    public boolean computeResultWasFinished() {
+        return computeResultWasFinished;
+    }
+
+}

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

@@ -7,6 +7,6 @@ public interface ProcessorInstructionHandler<ResultType> {
     void instructionSucceeded(ResultType result);
     void instructionFailed(Exception e);
     
-    void afterInstructionFinished();
+    void afterInstructionFinished(ProcessorInstruction<ResultType> instruction);
 
 }

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

@@ -59,18 +59,21 @@ public abstract class AbstractParallelProcessor<InputType, ResultType> extends A
     private boolean isInstructionValid(ProcessorInstruction<ResultType> instruction) {
         return instruction != null;
     }
-    
+
+    @Override
     public void instructionSucceeded(ResultType result) {
         forwardResultToReceivers(result);
     }
-    
+
+    @Override
     public void instructionFailed(Exception e) {
         if (!isAborted() || !(e instanceof InterruptedException)) {
             onFailure(e);
         }
     }
-    
-    public void afterInstructionFinished() {
+
+    @Override
+    public void afterInstructionFinished(ProcessorInstruction<ResultType> instruction) {
         unfinishedInstructionsCounter.getAndDecrement();
     }
     

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

@@ -59,12 +59,16 @@ public abstract class AbstractProcessorInstruction<ResultType> implements Proces
         } catch (Exception e) {
             handler.instructionFailed(e);
         } finally {
-            handler.afterInstructionFinished();
+            handler.afterInstructionFinished(this);
         }
     }
 
     protected abstract ResultType computeResult() throws Exception;
     
+    public ProcessorInstructionHandler<ResultType> getHandler() {
+        return handler;
+    }
+    
     @Override
     public int getPriority() {
         return priority;

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

@@ -5,6 +5,7 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.AdditionalResultDataBuilder;
 import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.components.ProcessorInstruction;
 import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
 import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
@@ -26,7 +27,7 @@ public abstract class AbstractParallelAggregationProcessor<InputType, Aggregated
     }
 
     @Override
-    protected AbstractProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
+    protected ProcessorInstruction<AggregatedType> createInstruction(final InputType element) {
         if (needsSynchronization()) {
             return new AbstractProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
                 @Override