diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index 66b57d300..93f217f2f 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -279,7 +279,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { TransactionStatus transaction = streamManager.getTransaction(stepExecution); try { - + itemReader.mark(); result = processChunk(contribution); // TODO: check that stepExecution can @@ -327,6 +327,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean { stepExecution.rollback(); } try { + itemReader.reset(); itemWriter.clear(); streamManager.rollback(transaction); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java index 471b41c63..061f1c992 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java @@ -27,8 +27,6 @@ import org.springframework.batch.io.exception.ReadFailureException; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.exception.MarkFailedException; -import org.springframework.batch.item.exception.ResetFailedException; import org.springframework.batch.item.exception.StreamException; import org.springframework.util.Assert; @@ -59,6 +57,7 @@ public class ItemChunker implements Chunker { public ChunkingResult chunk(int size, StepContribution stepContribution) throws ReadFailureException { Assert.isTrue(size > 0, "Chunk size must be greater than 0"); + itemReader.mark(); int counter = 0; List items = new ArrayList(size); List exceptions = new ArrayList(); @@ -75,6 +74,7 @@ public class ItemChunker implements Chunker { } catch (Exception ex) { exceptions.add(ex); if(!itemSkipPolicy.shouldSkip(ex, stepContribution)){ + itemReader.reset(); rethrow(ex); } } @@ -108,32 +108,12 @@ public class ItemChunker implements Chunker { } } - public boolean isMarkSupported() { - if(itemReader instanceof ItemStream){ - return ((ItemStream)itemReader).isMarkSupported(); - } - - return false; - } - - public void mark() throws MarkFailedException { - if(itemReader instanceof ItemStream){ - ((ItemStream)itemReader).mark(); - } - } - public void open() throws StreamException { if(itemReader instanceof ItemStream){ ((ItemStream)itemReader).open(); } } - public void reset() throws ResetFailedException { - if(itemReader instanceof ItemStream){ - ((ItemStream)itemReader).reset(); - } - } - public void restoreFrom(ExecutionContext context) { if(itemReader instanceof ItemStream){ ((ItemStream)itemReader).restoreFrom(context); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java index 8fe26c5f1..e5cc044ac 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java @@ -20,16 +20,14 @@ import java.util.Iterator; import java.util.List; import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.DechunkingResult; import org.springframework.batch.core.domain.Dechunker; +import org.springframework.batch.core.domain.DechunkingResult; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.io.exception.WriteFailureException; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.exception.MarkFailedException; -import org.springframework.batch.item.exception.ResetFailedException; import org.springframework.batch.item.exception.StreamException; import org.springframework.util.Assert; @@ -99,33 +97,12 @@ public class ItemDechunker implements Dechunker { } } - public boolean isMarkSupported() { - if(itemWriter instanceof ItemStream){ - return ((ItemStream)itemWriter).isMarkSupported(); - } - else{ - return false; - } - } - - public void mark() throws MarkFailedException { - if(itemWriter instanceof ItemStream){ - ((ItemStream)itemWriter).mark(); - } - } - public void open() throws StreamException { if(itemWriter instanceof ItemStream){ ((ItemStream)itemWriter).open(); } } - public void reset() throws ResetFailedException { - if(itemWriter instanceof ItemStream){ - ((ItemStream)itemWriter).reset(); - } - } - public void restoreFrom(ExecutionContext context) { if(itemWriter instanceof ItemStream){ ((ItemStream)itemWriter).restoreFrom(context); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java index 0ccd85308..e2d86b07d 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java @@ -36,6 +36,7 @@ import org.springframework.batch.execution.step.support.ItemDechunker; import org.springframework.batch.execution.step.support.JobRepositorySupport; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.writer.AbstractItemWriter; @@ -118,7 +119,7 @@ public class ChunkedStepTests extends TestCase { final JobExecution jobExecution = new JobExecution(jobInstance); final StepExecution stepExecution = new StepExecution("testStep", jobExecution); - chunkedStep.setChunker(new ItemChunker(new ItemReader() { + chunkedStep.setChunker(new ItemChunker(new AbstractItemReader() { int counter = 0; public Object read() throws Exception { assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); @@ -192,7 +193,7 @@ public class ChunkedStepTests extends TestCase { //fail. public void testReadFailure() { - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { int counter = 0; public Object read() throws Exception { @@ -242,7 +243,7 @@ public class ChunkedStepTests extends TestCase { public void testExitCodeDefaultClassification() throws Exception { - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { int counter = 0; public Object read() throws Exception { counter++; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java index a88f90037..4e79d67e3 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java @@ -46,6 +46,7 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.exception.ResetFailedException; import org.springframework.batch.item.exception.StreamException; +import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.item.stream.SimpleStreamManager; @@ -159,7 +160,7 @@ public class ItemOrientedStepTests extends TestCase { final JobExecution jobExecution = new JobExecution(jobInstance); final StepExecution stepExecution = new StepExecution(step, jobExecution); - itemOrientedStep.setItemReader(new ItemReader() { + itemOrientedStep.setItemReader(new AbstractItemReader() { public Object read() throws Exception { assertEquals(step, stepExecution.getStepName()); assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); @@ -214,7 +215,7 @@ public class ItemOrientedStepTests extends TestCase { public void testIncrementRollbackCount() { - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { public Object read() throws Exception { int counter = 0; @@ -245,7 +246,7 @@ public class ItemOrientedStepTests extends TestCase { public void testExitCodeDefaultClassification() throws Exception { - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { public Object read() throws Exception { int counter = 0; @@ -348,7 +349,7 @@ public class ItemOrientedStepTests extends TestCase { public void testRestartJobOnNonRestartableTasklet() throws Exception { String step = "stepName"; // step.setStepExecutionCount(1); - itemOrientedStep.setItemReader(new ItemReader() { + itemOrientedStep.setItemReader(new AbstractItemReader() { public Object read() throws Exception { return "foo"; } @@ -402,7 +403,7 @@ public class ItemOrientedStepTests extends TestCase { public void testStreamManager() throws Exception { String step = "stepName"; // step.setStepExecutionCount(1); - itemOrientedStep.setItemReader(new ItemReader() { + itemOrientedStep.setItemReader(new AbstractItemReader() { public Object read() throws Exception { return "foo"; } @@ -484,7 +485,7 @@ public class ItemOrientedStepTests extends TestCase { itemOrientedStep.setInterruptionPolicy(interruptionPolicy); - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { public Object read() throws Exception { int counter = 0; @@ -523,7 +524,7 @@ public class ItemOrientedStepTests extends TestCase { public void testStatusForResetFailedException() throws Exception { - ItemReader itemReader = new ItemReader() { + ItemReader itemReader = new AbstractItemReader() { public Object read() throws Exception { // Trigger a rollback throw new RuntimeException("Foo"); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java index da2711232..5dc162ff3 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java @@ -64,6 +64,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("1"); itemWriter.write("2"); + itemWriter.flush(); writerControl.replay(); dechunker.dechunk(chunk, stepContribution); writerControl.verify(); @@ -75,6 +76,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("1"); itemWriter.write("2"); writerControl.setThrowable(new Exception()); + itemWriter.flush(); writerControl.replay(); DechunkingResult result = dechunker.dechunk(chunk, stepContribution); writerControl.verify(); @@ -89,6 +91,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("1"); itemWriter.write("2"); writerControl.setThrowable(new NullPointerException()); + itemWriter.clear(); writerControl.replay(); try{ dechunker.dechunk(chunk, stepContribution); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/MockItemReader.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/MockItemReader.java index 731cb5e9e..809305564 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/MockItemReader.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/MockItemReader.java @@ -16,6 +16,8 @@ package org.springframework.batch.execution.step.support; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.exception.MarkFailedException; +import org.springframework.batch.item.exception.ResetFailedException; public class MockItemReader implements ItemReader { @@ -57,4 +59,10 @@ public class MockItemReader implements ItemReader { return null; } + public void mark() throws MarkFailedException { + } + + public void reset() throws ResetFailedException { + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepTests.java index 42d0bd305..e04114f7f 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepTests.java @@ -25,7 +25,7 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.execution.step.ItemOrientedStep; -import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.reader.ItemReaderAdapter; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.item.writer.ItemWriterAdapter; @@ -82,7 +82,7 @@ public class RepeatOperationsStepTests extends TestCase { }); repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); ItemOrientedStep configuration = new ItemOrientedStep(); - configuration.setItemReader(new ItemReader(){ + configuration.setItemReader(new AbstractItemReader(){ public Object read() throws Exception { throw new NullPointerException(); }}); @@ -122,7 +122,7 @@ public class RepeatOperationsStepTests extends TestCase { }); stepTemplate.setCompletionPolicy(new SimpleCompletionPolicy(1)); RepeatOperationsStep configuration = new RepeatOperationsStep(); - configuration.setItemReader(new ItemReader(){ + configuration.setItemReader(new AbstractItemReader(){ public Object read() throws Exception { return new Object(); }}); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java index 5e151e1e3..370cd1be7 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java @@ -34,7 +34,7 @@ import org.springframework.batch.execution.repository.dao.JobInstanceDao; import org.springframework.batch.execution.repository.dao.MapJobDao; import org.springframework.batch.execution.repository.dao.MapStepDao; import org.springframework.batch.execution.repository.dao.StepExecutionDao; -import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.reader.ItemReaderAdapter; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; @@ -80,7 +80,7 @@ public class StepExecutorInterruptionTests extends TestCase { final String stepName = (String) steps.get(0); JobExecution jobExecutionContext = new JobExecution(new JobInstance(new Long(0L), new JobParameters())); final StepExecution stepExecution = new StepExecution(stepName, jobExecutionContext); - step.setItemReader(new ItemReader() { + step.setItemReader(new AbstractItemReader() { public Object read() throws Exception { // do something non-trivial (and not Thread.sleep()) double foo = 1;