From ab381ac72a4cc43de6b91ef34a266c5ca70a7fbc Mon Sep 17 00:00:00 2001 From: Lennart Hensler Date: Mon, 30 Jul 2018 17:22:17 +0200 Subject: [PATCH] Finished the integration test to check the behavior when a query with heavy load instructions is aborted --- .../impl/TestAbortingHeavyLoadQuery.java | 281 ++++++++++++++++-- .../StatefulBlockingInstruction.java | 29 +- .../StatefulProcessorInstruction.java | 54 ++++ .../ProcessorInstructionHandler.java | 2 +- .../components/AbstractParallelProcessor.java | 11 +- .../AbstractProcessorInstruction.java | 6 +- .../AbstractParallelAggregationProcessor.java | 3 +- 7 files changed, 327 insertions(+), 59 deletions(-) create mode 100644 java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulProcessorInstruction.java diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestAbortingHeavyLoadQuery.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestAbortingHeavyLoadQuery.java index 27db549e027..2aa3e6ada0e 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestAbortingHeavyLoadQuery.java +++ b/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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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 executionRecord; + //------------------------------------------------------------------------------------------------------------ + // Test State - Initialized in initialize() ------------------------------------------------------------------ private ExecutorService executor; private Query> query; private Collection> processors; + private Set> unfinishedInstructions; + //------------------------------------------------------------------------------------------------------------ @Test public void testAbortingHeavyLoadQuery() throws InterruptedException, ExecutionException { - RunnableFuture>> queryTask = new FutureTask<>(() -> { + // Executing query in separate thread + RunnableFuture>> queryTask = new FutureTask<>(() -> { logExecution("Starting query execution"); long start = System.currentTimeMillis(); QueryResult> 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> 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> runningInstructions = new HashSet<>(); + Set> 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 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: + *
    + *
  1. + * The initial data source is a collection of strings from {@value #GroupKeyPrefix}n to + * {@value #GroupKeyPrefix}n-1, denoting a group. n is specified in {@link #DataSourceSize}. + *
    2. + *
  2. + * x (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 x. + *
    3. + *
  3. + * A heavy load instruction for each element is scheduled, which blocks the running thread for + * {@value #HeavyLoadInstructionDuration}ms. + *
    4. + *
  4. + * Each element is grouped by its name and its value is used as value for the {@link GroupedDataEntry}. + *
    5. + *
  5. The grouped data is collected as set.
    6. + *
+ * This results in n * x data elements (and heavy load instructions), with x instructions + * executed concurrently.
+ *
+ * Each processor creates {@link StatefulProcessorInstruction} or uses {@link StatefulInstructionWrapper}, which allows + * to verify if instruction.run() was called, its computation started and its computation finished.
+ *
+ * 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> dataSourceType = (Class>)(Class) Iterable.class; Class> groupedType = (Class>)(Class) GroupedDataEntry.class; @@ -118,29 +250,51 @@ public class TestAbortingHeavyLoadQuery { for (int i = 0; i < DataSourceSize; i++) { dataSource.add(GroupKeyPrefix + i); } - query = new ProcessorQuery, Iterable>(dataSource, resultType) { @Override protected Processor, ?> createChainAndReturnFirstProcessor(Processor>, Void> resultReceiver) { Processor, Map>> aggregator = new ParallelGroupedDataCollectingAsSetProcessor(executor, Collections.singleton(resultReceiver)) { + @Override + protected ProcessorInstruction>> createInstruction(GroupedDataEntry element) { + AbstractProcessorInstruction>> instruction = (AbstractProcessorInstruction>>) super.createInstruction(element); + StatefulProcessorInstruction>> statefulInstruction = new StatefulInstructionWrapper<>(instruction); + unfinishedInstructions.add(statefulInstruction); + return statefulInstruction; + } + @Override protected void storeElement(GroupedDataEntry element) { logExecution("Storing " + element); super.storeElement(element); } + + @Override + public void afterInstructionFinished(ProcessorInstruction>> instruction) { + super.afterInstructionFinished(instruction); + if(canProcessElements()) unfinishedInstructions.remove(instruction); + } }; Processor> grouper = new AbstractParallelProcessor>(Element.class, groupedType, executor, Collections.singleton(aggregator)) { @Override protected ProcessorInstruction> createInstruction(Element element) { - return new AbstractProcessorInstruction>(this, ProcessorInstructionPriority.Grouping) { + StatefulProcessorInstruction> instruction = new StatefulProcessorInstruction>(this, ProcessorInstructionPriority.Grouping) { @Override - protected GroupedDataEntry computeResult() throws Exception { + protected GroupedDataEntry 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> instruction) { + super.afterInstructionFinished(instruction); + if(canProcessElements()) unfinishedInstructions.remove(instruction); + } + @Override protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { } }; @@ -148,20 +302,35 @@ public class TestAbortingHeavyLoadQuery { Processor heavyLoadProcessor = new AbstractParallelProcessor(Element.class, Element.class, executor, Collections.singleton(grouper)) { @Override protected ProcessorInstruction createInstruction(Element element) { - ProcessorInstruction instruction = new RecordingBlockingInstruction(this, + StatefulProcessorInstruction instruction = new HeavyLoadInstruction(this, ProcessorInstructionPriority.Extraction, HeavyLoadInstructionDuration, element, TestAbortingHeavyLoadQuery.this::logExecution); + unfinishedInstructions.add(instruction); return instruction; } + + @Override + public void afterInstructionFinished(ProcessorInstruction instruction) { + super.afterInstructionFinished(instruction); + if(canProcessElements()) unfinishedInstructions.remove(instruction); + } + @Override protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) { } - }; Processor retriever1 = new AbstractRetrievalProcessor(String.class, Element.class, executor, Collections.singleton(heavyLoadProcessor), 1) { + @Override + protected ProcessorInstruction createInstruction(String element) { + AbstractProcessorInstruction instruction = (AbstractProcessorInstruction) super.createInstruction(element); + StatefulProcessorInstruction statefulInstruction = new StatefulInstructionWrapper<>(instruction); + unfinishedInstructions.add(statefulInstruction); + return statefulInstruction; + } + @Override protected Iterable retrieveData(String element) { - logExecution("Retrieving " + ExecutorPoolSize + " elements from " + element); + logExecution("Retrieving " + ExecutorPoolSize + " elements for " + element); Collection 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 instruction) { + super.afterInstructionFinished(instruction); + if(canProcessElements()) unfinishedInstructions.remove(instruction); + } }; Processor, String> retriever0 = new AbstractRetrievalProcessor, String>(dataSourceType, String.class, executor, Collections.singleton(retriever1), 0) { + @Override + protected ProcessorInstruction createInstruction(Iterable element) { + AbstractProcessorInstruction instruction = (AbstractProcessorInstruction) super.createInstruction(element); + StatefulProcessorInstruction statefulInstruction = new StatefulInstructionWrapper<>(instruction); + unfinishedInstructions.add(statefulInstruction); + return statefulInstruction; + } + @Override protected Iterable retrieveData(Iterable 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 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 { + private static class HeavyLoadInstruction extends StatefulBlockingInstruction { private final Consumer recorder; - public RecordingBlockingInstruction(ProcessorInstructionHandler handler, + public HeavyLoadInstruction(ProcessorInstructionHandler handler, ProcessorInstructionPriority priority, long blockDuration, Element result, Consumer recorder) { super(handler, priority, blockDuration, result); this.recorder = recorder; @@ -217,6 +417,37 @@ public class TestAbortingHeavyLoadQuery { } + private static class StatefulInstructionWrapper extends StatefulProcessorInstruction { + + private static Method computeResult; + + private final AbstractProcessorInstruction instruction; + + public StatefulInstructionWrapper(AbstractProcessorInstruction 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; diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulBlockingInstruction.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulBlockingInstruction.java index 8ad2210288e..0efe183c595 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulBlockingInstruction.java +++ b/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 extends AbstractProcessorInstruction { +public class StatefulBlockingInstruction extends StatefulProcessorInstruction { protected final long blockDuration; protected final ResultType result; - - private boolean runWasCalled; - private boolean computeResultWasCalled; - private boolean computeResultWasFinished; public StatefulBlockingInstruction(ProcessorInstructionHandler handler, long blockDuration) { this(handler, 0, blockDuration, null); @@ -26,22 +21,14 @@ public class StatefulBlockingInstruction 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 extends AbstractProcessorIn public long getBlockDuration() { return blockDuration; } - - public boolean runWasCalled() { - return runWasCalled; - } - - public boolean computeResultWasCalled() { - return computeResultWasCalled; - } - - public boolean computeResultWasFinished() { - return computeResultWasFinished; - } } \ No newline at end of file diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulProcessorInstruction.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/StatefulProcessorInstruction.java new file mode 100644 index 00000000000..b34a376e1eb --- /dev/null +++ b/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 extends AbstractProcessorInstruction { + + private boolean runWasCalled; + private boolean computeResultWasCalled; + private boolean computeResultWasFinished; + + public StatefulProcessorInstruction(ProcessorInstructionHandler processor, int priority) { + super(processor, priority); + } + + public StatefulProcessorInstruction(ProcessorInstructionHandler handler, + ProcessorInstructionPriority priority) { + super(handler, priority); + } + + public StatefulProcessorInstruction(ProcessorInstructionHandler 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; + } + +} diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessorInstructionHandler.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessorInstructionHandler.java index e01461324d7..b5da6b06261 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessorInstructionHandler.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessorInstructionHandler.java @@ -7,6 +7,6 @@ public interface ProcessorInstructionHandler { void instructionSucceeded(ResultType result); void instructionFailed(Exception e); - void afterInstructionFinished(); + void afterInstructionFinished(ProcessorInstruction instruction); } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java index caa9f427424..77e478588f3 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java @@ -59,18 +59,21 @@ public abstract class AbstractParallelProcessor extends A private boolean isInstructionValid(ProcessorInstruction 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 instruction) { unfinishedInstructionsCounter.getAndDecrement(); } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java index ff404891ac0..a785e6481de 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractProcessorInstruction.java @@ -59,12 +59,16 @@ public abstract class AbstractProcessorInstruction implements Proces } catch (Exception e) { handler.instructionFailed(e); } finally { - handler.afterInstructionFinished(); + handler.afterInstructionFinished(this); } } protected abstract ResultType computeResult() throws Exception; + public ProcessorInstructionHandler getHandler() { + return handler; + } + @Override public int getPriority() { return priority; diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java index 2506db2ffc1..eac75921afb 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java +++ b/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 createInstruction(final InputType element) { + protected ProcessorInstruction createInstruction(final InputType element) { if (needsSynchronization()) { return new AbstractProcessorInstruction(this, ProcessorInstructionPriority.Aggregation) { @Override