diff --git a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java
index eb0ad98f82d..6eb6d780dda 100644
--- a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java
+++ b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/GPSFixRetrievalProcessor.java
@@ -1,6 +1,7 @@
 package com.sap.sailing.datamining.impl.components;
 
 import java.util.Collection;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 
 import com.sap.sailing.datamining.data.GPSFixWithContext;
@@ -17,7 +18,7 @@ public class GPSFixRetrievalProcessor extends
     }
 
     @Override
-    protected Runnable createInstruction(LeaderboardGroup partialElement) {
+    protected Callable<GPSFixWithContext> createInstruction(LeaderboardGroup partialElement) {
         // TODO Auto-generated method stub
         return null;
     }
diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractDirectForwardingProcessingInstruction.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractDirectForwardingProcessingInstruction.java
deleted file mode 100644
index 0388c50fa27..00000000000
--- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestAbstractDirectForwardingProcessingInstruction.java
+++ /dev/null
@@ -1,47 +0,0 @@
-package com.sap.sse.datamining.components;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
-
-import java.util.ArrayList;
-import java.util.Collection;
-
-import org.junit.Test;
-
-import com.sap.sse.datamining.impl.components.AbstractDirectForwardProcessingInstruction;
-
-public class TestAbstractDirectForwardingProcessingInstruction {
-
-    private int calls = 0;
-
-    @Test
-    public void testThatTheResultIsDirectlyForwarded() {
-        Processor<Integer> processor = new Processor<Integer>() {
-            @Override
-            public void onElement(Integer element) {
-                calls++;
-            }
-
-            @Override
-            public void finish() throws InterruptedException { }
-        };
-        ProcessingInstruction<Integer, Integer> instruction = createInstruction(processor);
-
-        instruction.run();
-        assertThat(calls, is(1));
-    }
-
-    private ProcessingInstruction<Integer, Integer> createInstruction(Processor<Integer> resultReceiver) {
-        Integer input = 10;
-        Collection<Processor<Integer>> resultReceivers = new ArrayList<>();
-        resultReceivers.add(resultReceiver);
-        ProcessingInstruction<Integer, Integer> instruction = new AbstractDirectForwardProcessingInstruction<Integer, Integer>(input, resultReceivers) {
-            @Override
-            protected Integer doWork() {
-                return 0;
-            }
-        };
-        return instruction;
-    }
-
-}
diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/util/InputForwardingInstruction.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/util/InputForwardingInstruction.java
deleted file mode 100644
index 43d8cefcec5..00000000000
--- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/util/InputForwardingInstruction.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package com.sap.sse.datamining.test.components.util;
-
-import java.util.Collection;
-
-import com.sap.sse.datamining.components.Processor;
-import com.sap.sse.datamining.impl.components.AbstractDirectForwardProcessingInstruction;
-
-public class InputForwardingInstruction<T> extends AbstractDirectForwardProcessingInstruction<T, T> {
-
-    public InputForwardingInstruction(T input, Collection<Processor<T>> resultReceivers) {
-        super(input, resultReceivers);
-    }
-
-    @Override
-    protected T doWork() {
-        return super.getInput();
-    }
-
-}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ParallelFilteringProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ParallelFilteringProcessor.java
deleted file mode 100644
index be517f4e867..00000000000
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ParallelFilteringProcessor.java
+++ /dev/null
@@ -1,29 +0,0 @@
-package com.sap.sse.datamining.components;
-
-import java.util.Collection;
-import java.util.concurrent.Executor;
-
-import com.sap.sse.datamining.impl.components.AbstractDirectForwardProcessingInstruction;
-import com.sap.sse.datamining.impl.components.AbstractSimpleParallelProcessor;
-
-public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParallelProcessor<InputType, InputType> {
-
-    private final FilterCriteria<InputType> filterCriteria;
-
-    public ParallelFilteringProcessor(Executor executor, Collection<Processor<InputType>> resultReceivers, FilterCriteria<InputType> filterCriteria) {
-        super(executor, resultReceivers);
-        this.filterCriteria = filterCriteria;
-    }
-
-    @Override
-    protected Runnable createInstruction(InputType element) {
-        return new AbstractDirectForwardProcessingInstruction<InputType, InputType>(element, getResultReceivers()) {
-            @Override
-            protected InputType doWork() {
-                InputType input = super.getInput();
-                return filterCriteria.matches(input) ? input : super.getInvalidResult();
-            }
-        };
-    }
-
-}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java
deleted file mode 100644
index b708aa319c2..00000000000
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package com.sap.sse.datamining.components;
-
-
-public interface ProcessingInstruction<InputType, ResultType> extends Runnable {
-
-}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java
deleted file mode 100644
index 08f576a39af..00000000000
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java
+++ /dev/null
@@ -1,54 +0,0 @@
-package com.sap.sse.datamining.impl.components;
-
-import java.util.Collection;
-
-import com.sap.sse.datamining.components.ProcessingInstruction;
-import com.sap.sse.datamining.components.Processor;
-
-/**
- * A processing construction, that processes the input and directly forwards it to the result receivers.
- */
-public abstract class AbstractDirectForwardProcessingInstruction<InputType, ResultType>
-                      implements ProcessingInstruction<InputType, ResultType> {
-
-    private final InputType input;
-    private ResultType result;
-    private final Collection<Processor<ResultType>> resultReceivers;
-
-    public AbstractDirectForwardProcessingInstruction(InputType input, Collection<Processor<ResultType>> resultReceivers) {
-        this.input = input;
-        this.resultReceivers = resultReceivers;
-    }
-
-    @Override
-    public void run() {
-        result = doWork();
-        if (resultIsValid()) {
-            forwardResult();
-        }
-    }
-
-    private boolean resultIsValid() {
-        return result != null;
-    }
-    
-    /**
-     * @return A result, that won't be forwarded to the result receivers.
-     */
-    protected ResultType getInvalidResult() {
-        return null;
-    }
-
-    private void forwardResult() {
-        for (Processor<ResultType> resultReceiver : resultReceivers) {
-            resultReceiver.onElement(result);
-        }
-    }
-
-    protected abstract ResultType doWork();
-    
-    protected InputType getInput() {
-        return input;
-    }
-
-}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java
index c56e734b209..5dea0d4ba3e 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractPartitioningParallelProcessor.java
@@ -3,7 +3,11 @@ package com.sap.sse.datamining.impl.components;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -28,13 +32,22 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
     @Override
     public void onElement(InputType element) {
         for (WorkingType partialElement : partitionElement(element)) {
-            final Runnable instruction = createInstruction(partialElement);
-            if (instructionIsValid(instruction)) {
+            final RunnableFuture<ResultType> instruction = new FutureTask<>(createInstruction(partialElement));
+            if (isInstructionValid(instruction)) {
                 Runnable instructionWrapper = new Runnable() {
                     @Override
                     public void run() {
                         instruction.run();
-                        AbstractPartitioningParallelProcessor.this.unfinishedInstructionsCounter.decrement();
+                        try {
+                            ResultType result = instruction.get();
+                            if (isResultValid(result)) {
+                                forwardResultToReceivers(result);
+                            }
+                        } catch (InterruptedException | ExecutionException e) {
+                            LOGGER.log(Level.FINEST, "Error getting the result from the instruction: ", e);
+                        } finally {
+                            AbstractPartitioningParallelProcessor.this.unfinishedInstructionsCounter.decrement();
+                        }
                     }
                 };
                 unfinishedInstructionsCounter.increment();
@@ -43,18 +56,35 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
         }
     }
 
-    protected abstract Runnable createInstruction(WorkingType partialElement);
-
-    protected abstract Iterable<WorkingType> partitionElement(InputType element);
-
-    private boolean instructionIsValid(Runnable instruction) {
+    protected boolean isInstructionValid(Runnable instruction) {
         return instruction != null;
     }
 
+    protected boolean isResultValid(ResultType result) {
+        return result != null;
+    }
+    
+    /**
+     * @return An invalid result, that won't be forwarded to the result receivers.
+     */
+    protected ResultType createInvalidResult() {
+        return null;
+    }
+    
+    private void forwardResultToReceivers(ResultType result) {
+        for (Processor<ResultType> resultReceiver : resultReceivers) {
+            resultReceiver.onElement(result);
+        }
+    }
+
     protected Set<Processor<ResultType>> getResultReceivers() {
         return resultReceivers;
     }
 
+    protected abstract Callable<ResultType> createInstruction(final WorkingType partialElement);
+
+    protected abstract Iterable<WorkingType> partitionElement(InputType element);
+
     @Override
     public void finish() throws InterruptedException {
         while (areUnfinishedInstructionsLeft()) {
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleParallelProcessor.java
index 0d771d63195..37f22e39f49 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleParallelProcessor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractSimpleParallelProcessor.java
@@ -2,6 +2,7 @@ package com.sap.sse.datamining.impl.components;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 
 import com.sap.sse.datamining.components.Processor;
@@ -21,6 +22,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 Runnable createInstruction(InputType element);
+    protected abstract Callable<ResultType> createInstruction(InputType element);
 
 }
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelFilteringProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelFilteringProcessor.java
new file mode 100644
index 00000000000..49bd24e87d5
--- /dev/null
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelFilteringProcessor.java
@@ -0,0 +1,30 @@
+package com.sap.sse.datamining.impl.components;
+
+import java.util.Collection;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+
+import com.sap.sse.datamining.components.FilterCriteria;
+import com.sap.sse.datamining.components.Processor;
+
+public class ParallelFilteringProcessor<InputType> extends AbstractSimpleParallelProcessor<InputType, InputType> {
+
+    private final FilterCriteria<InputType> filterCriteria;
+
+    public ParallelFilteringProcessor(Executor executor, Collection<Processor<InputType>> resultReceivers, FilterCriteria<InputType> filterCriteria) {
+        super(executor, resultReceivers);
+        this.filterCriteria = filterCriteria;
+    }
+
+    @Override
+    protected Callable<InputType> createInstruction(final InputType element) {
+        return new Callable<InputType>() {
+            @Override
+            public InputType call() throws Exception {
+                InputType input = element;
+                return filterCriteria.matches(input) ? input : ParallelFilteringProcessor.super.createInvalidResult();
+            }
+        };
+    }
+
+}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java
index 7c16515d155..021de42f300 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java
@@ -1,6 +1,7 @@
 package com.sap.sse.datamining.impl.components;
 
 import java.util.Collection;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import com.sap.sse.datamining.components.Processor;
@@ -19,12 +20,13 @@ public class ParallelGroupedElementsValueExtractionProcessor<ElementType, Functi
     }
 
     @Override
-    protected Runnable createInstruction(GroupedDataEntry<ElementType> element) {
-        return new AbstractDirectForwardProcessingInstruction<GroupedDataEntry<ElementType>, GroupedDataEntry<FunctionReturnType>>(element, getResultReceivers()) {
+    protected Callable<GroupedDataEntry<FunctionReturnType>> createInstruction(final GroupedDataEntry<ElementType> element) {
+        return new Callable<GroupedDataEntry<FunctionReturnType>>() {
             @Override
-            protected GroupedDataEntry<FunctionReturnType> doWork() {
-                FunctionReturnType value = extractionFunction.tryToInvoke(super.getInput().getDataEntry());
-                return value != null ? new GroupedDataEntry<FunctionReturnType>(super.getInput().getKey(), value) : super.getInvalidResult();
+            public GroupedDataEntry<FunctionReturnType> call() throws Exception {
+                FunctionReturnType value = extractionFunction.tryToInvoke(element.getDataEntry());
+                return value != null ? new GroupedDataEntry<FunctionReturnType>(element.getKey(), value) :
+                                       ParallelGroupedElementsValueExtractionProcessor.super.createInvalidResult();
             }
         };
     }
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelMultiDimensionalGroupingProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelMultiDimensionalGroupingProcessor.java
index 552caf17027..8e04401e71c 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelMultiDimensionalGroupingProcessor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelMultiDimensionalGroupingProcessor.java
@@ -2,6 +2,7 @@ package com.sap.sse.datamining.impl.components;
 
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 
 import com.sap.sse.datamining.components.Processor;
@@ -37,12 +38,11 @@ public class ParallelMultiDimensionalGroupingProcessor<DataType>
     }
 
     @Override
-    protected Runnable createInstruction(DataType element) {
-        return new AbstractDirectForwardProcessingInstruction<DataType, GroupedDataEntry<DataType>>(element, getResultReceivers()) {
+    protected Callable<GroupedDataEntry<DataType>> createInstruction(final DataType element) {
+        return new Callable<GroupedDataEntry<DataType>>() {
             @Override
-            protected GroupedDataEntry<DataType> doWork() {
-                DataType input = super.getInput();
-                return new GroupedDataEntry<DataType>(createCompoundKeyFor(input, dimensions.iterator()), input);
+            public GroupedDataEntry<DataType> call() throws Exception {
+                return new GroupedDataEntry<DataType>(createCompoundKeyFor(element, dimensions.iterator()), element);
             }
         };
     }