diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java index f276391be..61eda773e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java @@ -26,4 +26,15 @@ package org.springframework.batch.core; public enum BatchStatus { COMPLETED, STARTED, STARTING, FAILED, STOPPING, STOPPED, UNKNOWN; + + public static BatchStatus max(BatchStatus status1, BatchStatus status2) { + if (status1.compareTo(status2)<0) { + return status2; + } + if (status1.compareTo(status2)>0) { + return status1; + } + else return status1; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/PartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/PartitionHandler.java new file mode 100644 index 000000000..2169cca6d --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/PartitionHandler.java @@ -0,0 +1,38 @@ +package org.springframework.batch.core.partition; + +import java.util.Collection; + +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.ExecutionContext; + +/** + * Interface defining the responsibilities of controlling the execution of a + * partitioned {@link StepExecution}. Implementations will need to create a + * partition with the {@link StepExecutionSplitter}, and then use an execution + * fabric (grid, etc.), to execute the partitioned step. The results of the + * executions can be returned raw from remote workers to be aggregated by the + * caller. + * + * @author Dave Syer + * + */ +public interface PartitionHandler { + + /** + * Main entry point for {@link PartitionHandler} interface. The splitter + * creates all the executions that need to be farmed out, along with their + * input parameters (in the form of their {@link ExecutionContext}). The + * master step execution is used to identify the partition and group + * together the results logically. + * + * @param stepSplitter a strategy for generating a collection of + * {@link StepExecution} instances + * @param stepExecution the master step execution for the whole partition + * @return a collection of 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. + */ + Collection handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution) throws Exception; + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/StepExecutionSplitter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/StepExecutionSplitter.java new file mode 100644 index 000000000..f49a36132 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/StepExecutionSplitter.java @@ -0,0 +1,47 @@ +package org.springframework.batch.core.partition; + +import java.util.Set; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.StepExecution; + +/** + * Strategy interface for generating input contexts for a partitioned step + * execution. + * + * @author Dave Syer + * + */ +public interface StepExecutionSplitter { + + /** + * The name of the step configuration that will be executed remotely. Remote + * workers are going to execute a the same step for each execution context + * in the partition. + * @return the name of the step that will execute the business logic + */ + String getStepName(); + + /** + * Partition the provided {@link StepExecution} into a set of parallel + * executable instances with the same parent {@link JobExecution}. The grid + * size will be treated as a hint for the size of the collection to be + * returned. It may or may not correspond to the physical size of an + * execution grid.
+ *
+ * + * On a restart clients of the {@link StepExecutionSplitter} should expect + * it to reconstitute the state of the last failed execution and only return + * those executions that need to be restarted. Thus the grid size hint + * should be ignored on a restart. + * + * @param stepExecution the {@link StepExecution} to be partitioned. + * @param gridSize a hint for the splitter if the size of the grid is known + * @return a set of {@link StepExecution} instances for remote processing + * + * @throws JobExecutionException if the split cannot be made + */ + Set split(StepExecution stepExecution, int gridSize) throws JobExecutionException; + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java new file mode 100644 index 000000000..261a37f70 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java @@ -0,0 +1,86 @@ +package org.springframework.batch.core.partition.support; + +import java.util.Collection; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecutionException; +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.AbstractStep; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.util.Assert; + +/** + * Implementation of {@link Step} which partitions the execution and spreads the + * load using a {@link PartitionHandler}. + * + * @author Dave Syer + * + */ +public class PartitionStep extends AbstractStep { + + private StepExecutionSplitter stepExecutionSplitter; + + private PartitionHandler partitionHandler; + + private StepExecutionAggregator aggregator = new StepExecutionAggregator(); + + /** + * Public setter for mandatory property {@link PartitionHandler}. + * @param partitionHandler the {@link PartitionHandler} to set + */ + public void setPartitionHandler(PartitionHandler partitionHandler) { + this.partitionHandler = partitionHandler; + } + + /** + * Public setter for mandatory property {@link StepExecutionSplitter}. + * @param stepExecutionSplitter the {@link StepExecutionSplitter} to set + */ + public void setStepExecutionSplitter(StepExecutionSplitter stepExecutionSplitter) { + this.stepExecutionSplitter = stepExecutionSplitter; + } + + /** + * Assert that mandatory properties are set (stepExecutionSplitter, + * partitionHandler) and delegate top superclass. + * + * @see AbstractStep#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(stepExecutionSplitter, "StepExecutionSplitter must be provided"); + Assert.notNull(partitionHandler, "PartitionHandler must be provided"); + super.afterPropertiesSet(); + } + + /** + * Delegate execution to the {@link PartitionHandler} provided. The + * {@link StepExecution} passed in here becomes the parent or master + * execution for the partition, summarising the status on exit of the + * logical grouping of work carried out by the {@link PartitionHandler}. The + * individual step executions and their input parameters (through + * {@link ExecutionContext}) for the partition elements are provided by the + * {@link StepExecutionSplitter}. + * + * @param stepExecution the master step execution for the partition + * + * @see Step#execute(StepExecution) + */ + @Override + protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { + + // Wait for task completion and then aggregate the results + Collection executions = partitionHandler.handle(stepExecutionSplitter, stepExecution); + aggregator.aggregate(stepExecution, executions); + if (stepExecution.getStatus()!=BatchStatus.COMPLETED) { + throw new JobExecutionException("Partition handler returned an incomplete step"); + } + + return stepExecution.getExitStatus(); + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/Partitioner.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/Partitioner.java new file mode 100644 index 000000000..57c59a24d --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/Partitioner.java @@ -0,0 +1,11 @@ +package org.springframework.batch.core.partition.support; + +import java.util.Map; + +import org.springframework.batch.item.ExecutionContext; + +public interface Partitioner { + + Map partition(int gridSize); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimplePartitioner.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimplePartitioner.java new file mode 100644 index 000000000..8755532db --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimplePartitioner.java @@ -0,0 +1,20 @@ +package org.springframework.batch.core.partition.support; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.batch.item.ExecutionContext; + +public class SimplePartitioner implements Partitioner { + + private static final String PARTITION_KEY = "partition"; + + public Map partition(int gridSize) { + Map map = new HashMap(gridSize); + for (int i = 0; i < gridSize; i++) { + map.put(PARTITION_KEY + i, new ExecutionContext()); + } + return map; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java new file mode 100644 index 000000000..bfe37b152 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitter.java @@ -0,0 +1,134 @@ +package org.springframework.batch.core.partition.support; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.item.ExecutionContext; + +public class SimpleStepExecutionSplitter implements StepExecutionSplitter { + + private static final String STEP_NAME_SEPARATOR = ":"; + + private final String stepName; + + private final Partitioner partitioner; + + private final Step step; + + private final JobRepository jobRepository; + + public SimpleStepExecutionSplitter(JobRepository jobRepository, Step step) { + this(jobRepository, step, new SimplePartitioner()); + } + + public SimpleStepExecutionSplitter(JobRepository jobRepository, Step step, Partitioner partitioner) { + this.jobRepository = jobRepository; + this.step = step; + this.partitioner = partitioner; + this.stepName = step.getName(); + } + + /** + * @see StepExecutionSplitter#getStepName() + */ + public String getStepName() { + return this.stepName; + } + + /** + * @see StepExecutionSplitter#split(StepExecution, int) + */ + public Set split(StepExecution stepExecution, int gridSize) throws JobExecutionException { + + JobExecution jobExecution = stepExecution.getJobExecution(); + + // If this is a restart we must retain the same grid size, ignoring the + // one passed in... + int splitSize = getSplitSize(stepExecution, gridSize); + + Map contexts = partitioner.partition(splitSize); + Set set = new HashSet(contexts.size()); + + for (String key : contexts.keySet()) { + + // Make the step execution name unique and repeatable + String stepName = this.stepName + STEP_NAME_SEPARATOR + key; + + StepExecution currentStepExecution = jobExecution.createStepExecution(stepName); + + boolean startable = getStartable(currentStepExecution, contexts.get(key)); + + if (startable) { + set.add(currentStepExecution); + } + + } + + return set; + + } + + private int getSplitSize(StepExecution stepExecution, int gridSize) { + ExecutionContext context = stepExecution.getExecutionContext(); + int result = (int) context.getLong("GRID_SIZE", gridSize); + context.putLong("GRID_SIZE", result); + if (context.isDirty()) { + jobRepository.updateExecutionContext(stepExecution); + } + return result; + } + + private boolean getStartable(StepExecution stepExecution, ExecutionContext context) throws JobExecutionException { + + JobInstance jobInstance = stepExecution.getJobExecution().getJobInstance(); + String stepName = stepExecution.getStepName(); + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, stepName); + + boolean isRestart = (lastStepExecution != null && lastStepExecution.getStatus() != BatchStatus.COMPLETED) ? true + : false; + + if (isRestart) { + stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); + } + else { + stepExecution.setExecutionContext(context); + } + + return shouldStart(step, lastStepExecution) || isRestart; + + } + + private boolean shouldStart(Step step, StepExecution lastStepExecution) throws JobExecutionException { + + if (lastStepExecution == null) { + return true; + } + + BatchStatus stepStatus = lastStepExecution.getStatus(); + + if (stepStatus == BatchStatus.UNKNOWN) { + throw new JobExecutionException("Cannot restart step from UNKNOWN status. " + + "The last execution ended with a failure that could not be rolled back, " + + "so it may be dangerous to proceed. " + "Manual intervention is probably necessary."); + } + + if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { + // step is complete, false should be returned, indicating that the + // step should not be started + return false; + } + + return true; + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java new file mode 100644 index 000000000..eaa49b435 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java @@ -0,0 +1,31 @@ +package org.springframework.batch.core.partition.support; + +import java.util.Collection; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.StepExecution; +import org.springframework.util.Assert; + +public class StepExecutionAggregator { + + public void aggregate(StepExecution result, Collection executions) { + Assert.notNull(result, "To aggregate into a result it must be non-null."); + if (executions == null || executions.isEmpty()) { + throw new IllegalArgumentException("Cannot aggregate empty or null executions: " + executions); + } + // Start with assumption that it is complete... + result.setStatus(BatchStatus.COMPLETED); + for (StepExecution stepExecution : executions) { + BatchStatus status = stepExecution.getStatus(); + result.setStatus(BatchStatus.max(result.getStatus(), status)); + result.setExitStatus(result.getExitStatus().and(stepExecution.getExitStatus())); + result.setCommitCount(result.getCommitCount() + stepExecution.getCommitCount()); + result.setRollbackCount(result.getRollbackCount() + stepExecution.getRollbackCount()); + result.setReadCount(result.getReadCount() + stepExecution.getReadCount()); + result.setReadSkipCount(result.getReadSkipCount() + stepExecution.getReadSkipCount()); + result.setWriteCount(result.getWriteCount() + stepExecution.getWriteCount()); + result.setWriteSkipCount(result.getWriteSkipCount() + stepExecution.getWriteSkipCount()); + } + } + +} 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 new file mode 100644 index 000000000..a15478039 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandler.java @@ -0,0 +1,97 @@ +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; +import java.util.concurrent.FutureTask; + +import org.springframework.batch.core.BatchStatus; +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.repeat.ExitStatus; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.TaskRejectedException; +import org.springframework.util.Assert; + +public class TaskExecutorPartitionHandler implements PartitionHandler, InitializingBean { + + private int gridSize = 1; + + private TaskExecutor taskExecutor = new SyncTaskExecutor(); + + private Step step; + + public void afterPropertiesSet() throws Exception { + Assert.notNull(step, "A Step must be provided."); + } + + public void setGridSize(int gridSize) { + this.gridSize = gridSize; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setStep(Step step) { + this.step = step; + } + + /** + * @see PartitionHandler#handle(StepExecutionSplitter, StepExecution) + */ + public Collection handle(StepExecutionSplitter stepExecutionSplitter, + StepExecution masterStepExecution) throws Exception { + + Set> tasks = new HashSet>(gridSize); + + Collection result = new ArrayList(); + + 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(new Runnable() { + public void run() { + task.run(); + } + }); + 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."); + /* + * This stepExecution hasn't been saved yet, but we'll set the + * status anyway in case the caller is tracking it through the + * JobExecution. + */ + stepExecution.setStatus(BatchStatus.FAILED); + stepExecution.setExitStatus(exitStatus); + result.add(stepExecution); + } + + } + + for (FutureTask task : tasks) { + // TODO: timeout / heart beat + result.add(task.get()); + } + return result; + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java index 63a18e6ec..08de29a2c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java @@ -125,7 +125,10 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean, I createExecutionContextDao()); } - public Object getObject() throws Exception { + public Object getObject() throws Exception { + if (proxyFactory==null) { + afterPropertiesSet(); + } return proxyFactory.getProxy(); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/ChunkContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/ChunkContext.java new file mode 100644 index 000000000..ead7c8ff0 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/ChunkContext.java @@ -0,0 +1,10 @@ +package org.springframework.batch.core.scope; + +import org.springframework.core.AttributeAccessorSupport; + +/** + * @author Dave Syer + * + */ +public class ChunkContext extends AttributeAccessorSupport { +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java index f1cb6ce7e..55c7f7bcd 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/StepContextRepeatCallback.java @@ -15,7 +15,11 @@ */ package org.springframework.batch.core.scope; +import java.util.Queue; +import java.util.concurrent.LinkedBlockingQueue; + import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; @@ -29,13 +33,15 @@ import org.springframework.batch.repeat.RepeatContext; */ public abstract class StepContextRepeatCallback implements RepeatCallback { + private final Queue attributeQueue = new LinkedBlockingQueue(); + private final StepContext stepContext; /** - * @param stepContext + * @param stepExecution */ - public StepContextRepeatCallback(StepContext stepContext) { - this.stepContext = stepContext; + public StepContextRepeatCallback(StepExecution stepExecution) { + this.stepContext = new StepContext(stepExecution); } /** @@ -47,20 +53,44 @@ public abstract class StepContextRepeatCallback implements RepeatCallback { * @see RepeatCallback#doInIteration(RepeatContext) */ public ExitStatus doInIteration(RepeatContext context) throws Exception { + + ChunkContext chunkContext = attributeQueue.poll(); + if (chunkContext == null) { + chunkContext = new ChunkContext(); + } + int start = stepContext.attributeNames().length; + + // The StepContext has to be the same for all chunks, + // otherwise step-scoped beans will be re-initialised for each chunk. StepSynchronizationManager.register(stepContext); try { return doInStepContext(context, stepContext); } finally { + // Still some stuff to do with the data in this chunk, + // pass it back + if (chunkContext.attributeNames().length > start) { + attributeQueue.add(chunkContext); + } StepSynchronizationManager.close(); } } /** - * @param context - * @param stepContext + * Do the work required for this portion of the step. The + * {@link StepContext} provided is managed by the base class, so that if + * there is still work to do for the task in hand state can be stored here. + * In a multi-threaded client, the base class ensures that only one thread + * at a time can be working on each instance of {@link StepContext}. Workers + * should signal that they are finished with a context by removing all the + * attributes they have added. If a worker does not remove them another + * thread might see stale state. + * + * @param context the current {@link RepeatContext} + * @param stepContext the step context in which to carry out the work * @return the exit status from the execution - * @throws Exception + * @throws Exception implementations can throw an exception if anything goes + * wrong */ public abstract ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/BasicAttributeAccessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/BasicAttributeAccessor.java deleted file mode 100644 index 874500af4..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/BasicAttributeAccessor.java +++ /dev/null @@ -1,10 +0,0 @@ -package org.springframework.batch.core.step.tasklet; - -import org.springframework.core.AttributeAccessorSupport; - -/** - * @author Dave Syer - * - */ -public class BasicAttributeAccessor extends AttributeAccessorSupport { -} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 75d6b56bf..5439753b9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -15,9 +15,6 @@ */ package org.springframework.batch.core.step.tasklet; -import java.util.Queue; -import java.util.concurrent.LinkedBlockingQueue; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; @@ -42,7 +39,6 @@ import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.core.AttributeAccessor; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; @@ -79,13 +75,13 @@ public class TaskletStep extends AbstractStep { private PlatformTransactionManager transactionManager; - private TransactionAttribute transactionAttribute = new DefaultTransactionAttribute(){ + private TransactionAttribute transactionAttribute = new DefaultTransactionAttribute() { @Override public boolean rollbackOn(Throwable ex) { return true; } - + }; private Tasklet tasklet; @@ -217,9 +213,8 @@ public class TaskletStep extends AbstractStep { * a transaction. The step and its execution and execution context are all * given an up to date {@link BatchStatus}, and the {@link JobRepository} is * used to store the result. Various reporting information are also added to - * the current context (the {@link RepeatContext} governing the step - * execution, which would normally be available to the caller somehow - * through the step's {@link ExecutionContext}.
+ * the current context governing the step execution, which would normally be + * available to the caller through the step's {@link ExecutionContext}.
* * @throws JobInterruptedException if the step or a chunk is interrupted * @throws RuntimeException if there is an exception during a chunk @@ -228,15 +223,14 @@ public class TaskletStep extends AbstractStep { */ @Override protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { - StepContext stepContext = new StepContext(stepExecution); + stream.update(stepExecution.getExecutionContext()); getJobRepository().updateExecutionContext(stepExecution); - return stepOperations.iterate(new StepContextRepeatCallback(stepContext ) { + return stepOperations.iterate(new StepContextRepeatCallback(stepExecution) { - final Queue attributeQueue = new LinkedBlockingQueue(); - - public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception { + @Override + public ExitStatus doInStepContext(RepeatContext repeatContext, StepContext stepContext) throws Exception { StepExecution stepExecution = stepContext.getStepExecution(); ExceptionHolder fatalException = new ExceptionHolder(); @@ -254,23 +248,12 @@ public class TaskletStep extends AbstractStep { boolean locked = false; - AttributeAccessor attributes = attributeQueue.poll(); - if (attributes == null) { - attributes = new BasicAttributeAccessor(); - } - try { try { - exitStatus = tasklet.execute(contribution, attributes); + exitStatus = tasklet.execute(contribution, stepContext); } finally { - // Still some stuff to do with the data in this chunk, - // pass it back - if (attributes.attributeNames().length > 0) { - attributeQueue.add(attributes); - } - // Apply the contribution to the step // even if unsuccessful logger.debug("Applying contribution: " + contribution); @@ -294,8 +277,7 @@ public class TaskletStep extends AbstractStep { // Check to make sure the ExecutionContext hasn't be // modified outside a chunk boundary. Doing so will cause - // potential - // rollback issues. + // potential rollback issues. if (stepExecution.getExecutionContext().isDirty()) { throw new IllegalStateException( "The ExecutionContext cannot be modified outside of the ItemStream#Update method"); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/BatchStatusTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/BatchStatusTests.java index 3022ecde8..dcfa815d9 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/BatchStatusTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/BatchStatusTests.java @@ -39,6 +39,13 @@ public class BatchStatusTests { assertEquals("FAILED", BatchStatus.FAILED.toString()); } + @Test + public void testMaxStatus() { + assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.FAILED,BatchStatus.COMPLETED)); + assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.COMPLETED, BatchStatus.FAILED)); + assertEquals(BatchStatus.FAILED, BatchStatus.max(BatchStatus.FAILED, BatchStatus.FAILED)); + } + @Test public void testGetStatus() { assertEquals(BatchStatus.FAILED, BatchStatus.valueOf(BatchStatus.FAILED.toString())); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java new file mode 100644 index 000000000..f64225174 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReader.java @@ -0,0 +1,57 @@ +package org.springframework.batch.core.partition; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; + +/** + * {@link ItemReader} with hard-coded input data. + */ +public class ExampleItemReader implements ItemReader, ItemStream { + + private Log logger = LogFactory.getLog(getClass()); + + private String[] input = { "Hello", "world!", "Go", "on", "punk", "make", "my", "day!" }; + + private int index = 0; + + public static volatile boolean fail = false; + + /** + * Reads next record from input + */ + public String read() throws Exception { + if (index >= input.length) { + return null; + } + logger.info(String.format("Processing input index=%s, item=%s, in (%s)", index, input[index], this)); + if (fail && index == 4) { + synchronized (ExampleItemReader.class) { + if (fail) { + // Only fail once per flag setting... + fail = false; + logger.info(String.format("Throwing exception index=%s, item=%s, in (%s)", index, input[index], + this)); + index++; + throw new RuntimeException("Planned failure"); + } + } + } + return input[index++]; + } + + public void close(ExecutionContext executionContext) throws ItemStreamException { + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + index = (int) executionContext.getLong("POSITION", 0); + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + executionContext.putLong("POSITION", index); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReaderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReaderTests.java new file mode 100644 index 000000000..e78e4b5a1 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemReaderTests.java @@ -0,0 +1,70 @@ +package org.springframework.batch.core.partition; + +import static org.junit.Assert.*; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; + +public class ExampleItemReaderTests { + + private ExampleItemReader reader = new ExampleItemReader(); + + @Before + @After + public void ensureFailFlagUnset() { + ExampleItemReader.fail = false; + } + + @Test + public void testRead() throws Exception { + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(8, count); + } + + @Test + public void testOpen() throws Exception { + ExecutionContext context = new ExecutionContext(); + for (int i=0; i<4; i++) { + reader.read(); + } + reader.update(context); + reader.open(context); + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(4, count); + } + + @Test + public void testFailAndRestart() throws Exception { + ExecutionContext context = new ExecutionContext(); + ExampleItemReader.fail = true; + for (int i=0; i<4; i++) { + reader.read(); + reader.update(context); + } + try { + reader.read(); + reader.update(context); + fail("Expected Exception"); + } + catch (Exception e) { + // expected + assertEquals("Planned failure", e.getMessage()); + } + assertFalse(ExampleItemReader.fail); + reader.open(context); + int count = 0; + while (reader.read()!=null) { + count++; + } + assertEquals(4, count); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemWriter.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemWriter.java new file mode 100644 index 000000000..2984e21fe --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/ExampleItemWriter.java @@ -0,0 +1,23 @@ +package org.springframework.batch.core.partition; + +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ItemWriter; + +/** + * Dummy {@link ItemWriter} which only logs data it receives. + */ +public class ExampleItemWriter implements ItemWriter { + + private static final Log log = LogFactory.getLog(ExampleItemWriter.class); + + /** + * @see ItemWriter#write(List) + */ + public void write(List data) throws Exception { + log.info(data); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/RestartIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/RestartIntegrationTests.java new file mode 100644 index 000000000..4d3a617c7 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/RestartIntegrationTests.java @@ -0,0 +1,94 @@ +/* + * 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; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import javax.sql.DataSource; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * + */ +@ContextConfiguration(locations = "launch-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class RestartIntegrationTests { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private Job job; + + private SimpleJdbcTemplate jdbcTemplate; + + @Autowired + public void setDataSource(DataSource dataSource) { + this.jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + @Test + public void testSimpleProperties() throws Exception { + assertNotNull(jobLauncher); + } + + @Before + @After + public void start() { + ExampleItemReader.fail = false; + } + + @Test + public void testLaunchJob() throws Exception { + + // Force failure in one of the parallel steps + ExampleItemReader.fail = true; + JobParameters jobParameters = new JobParametersBuilder().addString("restart", "yes").toJobParameters(); + + int beforeMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'"); + int beforePartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'"); + + JobExecution execution = jobLauncher.run(job, jobParameters); + assertEquals(BatchStatus.FAILED,execution.getStatus()); + assertNotNull(jobLauncher.run(job, jobParameters)); + + int afterMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'"); + int afterPartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'"); + + // Two attempts + assertEquals(2, afterMaster-beforeMaster); + // One failure and two successes + assertEquals(3, afterPartition-beforePartition); + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/VanillaIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/VanillaIntegrationTests.java new file mode 100644 index 000000000..b96c8a98b --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/VanillaIntegrationTests.java @@ -0,0 +1,71 @@ +/* + * 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; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import javax.sql.DataSource; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * + */ +@ContextConfiguration(locations="launch-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class VanillaIntegrationTests { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private Job job; + + private SimpleJdbcTemplate jdbcTemplate; + + @Autowired + public void setDataSource(DataSource dataSource) { + this.jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + @Test + public void testSimpleProperties() throws Exception { + assertNotNull(jobLauncher); + } + + @Test + public void testLaunchJob() throws Exception { + int beforeMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'"); + int beforePartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'"); + assertNotNull(jobLauncher.run(job, new JobParameters())); + int afterMaster = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME='step1:master'"); + int afterPartition = jdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STEP_EXECUTION where STEP_NAME like 'step1:partition%'"); + assertEquals(1, afterMaster-beforeMaster); + // Should be same as grid size in step splitter + assertEquals(2, afterPartition-beforePartition); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java new file mode 100644 index 000000000..7362a0338 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java @@ -0,0 +1,100 @@ +/* + * 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 static org.junit.Assert.assertEquals; + +import java.util.Collection; +import java.util.Set; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +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.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.StepSupport; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + + +/** + * @author Dave Syer + * + */ +public class PartitionStepTests { + + private PartitionStep step = new PartitionStep(); + private Step remote = new StepSupport("remote"); + private JobRepository jobRepository; + + @Before + public void setUp() throws Exception { + MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); + factory.setTransactionManager(new ResourcelessTransactionManager()); + jobRepository = (JobRepository) factory.getObject(); + step.setJobRepository(jobRepository); + } + + @Test + public void testVanillaStepExecution() throws Exception { + step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote)); + step.setPartitionHandler(new PartitionHandler() { + public Collection handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution) + throws Exception { + Set executions = stepSplitter.split(stepExecution, 2); + for (StepExecution execution : executions) { + execution.setStatus(BatchStatus.COMPLETED); + execution.setExitStatus(ExitStatus.FINISHED); + } + return executions; + } + }); + step.afterPropertiesSet(); + StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo"); + step.execute(stepExecution); + // one master and two workers + assertEquals(3, stepExecution.getJobExecution().getStepExecutions().size()); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + } + + @Test + public void testFailedStepExecution() throws Exception { + step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote)); + step.setPartitionHandler(new PartitionHandler() { + public Collection handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution) + throws Exception { + Set executions = stepSplitter.split(stepExecution, 2); + for (StepExecution execution : executions) { + execution.setStatus(BatchStatus.FAILED); + execution.setExitStatus(ExitStatus.FAILED); + } + return executions; + } + }); + step.afterPropertiesSet(); + StepExecution stepExecution = new JobExecution(0L).createStepExecution("foo"); + step.execute(stepExecution); + // one master and two workers + assertEquals(3, stepExecution.getJobExecution().getStepExecutions().size()); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java new file mode 100644 index 000000000..56f5546a4 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java @@ -0,0 +1,67 @@ +package org.springframework.batch.core.partition.support; + +import static org.junit.Assert.assertEquals; + +import java.util.Collections; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + +public class SimpleStepExecutionSplitterTests { + + private Step step; + + private JobRepository jobRepository; + + private StepExecution stepExecution = new StepExecution("bar", new JobExecution(11L)); + + @Before + public void setUp() throws Exception { + step = new TaskletStep("step"); + MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); + factory.setTransactionManager(new ResourcelessTransactionManager()); + jobRepository = (JobRepository) factory.getObject(); + } + + @Test + public void testSimpleStepExecutionProviderJobRepositoryStep() throws Exception { + SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step); + assertEquals(2, provider.split(stepExecution, 2).size()); + } + + @Test + public void testSimpleStepExecutionProviderJobRepositoryStepPartitioner() throws Exception { + final Map map = Collections.singletonMap("foo", new ExecutionContext()); + SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step, new Partitioner() { + public Map partition(int gridSize) { + return map; + } + }); + assertEquals(1, provider.split(stepExecution, 2).size()); + } + + @Test + public void testRememberGridSize() throws Exception { + SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step); + assertEquals(2, provider.split(stepExecution, 2).size()); + assertEquals(2, provider.split(stepExecution, 3).size()); + } + + @Test + public void testGetStepName() { + SimpleStepExecutionSplitter provider = new SimpleStepExecutionSplitter(jobRepository, step); + assertEquals("step", provider.getStepName()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java new file mode 100644 index 000000000..4edd8a8e9 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java @@ -0,0 +1,71 @@ +package org.springframework.batch.core.partition.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Collections; + +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.repeat.ExitStatus; + +public class StepExecutionAggregatorTests { + + private StepExecutionAggregator aggregator = new StepExecutionAggregator(); + + private JobExecution jobExecution = new JobExecution(11L); + + private StepExecution result = jobExecution.createStepExecution("aggregate"); + + private StepExecution stepExecution1 = jobExecution.createStepExecution("foo:1"); + + private StepExecution stepExecution2 = jobExecution.createStepExecution("foo:2"); + + @Test(expected = IllegalArgumentException.class) + public void testAggregateEmpty() { + aggregator.aggregate(result, Collections. emptySet()); + } + + @Test(expected = IllegalArgumentException.class) + public void testAggregateNull() { + aggregator.aggregate(result, null); + } + + @Test + public void testAggregateStatusSunnyDay() { + stepExecution1.setStatus(BatchStatus.COMPLETED); + stepExecution2.setStatus(BatchStatus.COMPLETED); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertNotNull(result); + assertEquals(BatchStatus.COMPLETED, result.getStatus()); + } + + @Test + public void testAggregateStatusIncomplete() { + stepExecution1.setStatus(BatchStatus.COMPLETED); + stepExecution2.setStatus(BatchStatus.FAILED); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertNotNull(result); + assertEquals(BatchStatus.FAILED, result.getStatus()); + } + + @Test + public void testAggregateExitStatusSunnyDay() { + stepExecution1.setExitStatus(ExitStatus.CONTINUABLE); + stepExecution2.setExitStatus(ExitStatus.FAILED); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertNotNull(result); + assertEquals(ExitStatus.FAILED.and(ExitStatus.CONTINUABLE), result.getExitStatus()); + } + + @Test + public void testAggregateCommitCountSunnyDay() { + stepExecution1.setCommitCount(10); + stepExecution2.setCommitCount(5); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertEquals(15, result.getCommitCount()); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java new file mode 100644 index 000000000..1e0b33dc0 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java @@ -0,0 +1,103 @@ +package org.springframework.batch.core.partition.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.core.step.StepSupport; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.core.task.TaskRejectedException; + +public class TaskExecutorPartitionHandlerTests { + + private TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler(); + + private int count = 0; + + private StepExecution stepExecution = new StepExecution("step", new JobExecution(1L)); + + private StepExecutionSplitter stepExecutionSplitter = new StepExecutionSplitter() { + + public String getStepName() { + return stepExecution.getStepName(); + } + + public Set split(StepExecution stepExecution, int gridSize) throws JobExecutionException { + HashSet result = new HashSet(); + for (int i = gridSize; i-- > 0;) { + result.add(stepExecution.getJobExecution().createStepExecution("foo" + i)); + } + return result; + } + }; + + @Before + public void setUp() throws Exception { + handler.setStep(new StepSupport() { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException { + count++; + } + }); + handler.afterPropertiesSet(); + } + + @Test + public void testAfterPropertiesSet() throws Exception { + handler = new TaskExecutorPartitionHandler(); + try { + handler.afterPropertiesSet(); + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException e) { + // expected + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.contains("Step")); + } + } + + @Test + public void testSetGridSize() throws Exception { + handler.setGridSize(2); + handler.handle(stepExecutionSplitter, stepExecution); + assertEquals(2, count); + } + + @Test + public void testSetTaskExecutor() throws Exception { + handler.setTaskExecutor(new SimpleAsyncTaskExecutor()); + handler.handle(stepExecutionSplitter, stepExecution); + assertEquals(1, count); + } + + @Test + public void testTaskExecutorFailure() throws Exception { + handler.setGridSize(2); + handler.setTaskExecutor(new TaskExecutor() { + public void execute(Runnable task) { + if (count > 0) { + throw new TaskRejectedException("foo"); + } + task.run(); + } + }); + Collection executions = handler.handle(stepExecutionSplitter, stepExecution); + new StepExecutionAggregator().aggregate(stepExecution, executions); + assertEquals(1, count); + assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java index f06b64816..388ed5906 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/StepContextRepeatCallbackTests.java @@ -16,6 +16,8 @@ package org.springframework.batch.core.scope; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.batch.core.JobExecution; @@ -25,21 +27,46 @@ import org.springframework.batch.repeat.RepeatContext; /** * @author Dave Syer - * + * */ public class StepContextRepeatCallbackTests { + + private StepExecution stepExecution = new StepExecution("foo", new JobExecution(0L), 123L); + private boolean addedAttribute = false; + private boolean removedAttribute = false; @Test public void testDoInIteration() throws Exception { - StepContext stepContext = new StepContext(new StepExecution("foo", new JobExecution(0L), 123L)); - StepContextRepeatCallback callback = new StepContextRepeatCallback(stepContext ) { + StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) { @Override public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception { assertEquals(Long.valueOf(123), stepContext.getStepExecution().getId()); return ExitStatus.NOOP; } - }; - assertEquals(ExitStatus.NOOP, callback.doInIteration(null)); + }; + assertEquals(ExitStatus.NOOP, callback.doInIteration(null)); } + @Test + public void testUnfinishedWork() throws Exception { + StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) { + @Override + public ExitStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception { + if (addedAttribute) { + removedAttribute = stepContext.hasAttribute("foo"); + stepContext.removeAttribute("foo"); + } else { + addedAttribute = true; + stepContext.setAttribute("foo", "bar"); + } + return ExitStatus.NOOP; + } + }; + callback.doInIteration(null); + assertTrue(addedAttribute); + callback.doInIteration(null); + assertTrue(removedAttribute); + callback.doInIteration(null); + assertFalse(removedAttribute); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java index 3c8bfa56a..84d4781a4 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/AbstractStepTests.java @@ -1,6 +1,7 @@ package org.springframework.batch.core.step; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -132,6 +133,34 @@ public class AbstractStepTests { public void setUp() throws Exception { tested.setJobRepository(repository); } + + @Test + public void testBeanName() throws Exception { + AbstractStep step = new AbstractStep() { + @Override + protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { + return null; + } + }; + assertNull(step.getName()); + step.setBeanName("foo"); + assertEquals("foo", step.getName()); + } + + @Test + public void testName() throws Exception { + AbstractStep step = new AbstractStep() { + @Override + protected ExitStatus doExecute(StepExecution stepExecution) throws Exception { + return null; + } + }; + assertNull(step.getName()); + step.setName("foo"); + assertEquals("foo", step.getName()); + step.setBeanName("bar"); + assertEquals("foo", step.getName()); + } /** * Typical step execution scenario. diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java index 4ce6ad683..89fc5266a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java @@ -28,9 +28,9 @@ import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.scope.ChunkContext; import org.springframework.batch.core.step.skip.ItemSkipPolicy; import org.springframework.batch.core.step.skip.SkipLimitExceededException; -import org.springframework.batch.core.step.tasklet.BasicAttributeAccessor; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; @@ -112,7 +112,7 @@ public class FaultTolerantChunkOrientedTaskletTests { handler = new FaultTolerantChunkOrientedTasklet(itemReader, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - handler.execute(contribution, new BasicAttributeAccessor()); + handler.execute(contribution, new ChunkContext()); assertEquals(limit, contribution.getReadCount()); } @@ -126,7 +126,7 @@ public class FaultTolerantChunkOrientedTaskletTests { writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + ChunkContext attributes = new ChunkContext(); try { handler.execute(contribution, attributes); fail("Expected SkipLimitExceededException"); @@ -148,7 +148,7 @@ public class FaultTolerantChunkOrientedTaskletTests { }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + ChunkContext attributes = new ChunkContext(); try { handler.execute(contribution, attributes); fail("Expected RuntimeException"); @@ -174,7 +174,7 @@ public class FaultTolerantChunkOrientedTaskletTests { }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + ChunkContext attributes = new ChunkContext(); // Count to 3: (try + skip + skip) for (int i = 0; i < 3; i++) { @@ -193,7 +193,7 @@ public class FaultTolerantChunkOrientedTaskletTests { // The last recovery for this chunk... handler.execute(contribution, attributes); - attributes = new BasicAttributeAccessor(); + attributes = new ChunkContext(); try { handler.execute(contribution, attributes); fail("Expected RuntimeException"); @@ -230,7 +230,7 @@ public class FaultTolerantChunkOrientedTaskletTests { writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + ChunkContext attributes = new ChunkContext(); // try try { @@ -267,7 +267,7 @@ public class FaultTolerantChunkOrientedTaskletTests { writeSkipPolicy); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - BasicAttributeAccessor attributes = new BasicAttributeAccessor(); + ChunkContext attributes = new ChunkContext(); // Count to 3: (try + skip + try) for (int i = 0; i < 2; i++) { @@ -288,7 +288,7 @@ public class FaultTolerantChunkOrientedTaskletTests { handler.execute(contribution, attributes); assertEquals(2, chunk.getSkips().size()); - attributes = new BasicAttributeAccessor(); + attributes = new ChunkContext(); try { handler.execute(contribution, attributes); fail("Expected RuntimeException"); diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml new file mode 100644 index 000000000..1b6492452 --- /dev/null +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/partition/launch-context.xml @@ -0,0 +1,67 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java index 2902a5a95..ec4098bbb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java @@ -129,9 +129,19 @@ public class ExitStatus implements Serializable { /** * Create a new {@link ExitStatus} with a logical combination of the - * continuable flag, and a concatenation of the descriptions. The exit code - * is only replaced if the result is continuable or the input is not - * continuable.
+ * continuable flag, and a concatenation of the descriptions. If either + * value has a higher severity then its exit code will be used in the + * result. In the case of equal severity, the exit code is only replaced if + * the result is continuable or the input is not continuable.
+ *
+ * + * Severity is defined by the exit code: + *
    + *
  • Codes beginning with NOOP have severity 1
    • + *
  • Codes beginning with FAILED have severity 2
    • + *
  • Codes beginning with UNKNOWN have severity 3
    • + *
+ * Others have severity 0.
* * If the input is null just return this. * @@ -144,19 +154,41 @@ public class ExitStatus implements Serializable { return this; } ExitStatus result = and(status.continuable).addExitDescription(status.exitDescription); - if (result.continuable || !status.continuable) { + if (severity(status) > severity(this)) { result = result.replaceExitCode(status.exitCode); } + else { + if (severity(this) == severity(status) && (result.continuable || !status.continuable)) { + result = result.replaceExitCode(status.exitCode); + } + } return result; } + /** + * @param status + * @return + */ + private int severity(ExitStatus status) { + if (status.exitCode.startsWith(NOOP.exitCode)) { + return 0; + } + if (status.exitCode.startsWith(FAILED.exitCode)) { + return 1; + } + if (status.exitCode.startsWith(UNKNOWN.exitCode)) { + return 2; + } + return 0; + } + /* * (non-Javadoc) * * @see java.lang.Object#toString() */ public String toString() { - return "continuable=" + continuable + ";exitCode=" + exitCode + ";exitDescription=" + exitDescription; + return String.format("continuable=%s;exitCode=%s;exitDescription=%s", continuable, exitCode, exitDescription); } /** diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/ExitStatusTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/ExitStatusTests.java index aef162b70..e42031a9b 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/ExitStatusTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/ExitStatusTests.java @@ -15,20 +15,25 @@ */ package org.springframework.batch.repeat; -import junit.framework.TestCase; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import org.apache.commons.lang.SerializationUtils; +import org.junit.Test; /** * @author Dave Syer * */ -public class ExitStatusTests extends TestCase { +public class ExitStatusTests { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)}. + * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)} + * . */ + @Test public void testExitStatusBooleanInt() { ExitStatus status = new ExitStatus(true, "10"); assertTrue(status.isContinuable()); @@ -37,8 +42,10 @@ public class ExitStatusTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)}. + * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)} + * . */ + @Test public void testExitStatusConstantsContinuable() { ExitStatus status = ExitStatus.CONTINUABLE; assertTrue(status.isContinuable()); @@ -47,8 +54,10 @@ public class ExitStatusTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)}. + * {@link org.springframework.batch.repeat.ExitStatus#ExitStatus(boolean, String)} + * . */ + @Test public void testExitStatusConstantsFinished() { ExitStatus status = ExitStatus.FINISHED; assertFalse(status.isContinuable()); @@ -60,15 +69,18 @@ public class ExitStatusTests extends TestCase { * * @throws Exception */ + @Test public void testEqualsWithSameProperties() throws Exception { assertEquals(ExitStatus.CONTINUABLE, new ExitStatus(true, "CONTINUABLE")); } + @Test public void testEqualsSelf() { ExitStatus status = new ExitStatus(true, "test"); assertEquals(status, status); } + @Test public void testEquals() { assertEquals(new ExitStatus(true, "test"), new ExitStatus(true, "test")); } @@ -78,6 +90,7 @@ public class ExitStatusTests extends TestCase { * * @throws Exception */ + @Test public void testEqualsWithNull() throws Exception { assertFalse(ExitStatus.CONTINUABLE.equals(null)); } @@ -87,6 +100,7 @@ public class ExitStatusTests extends TestCase { * * @throws Exception */ + @Test public void testHashcode() throws Exception { assertEquals(ExitStatus.CONTINUABLE.toString().hashCode(), ExitStatus.CONTINUABLE.hashCode()); } @@ -95,6 +109,7 @@ public class ExitStatusTests extends TestCase { * Test method for * {@link org.springframework.batch.repeat.ExitStatus#and(boolean)}. */ + @Test public void testAndBoolean() { assertTrue(ExitStatus.CONTINUABLE.and(true).isContinuable()); assertFalse(ExitStatus.CONTINUABLE.and(false).isContinuable()); @@ -105,8 +120,10 @@ public class ExitStatusTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)}. + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . */ + @Test public void testAndExitStatusStillContinuable() { assertTrue(ExitStatus.CONTINUABLE.and(ExitStatus.CONTINUABLE).isContinuable()); assertFalse(ExitStatus.CONTINUABLE.and(ExitStatus.FINISHED).isContinuable()); @@ -116,24 +133,30 @@ public class ExitStatusTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)}. + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . */ + @Test public void testAndExitStatusWhenFinishedAddedToContinuable() { assertEquals(ExitStatus.FINISHED.getExitCode(), ExitStatus.CONTINUABLE.and(ExitStatus.FINISHED).getExitCode()); } /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)}. + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . */ + @Test public void testAndExitStatusWhenContinuableAddedToFinished() { assertEquals(ExitStatus.FINISHED.getExitCode(), ExitStatus.FINISHED.and(ExitStatus.CONTINUABLE).getExitCode()); } /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)}. + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . */ + @Test public void testAndExitStatusWhenCustomContinuableAddedToContinuable() { assertEquals("CUSTOM", ExitStatus.CONTINUABLE.and(ExitStatus.CONTINUABLE.replaceExitCode("CUSTOM")) .getExitCode()); @@ -141,13 +164,27 @@ public class ExitStatusTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)}. + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . */ + @Test + public void testAndExitStatusFailedPlusFinished() { + assertEquals("FAILED", ExitStatus.FINISHED.and(ExitStatus.FAILED).getExitCode()); + assertEquals("FAILED", ExitStatus.FAILED.and(ExitStatus.FINISHED).getExitCode()); + } + + /** + * Test method for + * {@link org.springframework.batch.repeat.ExitStatus#and(org.springframework.batch.repeat.ExitStatus)} + * . + */ + @Test public void testAndExitStatusWhenCustomContinuableAddedToFinished() { assertEquals(ExitStatus.FINISHED.getExitCode(), ExitStatus.FINISHED.and( ExitStatus.CONTINUABLE.replaceExitCode("CUSTOM")).getExitCode()); } + @Test public void testAddExitCode() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.replaceExitCode("FOO"); assertTrue(ExitStatus.CONTINUABLE != status); @@ -155,6 +192,7 @@ public class ExitStatusTests extends TestCase { assertEquals("FOO", status.getExitCode()); } + @Test public void testAddExitCodeToExistingStatus() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.replaceExitCode("FOO").replaceExitCode("BAR"); assertTrue(ExitStatus.CONTINUABLE != status); @@ -162,6 +200,7 @@ public class ExitStatusTests extends TestCase { assertEquals("BAR", status.getExitCode()); } + @Test public void testAddExitCodeToSameStatus() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.replaceExitCode(ExitStatus.CONTINUABLE.getExitCode()); assertTrue(ExitStatus.CONTINUABLE != status); @@ -169,6 +208,7 @@ public class ExitStatusTests extends TestCase { assertEquals(ExitStatus.CONTINUABLE.getExitCode(), status.getExitCode()); } + @Test public void testAddExitDescription() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.addExitDescription("Foo"); assertTrue(ExitStatus.CONTINUABLE != status); @@ -176,6 +216,7 @@ public class ExitStatusTests extends TestCase { assertEquals("Foo", status.getExitDescription()); } + @Test public void testAddExitDescriptionToSameStatus() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.addExitDescription("Foo").addExitDescription("Foo"); assertTrue(ExitStatus.CONTINUABLE != status); @@ -183,21 +224,25 @@ public class ExitStatusTests extends TestCase { assertEquals("Foo", status.getExitDescription()); } + @Test public void testAddEmptyExitDescription() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.addExitDescription("Foo").addExitDescription(null); assertEquals("Foo", status.getExitDescription()); } + @Test public void testAddExitCodeWithDescription() throws Exception { ExitStatus status = new ExitStatus(true, "BAR", "Bar").replaceExitCode("FOO"); assertEquals("FOO", status.getExitCode()); assertEquals("Bar", status.getExitDescription()); } + @Test public void testUnkownIsRunning() throws Exception { assertTrue(ExitStatus.UNKNOWN.isRunning()); } + @Test public void testSerializable() throws Exception { ExitStatus status = ExitStatus.CONTINUABLE.replaceExitCode("FOO"); byte[] bytes = SerializationUtils.serialize(status);