diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/AbstractPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/AbstractPartitionHandler.java new file mode 100644 index 000000000..13e900954 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/AbstractPartitionHandler.java @@ -0,0 +1,84 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.core.partition.support; + +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.StepExecutionSplitter; + +import java.util.Collection; +import java.util.Set; + +/** + * Base {@link PartitionHandler} implementation providing common base + * features. Subclasses are expected to implement only the + * {@link #doHandle(org.springframework.batch.core.StepExecution, java.util.Set)} + * method which returns with the result of the execution(s) or an exception if + * the step failed to process. + * + * @author Sebastien Gerard + * @author Dave Syer + */ +public abstract class AbstractPartitionHandler implements PartitionHandler { + + private int gridSize = 1; + + /** + * Executes the specified {@link StepExecution} instances and returns an updated + * view of them. Throws an {@link Exception} if anything goes wrong. + * + * @param masterStepExecution the whole partition execution + * @param partitionStepExecutions the {@link StepExecution} instances to execute + * @return an updated view of these completed {@link StepExecution} instances + * @throws Exception if anything goes wrong. This allows implementations to + * be liberal and rely on the caller to translate an exception into a step + * failure as necessary. + */ + protected abstract Set doHandle(StepExecution masterStepExecution, + Set partitionStepExecutions) throws Exception; + + + public Collection handle(final StepExecutionSplitter stepSplitter, + final StepExecution masterStepExecution) throws Exception { + final Set stepExecutions = stepSplitter.split(masterStepExecution, gridSize); + + return doHandle(masterStepExecution, stepExecutions); + } + + /** + * Returns the number of step executions. + * + * @return the number of step executions + */ + public int getGridSize() { + return gridSize; + } + + /** + * Passed to the {@link StepExecutionSplitter} in the + * {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing + * it how many {@link StepExecution} instances are required, ideally. The + * {@link StepExecutionSplitter} is allowed to ignore the grid size in the + * case of a restart, since the input data partitions must be preserved. + * + * @param gridSize the number of step executions that will be created + */ + public void setGridSize(int gridSize) { + this.gridSize = gridSize; + } + +} + diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java index 7731b934b..cb40e5f1c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java @@ -16,8 +16,6 @@ package org.springframework.batch.core.partition.support; -import java.util.ArrayList; -import java.util.Collection; import java.util.HashSet; import java.util.Set; import java.util.concurrent.Callable; @@ -29,9 +27,9 @@ import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.partition.PartitionHandler; -import org.springframework.batch.core.partition.StepExecutionSplitter; import org.springframework.batch.core.step.StepHolder; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Required; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.core.task.TaskRejectedException; @@ -42,13 +40,14 @@ import org.springframework.util.Assert; * partitioned {@link Step} locally in multiple threads. This can be an * effective approach for scaling batch steps that are IO intensive, like * directory and filesystem scanning and copying. - * + *

+ * By default, the thread pool is synchronous. + * + * @author Sebastien Gerard * @author Dave Syer * @since 2.0 */ -public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolder, InitializingBean { - - private int gridSize = 1; +public class TaskExecutorPartitionHandler extends AbstractPartitionHandler implements StepHolder, InitializingBean { private TaskExecutor taskExecutor = new SyncTaskExecutor(); @@ -57,19 +56,6 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde public void afterPropertiesSet() throws Exception { } - /** - * Passed to the {@link StepExecutionSplitter} in the - * {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing - * it how many {@link StepExecution} instances are required, ideally. The - * {@link StepExecutionSplitter} is allowed to ignore the grid size in the - * case of a restart, since the input data partitions must be preserved. - * - * @param gridSize the number of step executions that will be created - */ - public void setGridSize(int gridSize) { - this.gridSize = gridSize; - } - /** * Setter for the {@link TaskExecutor} that is used to farm out step * executions to multiple threads. @@ -87,6 +73,7 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde * * @param step the {@link Step} instance to use to execute business logic */ + @Required public void setStep(Step step) { this.step = step; } @@ -101,51 +88,55 @@ public class TaskExecutorPartitionHandler implements PartitionHandler, StepHolde return this.step; } - /** - * @see PartitionHandler#handle(StepExecutionSplitter, StepExecution) - */ - public Collection handle(StepExecutionSplitter stepExecutionSplitter, - StepExecution masterStepExecution) throws Exception { + @Override + protected Set doHandle(StepExecution masterStepExecution, + Set partitionStepExecutions) throws Exception { + Assert.notNull(step, "A Step must be provided."); + final Set> tasks = new HashSet>(getGridSize()); + final Set result = new HashSet(); - Assert.notNull(step, "A Step must be provided."); - - Set> tasks = new HashSet>(gridSize); + for (final StepExecution stepExecution : partitionStepExecutions) { + final FutureTask task = createTask(step, stepExecution); - Collection result = new ArrayList(); + try { + taskExecutor.execute(task); + tasks.add(task); + } catch (TaskRejectedException e) { + // couldn't execute one of the tasks + ExitStatus exitStatus = ExitStatus.FAILED + .addExitDescription("TaskExecutor rejected the task for this step."); + /* + * Set the status in case the caller is tracking it through the + * JobExecution. + */ + stepExecution.setStatus(BatchStatus.FAILED); + stepExecution.setExitStatus(exitStatus); + result.add(stepExecution); + } + } - for (final StepExecution stepExecution : stepExecutionSplitter.split(masterStepExecution, gridSize)) { - - final FutureTask task = new FutureTask(new Callable() { - public StepExecution call() throws Exception { - step.execute(stepExecution); - return stepExecution; - } - }); - - try { - taskExecutor.execute(task); - tasks.add(task); - } - catch (TaskRejectedException e) { - // couldn't execute one of the tasks - ExitStatus exitStatus = ExitStatus.FAILED - .addExitDescription("TaskExecutor rejected the task for this step."); - /* - * Set the status in case the caller is tracking it through the - * JobExecution. - */ - stepExecution.setStatus(BatchStatus.FAILED); - stepExecution.setExitStatus(exitStatus); - result.add(stepExecution); - } - - } - - for (Future task : tasks) { - result.add(task.get()); - } - return result; + for (Future task : tasks) { + result.add(task.get()); + } + return result; } + /** + * Creates the task executing the given step in the context of the given execution. + * + * @param step the step to execute + * @param stepExecution the given execution + * @return the task executing the given step + */ + protected FutureTask createTask(final Step step, + final StepExecution stepExecution) { + return new FutureTask(new Callable() { + public StepExecution call() throws Exception { + step.execute(stepExecution); + return stepExecution; + } + }); + } + }