diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java index 227f67f3e..bf0ebbe14 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java @@ -84,7 +84,7 @@ public class ChunkOrientedTasklet implements Tasklet { chunkContext.removeAttribute(INPUTS_KEY); chunkContext.setComplete(); - logger.debug("Inputs not busy, ended: "+inputs.isEnd()); + logger.debug("Inputs not busy, ended: " + inputs.isEnd()); return RepeatStatus.continueIf(!inputs.isEnd()); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java index 07a04f157..6445278e6 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java @@ -114,7 +114,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi */ protected final O doProcess(I item) throws Exception { - if (itemProcessor==null) { + if (itemProcessor == null) { @SuppressWarnings("unchecked") O result = (O) item; return result; @@ -141,7 +141,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi */ protected final void doWrite(List items) throws Exception { - if (itemWriter==null) { + if (itemWriter == null) { return; } @@ -167,7 +167,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi } protected void writeItems(List items) throws Exception { - if (itemWriter!=null) { + if (itemWriter != null) { itemWriter.write(items); } } @@ -265,7 +265,17 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi * @throws Exception if there is a problem */ protected void write(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { - doWrite(outputs.getItems()); + try { + doWrite(outputs.getItems()); + } + catch (Exception e) { + /* + * For a simple chunk processor (no fault tolerance) we are done + * here, so prevent any more processing of these inputs. + */ + inputs.clear(); + throw e; + } contribution.incrementWriteCount(outputs.size()); } @@ -273,7 +283,18 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi Chunk outputs = new Chunk(); for (Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { final I item = iterator.next(); - O output = doProcess(item); + O output; + try { + output = doProcess(item); + } + catch (Exception e) { + /* + * For a simple chunk processor (no fault tolerance) we are done + * here, so prevent any more processing of these inputs. + */ + inputs.clear(); + throw e; + } if (output != null) { outputs.add(output); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java index 2a6ed338a..d66a64931 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java @@ -73,7 +73,6 @@ public class TaskletStepExceptionTests { @Test public void testApplicationException() throws Exception { - taskletStep.execute(stepExecution); assertEquals(FAILED, stepExecution.getStatus()); assertEquals(FAILED.toString(), stepExecution.getExitStatus().getExitCode()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java index 03dd65336..972a8b08a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java @@ -19,11 +19,13 @@ package org.springframework.batch.core.step.tasklet; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; -import org.junit.Before; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; @@ -32,9 +34,11 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.JobRepositorySupport; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemStreamSupport; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.item.support.PassThroughItemProcessor; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; @@ -44,6 +48,8 @@ import org.springframework.util.StringUtils; public class AsyncTaskletStepTests { + private static Log logger = LogFactory.getLog(AsyncTaskletStepTests.class); + private List processed = new CopyOnWriteArrayList(); private TaskletStep step; @@ -53,25 +59,32 @@ public class AsyncTaskletStepTests { ItemWriter itemWriter = new ItemWriter() { public void write(List data) throws Exception { // Thread.sleep(100L); + logger.info("Items: " + data); processed.addAll(data); + if (data.contains("fail")) { + throw new RuntimeException("Planned"); + } } }; private JobRepository jobRepository; - @Before - public void setUp() throws Exception { + private List items; + + private int concurrencyLimit = 300; + + private ItemProcessor itemProcessor = new PassThroughItemProcessor(); + + private void setUp() throws Exception { step = new TaskletStep("stepName"); ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager(); step.setTransactionManager(transactionManager); - List items = Arrays.asList(StringUtils - .commaDelimitedListToStringArray("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25")); RepeatTemplate chunkTemplate = new RepeatTemplate(); chunkTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); - step.setTasklet(new TestingChunkOrientedTasklet(new ListItemReader(items), itemWriter, + step.setTasklet(new TestingChunkOrientedTasklet(new ListItemReader(items), itemProcessor, itemWriter, chunkTemplate)); jobRepository = new JobRepositorySupport(); @@ -80,7 +93,7 @@ public class AsyncTaskletStepTests { TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); template.setThrottleLimit(throttleLimit); SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(300); + taskExecutor.setConcurrencyLimit(concurrencyLimit); template.setTaskExecutor(taskExecutor); step.setStepOperations(template); @@ -101,6 +114,11 @@ public class AsyncTaskletStepTests { @Test public void testStepExecutionUpdates() throws Exception { + items = new ArrayList(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25"))); + + setUp(); + JobExecution jobExecution = jobRepository.createJobExecution("JOB", new JobParameters()); StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); @@ -114,11 +132,86 @@ public class AsyncTaskletStepTests { // System.err.println(processed); // Check commit count didn't spin out of control waiting for other // threads to finish... - assertTrue("Not enough commits: " + stepExecution.getCommitCount(), stepExecution.getCommitCount() > processed - .size() / 2); - assertTrue("Too many commits: " + stepExecution.getCommitCount(), stepExecution.getCommitCount() <= processed - .size() - / 2 + throttleLimit + 1); + assertTrue("Not enough commits: " + stepExecution.getCommitCount(), + stepExecution.getCommitCount() > processed.size() / 2); + assertTrue("Too many commits: " + stepExecution.getCommitCount(), + stepExecution.getCommitCount() <= processed.size() / 2 + throttleLimit + 1); + + } + + /** + * StepExecution should fail immediately on error. + */ + @Test + public void testStepExecutionFails() throws Exception { + + throttleLimit = 1; + concurrencyLimit = 1; + items = Arrays.asList("one", "fail", "three", "four"); + setUp(); + + JobExecution jobExecution = jobRepository.createJobExecution("JOB", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); + + step.execute(stepExecution); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getReadCount()); + assertEquals(2, processed.size()); + + } + + /** + * StepExecution should fail immediately on error in processor. + */ + @Test + public void testStepExecutionFailsWithProcessor() throws Exception { + + throttleLimit = 1; + concurrencyLimit = 1; + items = Arrays.asList("one", "barf", "three", "four"); + itemProcessor = new ItemProcessor() { + public String process(String item) throws Exception { + logger.info("Item: "+item); + processed.add(item); + if (item.equals("barf")) { + throw new RuntimeException("Planned processor error"); + } + return item; + } + }; + setUp(); + + JobExecution jobExecution = jobRepository.createJobExecution("JOB", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); + + step.execute(stepExecution); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getReadCount()); + assertEquals(2, processed.size()); + + } + + /** + * StepExecution should fail immediately on error. + */ + @Test + public void testStepExecutionFailsOnLastItem() throws Exception { + + throttleLimit = 1; + concurrencyLimit = 1; + items = Arrays.asList("one", "two", "three", "fail"); + setUp(); + + JobExecution jobExecution = jobRepository.createJobExecution("JOB", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); + + step.execute(stepExecution); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(4, stepExecution.getReadCount()); + assertEquals(4, processed.size()); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java index 67b4fc89b..ad002b7be 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java @@ -18,6 +18,7 @@ package org.springframework.batch.core.step.tasklet; import org.springframework.batch.core.step.item.ChunkOrientedTasklet; import org.springframework.batch.core.step.item.SimpleChunkProcessor; import org.springframework.batch.core.step.item.SimpleChunkProvider; +import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.PassThroughItemProcessor; @@ -56,10 +57,19 @@ public class TestingChunkOrientedTasklet extends ChunkOrientedTasklet { * Creates a {@link PassThroughItemProcessor} and uses it to create an * instance of {@link Tasklet}. */ - public TestingChunkOrientedTasklet(ItemReader itemReader, ItemWriter itemWriter, + public TestingChunkOrientedTasklet(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations repeatOperations) { super(new SimpleChunkProvider(itemReader, repeatOperations), new SimpleChunkProcessor( - new PassThroughItemProcessor(), itemWriter)); + itemProcessor, itemWriter)); + } + + /** + * Creates a {@link PassThroughItemProcessor} and uses it to create an + * instance of {@link Tasklet}. + */ + public TestingChunkOrientedTasklet(ItemReader itemReader, ItemWriter itemWriter, + RepeatOperations repeatOperations) { + this(itemReader, new PassThroughItemProcessor(), itemWriter, repeatOperations); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java index 71a8e0c71..4825d10d3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java @@ -176,6 +176,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { if (future.getError() != null) { state.getThrowables().add(future.getError()); + result = false; } else { RepeatStatus status = future.getResult();