* Batch domain entity representing a step which is sequentially executed by a diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java index 92e4f5d08..34af99900 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/StepInstanceTests.java @@ -17,8 +17,6 @@ package org.springframework.batch.core.domain; import junit.framework.TestCase; -import org.springframework.batch.item.ExecutionAttributes; - /** * @author Dave Syer * diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/simple/SimpleJob.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/simple/SimpleJob.java index 94012c365..efc9bd524 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/simple/SimpleJob.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/simple/SimpleJob.java @@ -129,15 +129,23 @@ public class SimpleJob extends JobSupport { private boolean shouldStart(StepInstance stepInstance, Step step) { BatchStatus stepStatus; - //if the last execution is null, the step has never been executed. - if(stepInstance.getLastExecution() == null){ + // if the last execution is null, the step has never been executed. + if (stepInstance.getLastExecution() == null) { stepStatus = BatchStatus.STARTING; } - else{ + else { stepStatus = stepInstance.getLastExecution().getStatus(); } - - if (stepStatus== BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { + + if (stepStatus == BatchStatus.UNKNOWN) { + throw new BatchCriticalException( + "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; diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index c86df76cd..791d3d0bc 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -18,6 +18,8 @@ package org.springframework.batch.execution.step.simple; import java.util.Date; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.Step; import org.springframework.batch.core.domain.StepContribution; @@ -33,8 +35,9 @@ import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.io.Skippable; import org.springframework.batch.io.exception.BatchCriticalException; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.exception.ResetFailedException; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.ExitStatus; @@ -70,6 +73,8 @@ import org.springframework.util.Assert; */ public class SimpleStepExecutor { + private static final Log logger = LogFactory.getLog(SimpleStepExecutor.class); + private RepeatOperations chunkOperations = new RepeatTemplate(); private RepeatOperations stepOperations = new RepeatTemplate(); @@ -242,7 +247,22 @@ public class SimpleStepExecutor { synchronized (stepExecution) { stepExecution.rollback(); } - streamManager.rollback(transaction); + try { + streamManager.rollback(transaction); + } + catch (ResetFailedException e) { + // The original Throwable cause is in danger of + // being lost here, so we log the reset + // failure and re-throw with cause of the rollback. + logger.error("Encountered reset error on rollback: " + + "one of the streams may be in an inconsistent state, " + + "so this step should not proceed", e); + throw new ResetFailedException( + "Encountered reset error on rollback. " + + "Consult logs for the cause of the reet failure. " + + "The cause of the original rollback is incuded here.", + t); + } if (t instanceof RuntimeException) { throw (RuntimeException) t; } @@ -271,6 +291,10 @@ public class SimpleStepExecutor { updateStatus(stepExecution, BatchStatus.STOPPED); throw (StepInterruptedException) e.getCause(); } + else if (e instanceof ResetFailedException) { + updateStatus(stepExecution, BatchStatus.UNKNOWN); + throw (ResetFailedException) e; + } else { updateStatus(stepExecution, BatchStatus.FAILED); throw e; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java index 3e7bc577d..2b8aec7c2 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java @@ -18,8 +18,6 @@ package org.springframework.batch.execution.repository.dao; import java.util.Date; import java.util.List; -import java.util.Properties; - import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.Job; import org.springframework.batch.core.domain.JobExecution; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java index f303d9d0e..9fb053f2b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/JdbcStepDaoPrefixTests.java @@ -5,7 +5,6 @@ import java.util.List; import junit.framework.TestCase; import org.easymock.MockControl; -import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index c4bbe95c5..f10b8ca76 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -24,6 +24,7 @@ import java.util.Map; import junit.framework.TestCase; +import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.JobExecution; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; @@ -31,6 +32,7 @@ import org.springframework.batch.core.domain.JobSupport; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.core.domain.StepInterruptedException; import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.execution.repository.SimpleJobRepository; import org.springframework.batch.execution.repository.dao.MapJobDao; @@ -38,10 +40,11 @@ import org.springframework.batch.execution.repository.dao.MapStepDao; import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.ResetFailedException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.item.stream.SimpleStreamManager; @@ -55,6 +58,7 @@ import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.support.PropertiesConverter; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.transaction.TransactionStatus; public class SimpleStepExecutorTests extends TestCase { @@ -74,6 +78,8 @@ public class SimpleStepExecutorTests extends TestCase { private JobInstance jobInstance; + private ResourcelessTransactionManager transactionManager; + private ItemReader getReader(String[] args) { return new ListItemReader(Arrays.asList(args)); } @@ -98,8 +104,7 @@ public class SimpleStepExecutorTests extends TestCase { */ protected void setUp() throws Exception { super.setUp(); - ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager(); - + transactionManager = new ResourcelessTransactionManager(); stepConfiguration = new SimpleStep(); stepConfiguration.setTasklet(getTasklet(new String[] { "foo", "bar", "spam" })); stepConfiguration.setJobRepository(new JobRepositorySupport()); @@ -116,7 +121,7 @@ public class SimpleStepExecutorTests extends TestCase { jobInstance = new JobInstance(new Long(0), new JobParameters()); jobInstance.setJob(new JobSupport("FOO")); - + SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); streamManager.setUseClassNameAsPrefix(false); stepExecutor.setStreamManager(streamManager); @@ -312,9 +317,9 @@ public class SimpleStepExecutorTests extends TestCase { stepConfiguration.setSaveExecutionAttributes(true); JobExecution jobExecutionContext = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - stepExecution.setExecutionAttributes( - new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); + + stepExecution + .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); step.setLastExecution(stepExecution); stepExecutor.execute(stepExecution); @@ -324,8 +329,9 @@ public class SimpleStepExecutorTests extends TestCase { } /* - * Test that a job that is being restarted, but has saveExecutionAttributes set to - * false, doesn't have restore or getExecutionAttributes called on it. + * Test that a job that is being restarted, but has saveExecutionAttributes + * set to false, doesn't have restore or getExecutionAttributes called on + * it. */ public void testNoSaveExecutionAttributesRestartableJob() { StepInstance step = new StepInstance(new Long(1)); @@ -348,8 +354,9 @@ public class SimpleStepExecutorTests extends TestCase { } /* - * Even though the job is restarted, and saveExecutionAttributes is true, nothing - * will be restored because the Tasklet does not implement Restartable. + * Even though the job is restarted, and saveExecutionAttributes is true, + * nothing will be restored because the Tasklet does not implement + * Restartable. */ public void testRestartJobOnNonRestartableTasklet() throws Exception { StepInstance step = new StepInstance(new Long(1)); @@ -479,41 +486,88 @@ public class SimpleStepExecutorTests extends TestCase { } - /* - * StepExecutor will never pass StepInterruptedException to the - * exceptionClassifier. This may or may not stay the same, so the test will - * remain commented out for reference purposes. - */ - /* - * public void testExitCodeInterruptedClassification(){ - * - * StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy(){ - * - * public void checkInterrupted(RepeatContext context) throws - * StepInterruptedException { throw new StepInterruptedException(""); } }; - * - * stepExecutor.setInterruptionPolicy(interruptionPolicy); - * - * Tasklet tasklet = new Tasklet(){ - * - * public ExitStatus execute() throws Exception { int counter = 0; - * counter++; - * - * if(counter == 1){ throw new StepInterruptedException(""); } - * - * return ExitStatus.CONTINUABLE; } }; - * - * StepInstance step = new StepInstance(new Long(1)); - * stepConfiguration.setTasklet(tasklet); JobExecutionContext - * jobExecutionContext = new JobExecutionContext(new - * SimpleJobIdentifier("FOO"), new JobInstance(new Long(3))); StepExecution - * stepExecution = new StepExecution(step, jobExecutionContext); - * - * try{ stepExecutor.process(stepConfiguration, stepExecution); } - * catch(Exception ex){ - * assertEquals(ExitCodeExceptionClassifier.STEP_INTERRUPTED, - * step.getStepExecution().getExitCode() ); - * assertEquals(step.getStepExecution().getExitDescription(), - * "java.lang.RuntimeException"); } } - */ + public void testStatusForInterruptedException() { + + StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { + + public void checkInterrupted(RepeatContext context) throws StepInterruptedException { + throw new StepInterruptedException(""); + } + }; + + stepExecutor.setInterruptionPolicy(interruptionPolicy); + + Tasklet tasklet = new Tasklet() { + + public ExitStatus execute() throws Exception { + int counter = 0; + counter++; + + if (counter == 1) { + throw new StepInterruptedException(""); + } + + return ExitStatus.CONTINUABLE; + } + }; + stepExecutor.setTasklet(tasklet); + + StepInstance step = new StepInstance(new Long(1)); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + stepExecution + .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); + step.setLastExecution(stepExecution); + + try { + stepExecutor.execute(stepExecution); + fail("Expected StepInterruptedException"); + } + catch (StepInterruptedException ex) { + assertEquals(BatchStatus.STOPPED, stepExecution.getStatus()); + String msg = stepExecution.getExitStatus().getExitDescription(); + assertTrue("Message does not contain StepInterruptedException: " + msg, msg + .contains("StepInterruptedException")); + } + } + + public void testStatusForResetFailedException() throws Exception { + + Tasklet tasklet = new Tasklet() { + public ExitStatus execute() throws Exception { + // Trigger a rollback + throw new RuntimeException("Foo"); + } + }; + stepExecutor.setTasklet(tasklet); + stepExecutor.setStreamManager(new SimpleStreamManager(transactionManager) { + public void rollback(TransactionStatus status) { + super.rollback(status); + // Simulate failure on rollback when stream resets + throw new ResetFailedException("Bar"); + } + }); + + StepInstance step = new StepInstance(new Long(1)); + JobExecution jobExecutionContext = jobInstance.createJobExecution(); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + stepExecution + .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); + step.setLastExecution(stepExecution); + + try { + stepExecutor.execute(stepExecution); + fail("Expected ResetFailedException"); + } + catch (ResetFailedException ex) { + assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus()); + String msg = stepExecution.getExitStatus().getExitDescription(); + assertTrue("Message does not contain ResetFailedException: " + msg, msg.contains("ResetFailedException")); + // The original rollback was caused by this one: + assertEquals("Foo", ex.getCause().getMessage()); + } + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java index 672ab668f..db8904c1b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java @@ -27,7 +27,7 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.KeyedItemReader; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.context.RepeatContextSupport; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index 51531a976..1665a030b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -25,7 +25,7 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.io.file.separator.LineReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; /** *
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java index 1e4c4bc9b..0b5fb8110 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java @@ -34,7 +34,7 @@ import org.springframework.batch.io.support.AbstractTransactionalIoSource; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.writer.ItemTransformer; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java index 3297d5d28..7e1474db8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java @@ -31,7 +31,7 @@ import org.springframework.batch.io.file.transform.AbstractLineTokenizer; import org.springframework.batch.io.file.transform.DelimitedLineTokenizer; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java index 9ee964182..689882b31 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/separator/ResourceLineReader.java @@ -28,6 +28,8 @@ import java.util.Iterator; import org.springframework.batch.io.exception.BatchEnvironmentException; 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.stream.ItemStreamAdapter; import org.springframework.core.io.Resource; import org.springframework.util.Assert; @@ -212,9 +214,9 @@ public class ResourceLineReader extends ItemStreamAdapter implements LineReader, * * @see #reset() * - * @throws BatchEnvironmentException if the mark could not be set. + * @throws MarkFailedException if the mark could not be set. */ - public synchronized void mark() { + public synchronized void mark() throws MarkFailedException { getState().mark(); } @@ -223,10 +225,10 @@ public class ResourceLineReader extends ItemStreamAdapter implements LineReader, * * @see #mark() * - * @throws BatchEnvironmentException if the reset is unsuccessful, e.g. if + * @throws ResetFailedException if the reset is unsuccessful, e.g. if * the read-ahead limit was breached. */ - public synchronized void reset() { + public synchronized void reset() throws ResetFailedException { getState().reset(); } @@ -318,13 +320,13 @@ public class ResourceLineReader extends ItemStreamAdapter implements LineReader, /** * Mark the underlying reader and set the line counters. */ - public void mark() { + public void mark() throws MarkFailedException { try { reader.mark(READ_AHEAD_LIMIT); markedLineCount = currentLineCount; } catch (IOException e) { - throw new BatchEnvironmentException("Could not mark reader", e); + throw new MarkFailedException("Could not mark reader", e); } } @@ -332,7 +334,7 @@ public class ResourceLineReader extends ItemStreamAdapter implements LineReader, * Reset the reader and line counters to the last marked position if * possible. */ - public void reset() { + public void reset() throws ResetFailedException { if (markedLineCount < 0) { return; @@ -342,7 +344,7 @@ public class ResourceLineReader extends ItemStreamAdapter implements LineReader, currentLineCount = markedLineCount; } catch (IOException e) { - throw new BatchEnvironmentException("Could not reset reader", e); + throw new ResetFailedException("Could not reset reader", e); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index fb09dd066..a65646d77 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -18,7 +18,7 @@ import org.springframework.batch.io.xml.stax.NoStartEndDocumentStreamWriter; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.dao.DataAccessResourceFailureException; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java index 0a984d95f..e366d3dc9 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java @@ -16,6 +16,10 @@ package org.springframework.batch.item; +import org.springframework.batch.item.exception.MarkFailedException; +import org.springframework.batch.item.exception.ResetFailedException; +import org.springframework.batch.item.exception.StreamException; + /** *
* Marker interface defining a contract for periodically storing state and @@ -81,8 +85,11 @@ public interface ItemStream extends ExecutionAttributesProvider { * state from the current thread is saved. * * @throws UnsupportedOperationException if the operation is not supported + * @throws MarkFailedException if there is a problem with the mark. If a + * mark fails inside a transaction, it would be worrying, but not normally + * fatal. */ - void mark(); + void mark() throws MarkFailedException; /** * Reset the stream to the last mark. After a reset the stream state will be @@ -93,6 +100,10 @@ public interface ItemStream extends ExecutionAttributesProvider { * state from the current thread is reset. * * @throws UnsupportedOperationException if the operation is not supported + * @throws ResetFailedException if there is a problem with the reset. If a + * reset fails inside a transaction, it would normally be fatal, and would + * leave the stream in an inconsistent state. So while this is an unchecked + * exception, it may be important for a client to catch it explicitly. */ - void reset(); + void reset() throws ResetFailedException; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/StreamException.java similarity index 96% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/StreamException.java index a109493a5..b4d9e7417 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/StreamException.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.batch.item; +package org.springframework.batch.item.exception; import org.springframework.batch.io.exception.BatchCriticalException; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/UnexpectedInputException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/UnexpectedInputException.java index 3bdbcad9a..04d886706 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/UnexpectedInputException.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/exception/UnexpectedInputException.java @@ -16,6 +16,7 @@ package org.springframework.batch.item.exception; + /** * Used to signal an unexpected end of an input or message stream. This is an * abnormal condition, not just the end of the data - e.g. if a resource becomes @@ -23,7 +24,7 @@ package org.springframework.batch.item.exception; * * @author Dave Syer */ -public class UnexpectedInputException extends RuntimeException { +public class UnexpectedInputException extends StreamException { /** * Generated serial UID. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java index 13c9818cb..f8b723846 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReader.java @@ -17,7 +17,7 @@ package org.springframework.batch.item.reader; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; /** * Base class for {@link ItemReader} implementations. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java index 92f535a14..b307c69cd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java @@ -20,7 +20,7 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java index 42ab28106..f8102c571 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/ItemReaderAdapter.java @@ -17,7 +17,7 @@ package org.springframework.batch.item.reader; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.support.AbstractMethodInvokingDelegator; /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java index 67bbb4da2..71ed3d4e8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java @@ -17,7 +17,7 @@ package org.springframework.batch.item.stream; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; /** * @author Dave Syer diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java index 0f6220183..a9d364438 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java @@ -25,7 +25,7 @@ import java.util.Map.Entry; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.DefaultTransactionDefinition; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java index 5e55548df..eda1dd406 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/StreamManager.java @@ -17,7 +17,7 @@ package org.springframework.batch.item.stream; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.transaction.TransactionStatus; /** diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java index 9bbb56c75..7b9f639fb 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java @@ -25,7 +25,7 @@ import org.springframework.batch.io.file.mapping.FieldSet; import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java index 102c99685..510d55f7e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/stream/SimpleStreamManagerTests.java @@ -21,7 +21,7 @@ import java.util.List; import junit.framework.TestCase; import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.StreamException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.support.PropertiesConverter; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.transaction.TransactionException;