diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java index f4967e67f..dccb423b1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java @@ -38,7 +38,7 @@ import org.springframework.batch.item.ExecutionAttributes; * these Id's are set, an execution can be persisted. If the object is in a * transient state (i.e. it has no id of it's own) then an ID will be created * for that specific execution, and then stored ('saved'). (NOTE: The - * relationship between a Job/Step and Job/StepExecutions is 1:N) If an ID does + * relationship between a Job/Step and Job/StepExecutions is 1:N.) If an ID does * exist, then the execution will be stored ('updated'). * * 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 ccf5f7982..e638a11b1 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 @@ -60,16 +60,20 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.util.Assert; /** - * Simple implementation of executing the step as a set of chunks, each chunk surrounded by a transaction. The structure - * is therefore that of two nested loops, with transaction boundary around the whole inner loop. The outer loop is - * controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), and the inner loop by the chunk - * operations ({@link #setChunkOperations(RepeatOperations)}). The inner loop should always be executed in a single - * thread, so the chunk operations should not do any concurrent execution. N.B. usually that means that the chunk - * operations should be a {@link RepeatTemplate} (which is the default).
+ * Simple implementation of executing the step as a set of chunks, each chunk + * surrounded by a transaction. The structure is therefore that of two nested + * loops, with transaction boundary around the whole inner loop. The outer loop + * is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), + * and the inner loop by the chunk operations ({@link #setChunkOperations(RepeatOperations)}). + * The inner loop should always be executed in a single thread, so the chunk + * operations should not do any concurrent execution. N.B. usually that means + * that the chunk operations should be a {@link RepeatTemplate} (which is the + * default).
* - * Clients can use interceptors in the step operations to intercept or listen to the iteration on a step-wide basis, for - * instance to get a callback when the step is complete. Those that want callbacks at the level of an individual tasks, - * can specify interceptors for the chunk operations. + * Clients can use interceptors in the step operations to intercept or listen to + * the iteration on a step-wide basis, for instance to get a callback when the + * step is complete. Those that want callbacks at the level of an individual + * tasks, can specify interceptors for the chunk operations. * * @author Dave Syer * @author Lucas Ward @@ -114,18 +118,22 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Public setter for the {@link StreamManager}. This will be used to create the {@link StepContext}, and hence any - * component that is a {@link ItemStream} and in step scope will be registered with the service. The - * {@link StepContext} is then a source of aggregate statistics for the step. + * Public setter for the {@link StreamManager}. This will be used to create + * the {@link StepContext}, and hence any component that is a + * {@link ItemStream} and in step scope will be registered with the service. + * The {@link StepContext} is then a source of aggregate statistics for the + * step. * - * @param streamManager the {@link StreamManager} to set. Default is a {@link SimpleStreamManager}. + * @param streamManager the {@link StreamManager} to set. Default is a + * {@link SimpleStreamManager}. */ public void setStreamManager(StreamManager streamManager) { this.streamManager = streamManager; } /** - * Injected strategy for storage and retrieval of persistent step information. Mandatory property. + * Injected strategy for storage and retrieval of persistent step + * information. Mandatory property. * * @param jobRepository */ @@ -134,8 +142,9 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * The {@link RepeatOperations} to use for the outer loop of the batch processing. Should be set up by the caller - * through a factory. Defaults to a plain {@link RepeatTemplate}. + * The {@link RepeatOperations} to use for the outer loop of the batch + * processing. Should be set up by the caller through a factory. Defaults to + * a plain {@link RepeatTemplate}. * * @param stepOperations a {@link RepeatOperations} instance. */ @@ -144,8 +153,9 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * The {@link RepeatOperations} to use for the inner loop of the batch processing. Should be set up by the caller - * through a factory. Defaults to a plain {@link RepeatTemplate}. + * The {@link RepeatOperations} to use for the inner loop of the batch + * processing. Should be set up by the caller through a factory. Defaults to + * a plain {@link RepeatTemplate}. * * @param chunkOperations a {@link RepeatOperations} instance. */ @@ -154,8 +164,9 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Setter for the {@link StepInterruptionPolicy}. The policy is used to check whether an external request has been - * made to interrupt the job execution. + * Setter for the {@link StepInterruptionPolicy}. The policy is used to + * check whether an external request has been made to interrupt the job + * execution. * * @param interruptionPolicy a {@link StepInterruptionPolicy} */ @@ -164,8 +175,8 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Setter for the {@link ExitStatusExceptionClassifier} that will be used to classify any exception that causes a job - * to fail. + * Setter for the {@link ExitStatusExceptionClassifier} that will be used to + * classify any exception that causes a job to fail. * * @param exceptionClassifier */ @@ -186,7 +197,7 @@ public class SimpleStepExecutor implements InitializingBean { public void setItemWriter(ItemWriter itemWriter) { this.itemWriter = itemWriter; } - + /** * Setter for injecting optional recovery handler. * @@ -223,7 +234,7 @@ public class SimpleStepExecutor implements InitializingBean { if (retryPolicy != null) { Assert.state(itemReader instanceof KeyedItemReader, - "ItemReader must be instance of KeyedItemReader to use the retry policy"); + "ItemReader must be instance of KeyedItemReader to use the retry policy"); retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); retryCallback.setRecoverer(itemRecoverer); } @@ -231,7 +242,8 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Apply the configuration by inspecting it to see if it has any relevant policy information. + * Apply the configuration by inspecting it to see if it has any relevant + * policy information. * * @param step a step */ @@ -261,14 +273,18 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Process the step and update its context so that progress can be monitored by the caller. The step is broken down - * into chunks, each one executing in 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 JobExecutionContext}.
+ * Process the step and update its context so that progress can be monitored + * by the caller. The step is broken down into chunks, each one executing in + * 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 JobExecutionContext}.
* * @throws JobInterruptedException if the step or a chunk is interrupted - * @throws RuntimeException if there is an exception during a chunk execution + * @throws RuntimeException if there is an exception during a chunk + * execution * @see StepExecutor#execute(StepExecution) */ public void execute(final StepExecution stepExecution) throws BatchCriticalException, JobInterruptedException { @@ -342,31 +358,36 @@ public class SimpleStepExecutor implements InitializingBean { streamManager.commit(transaction); - } catch (Throwable t) { + } + catch (Throwable t) { /* - * Any exception thrown within the transaction template will automatically cause the transaction - * to rollback. We need to include exceptions during an attempted commit (e.g. Hibernate flush) - * so this catch block comes outside the transaction. + * Any exception thrown within the transaction template + * will automatically cause the transaction to rollback. + * We need to include exceptions during an attempted + * commit (e.g. Hibernate flush) so this catch block + * comes outside the transaction. */ synchronized (stepExecution) { stepExecution.rollback(); } try { streamManager.rollback(transaction); - } catch (ResetFailedException e) { + } + 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); + + "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); + + "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; - } else { + } + else { throw new RuntimeException(t); } } @@ -382,27 +403,32 @@ public class SimpleStepExecutor implements InitializingBean { }); updateStatus(stepExecution, BatchStatus.COMPLETED); - } catch (RuntimeException e) { + } + catch (RuntimeException e) { // classify exception so an exit code can be stored. status = exceptionClassifier.classifyForExitCode(e); if (e.getCause() instanceof JobInterruptedException) { updateStatus(stepExecution, BatchStatus.STOPPED); throw (JobInterruptedException) e.getCause(); - } else if (e instanceof ResetFailedException) { + } + else if (e instanceof ResetFailedException) { updateStatus(stepExecution, BatchStatus.UNKNOWN); throw (ResetFailedException) e; - } else { + } + else { updateStatus(stepExecution, BatchStatus.FAILED); throw e; } - } finally { + } + finally { stepExecution.setExitStatus(status); stepExecution.setEndTime(new Date(System.currentTimeMillis())); try { jobRepository.saveOrUpdate(stepExecution); - } finally { + } + finally { // clear any registered synchronizations StepSynchronizationManager.close(); } @@ -411,11 +437,13 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Execute a bunch of identical business logic operations all within a transaction. The transaction is - * programmatically started and stopped outside this method, so subclasses that override do not need to create a + * Execute a bunch of identical business logic operations all within a + * transaction. The transaction is programmatically started and stopped + * outside this method, so subclasses that override do not need to create a * transaction. * - * @param step the current step containing the {@link Tasklet} with the business logic. + * @param step the current step containing the {@link Tasklet} with the + * business logic. * @return true if there is more data to process. */ ExitStatus processChunk(final Step step, final StepContribution contribution) { @@ -437,10 +465,12 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Execute the business logic, delegating to the given {@link Tasklet}. Subclasses could extend the behaviour as - * long as they always return the value of this method call in their superclass.
+ * Execute the business logic, delegating to the given {@link Tasklet}. + * Subclasses could extend the behaviour as long as they always return the + * value of this method call in their superclass.
* - * If there is an exception and the {@link Tasklet} implements {@link Skippable} then the skip method is called. + * If there is an exception and the {@link Tasklet} implements + * {@link Skippable} then the skip method is called. * * @param tasklet the unit of business logic to execute * @param contribution the current step @@ -454,7 +484,8 @@ public class SimpleStepExecutor implements InitializingBean { exitStatus = execute(); - } catch (Exception e) { + } + catch (Exception e) { skip(); // Rethrow so that outer transaction is rolled back properly throw e; @@ -465,13 +496,16 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Read from the {@link ItemReader} and process (if not null) with the {@link ItemWriter}. The call to - * {@link ItemWriter} is wrapped in a stateful retry, if a {@link RetryPolicy} is provided. The - * {@link ItemRecoverer} is used (if provided) in the case of an exception to apply alternate processing to the - * item. If the stateful retry is in place then the recovery will happen in the next transaction automatically, - * otherwise it might be necessary for clients to make the recover method transactional with appropriate propagation - * behaviour (probably REQUIRES_NEW because the call will happen in the context of a transaction that is about to - * rollback). + * Read from the {@link ItemReader} and process (if not null) with the + * {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a + * stateful retry, if a {@link RetryPolicy} is provided. The + * {@link ItemRecoverer} is used (if provided) in the case of an exception + * to apply alternate processing to the item. If the stateful retry is in + * place then the recovery will happen in the next transaction + * automatically, otherwise it might be necessary for clients to make the + * recover method transactional with appropriate propagation behaviour + * (probably REQUIRES_NEW because the call will happen in the context of a + * transaction that is about to rollback). * * @see org.springframework.batch.core.tasklet.Tasklet#execute() */ @@ -484,7 +518,8 @@ public class SimpleStepExecutor implements InitializingBean { } try { itemWriter.write(item); - } catch (Exception e) { + } + catch (Exception e) { if (itemRecoverer != null) { itemRecoverer.recover(item, e); } @@ -500,9 +535,10 @@ public class SimpleStepExecutor implements InitializingBean { } /** - * Mark the current item as skipped if possible. If there is a retry policy in action there is no need to take any - * action now because it will be covered by the retry in the next transaction. Otherwise if the reader and / or - * writer are {@link Skippable} then delegate to them in that order. + * Mark the current item as skipped if possible. If there is a retry policy + * in action there is no need to take any action now because it will be + * covered by the retry in the next transaction. Otherwise if the reader and / + * or writer are {@link Skippable} then delegate to them in that order. * * @see org.springframework.batch.io.Skippable#skip() */ diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java deleted file mode 100644 index c797a1f2d..000000000 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java +++ /dev/null @@ -1,566 +0,0 @@ -/* - * 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.execution.step.simple; - -import java.util.ArrayList; -import java.util.Arrays; - -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; -import org.springframework.batch.core.domain.JobSupport; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepInstance; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.execution.scope.StepScope; -import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.item.ExecutionAttributes; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -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; -import org.springframework.batch.item.writer.AbstractItemWriter; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.interceptor.RepeatListenerAdapter; -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; - -/** - * - * - * @author Lucas Ward - * - */ -public class ChunkedStepTests extends TestCase { - - ArrayList processed = new ArrayList(); - - ItemWriter processor = new AbstractItemWriter() { - public void write(Object data) throws Exception { - processed.add((String) data); - } - }; - - private ChunkedStep chunkedStep; - - private JobInstance jobInstance; - - private ResourcelessTransactionManager transactionManager; - - private StepInstance stepInstance; - private JobExecution jobExecutionContext; - private StepExecution stepExecution; - - private ItemReader getReader(String[] args) { - return new ListItemReader(Arrays.asList(args)); - } - - - - private ChunkedStep getStep(String[] strings) throws Exception { - ChunkedStep step = new ChunkedStep(); - step.setItemWriter(processor); - step.setItemReader(getReader(strings)); - step.setJobRepository(new JobRepositorySupport()); - step.setStreamManager(new SimpleStreamManager(transactionManager)); - step.afterPropertiesSet(); - return step; - } - - protected void setUp() throws Exception { - transactionManager = new ResourcelessTransactionManager(); - chunkedStep = getStep(new String[] { "foo", "bar", "spam" }); - - // Only process one item: - chunkedStep.setChunkSize(1); - - jobInstance = new JobInstance(new Long(0), new JobParameters()); - jobInstance.setJob(new JobSupport("FOO")); - - SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); - streamManager.setUseClassNameAsPrefix(false); - chunkedStep.setStreamManager(streamManager); - chunkedStep.setJobRepository(new JobRepositorySupport()); - - stepInstance = new StepInstance(new Long(9)); - jobExecutionContext = new JobExecution(jobInstance); - stepExecution = new StepExecution(stepInstance, jobExecutionContext); - } - - public void testStepExecutor() throws Exception { - - chunkedStep.execute(stepExecution); - assertEquals(3, processed.size()); - assertEquals(stepExecution.getStatus(), BatchStatus.COMPLETED); - } - - public void testStepContextInitialized() throws Exception { - - final StepInstance step = new StepInstance(new Long(1)); - final JobExecution jobExecution = new JobExecution(jobInstance); - final StepExecution stepExecution = new StepExecution(step, jobExecution); - - chunkedStep.setChunker(new ItemChunker(new ItemReader() { - int counter = 0; - public Object read() throws Exception { - assertEquals(step, stepExecution.getStep()); - assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); - if(counter++ < 2){ - return "foo"; - } - else{ - return null; - } - } - })); - - chunkedStep.execute(stepExecution); - assertEquals(2, processed.size()); - } - - public void testStepContextInitializedBeforeTasklet() throws Exception { - - RepeatTemplate template = new RepeatTemplate(); - - // Only process one chunk: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - chunkedStep.setStepOperations(template); - - final StepInstance step = new StepInstance(new Long(1)); - final JobExecution jobExecution = new JobExecution(jobInstance); - jobExecution.setId(new Long(1)); - final StepExecution stepExecution = new StepExecution(step, jobExecution); - - template.setListener(new RepeatListenerAdapter() { - public void open(RepeatContext context) { - assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); - assertEquals(stepExecution, StepSynchronizationManager.getContext().getStepExecution()); - // StepScope can obtain id information.... - assertNotNull(StepSynchronizationManager.getContext().getAttribute(StepScope.ID_KEY)); - } - }); - - chunkedStep.execute(stepExecution); - assertEquals(1, processed.size()); - - } - - public void testRepository() throws Exception { - - MockControl repoControl = MockControl.createControl(JobRepository.class); - JobRepository repository = (JobRepository)repoControl.getMock(); - chunkedStep.setJobRepository(repository); - -// StepInstance step = new StepInstance(new Long(1)); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repoControl.replay(); - chunkedStep.execute(stepExecution); - assertEquals(3, processed.size()); - repoControl.verify(); - assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); - } - - //ReadFailures (meaning an item couldn't be skipped) should cause the job to - //fail. - public void testReadFailure() { - - ItemReader itemReader = new ItemReader() { - int counter = 0; - public Object read() throws Exception { - - counter++; - - if (counter > 1) { - throw new RuntimeException(); - } - - return "foo"; - } - - }; - - chunkedStep.setChunker(new ItemChunker(itemReader)); - - try { - chunkedStep.execute(stepExecution); - fail(); - } - catch (Exception ex) { - assertEquals( 1, processed.size()); - assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - } - - } - - public void testWriterFailure(){ - - ItemWriter itemWriter = new ItemWriter(){ - public void write(Object item) throws Exception { - throw new RuntimeException(); - } - }; - - chunkedStep.setDechunker(new ItemDechunker(itemWriter)); - - try{ - chunkedStep.execute(stepExecution); - fail(); - } - catch(Exception ex){ - //it should rollback three times since that's default behavior for a retry template. - assertEquals(new Integer(3), stepExecution.getRollbackCount()); - } - } - - public void testExitCodeDefaultClassification() throws Exception { - - ItemReader itemReader = new ItemReader() { - int counter = 0; - public Object read() throws Exception { - counter++; - - if (counter == 1) { - throw new RuntimeException(); - } - - return ExitStatus.CONTINUABLE; - } - - }; - - chunkedStep.setItemReader(itemReader); - - try { - chunkedStep.execute(stepExecution); - } - catch (Exception ex) { - ExitStatus status = stepExecution.getExitStatus(); - assertEquals("FATAL_EXCEPTION", status.getExitCode()); - assertFalse(status.isContinuable()); - } - } - - /* - * make sure a job that has never been executed before, but does have - * saveExecutionAttributes = true, doesn't have restoreFrom called on it. - */ - // I can't get this test to pass, I think there is something wrong with the code I - // pulled from SimpleStepExecutor -// public void testNonRestartedJob() throws Exception { -// MockRestartableItemReader reader = new MockRestartableItemReader(); -// chunkedStep.setItemReader(reader); -// chunkedStep.setSaveExecutionAttributes(true); -// -// chunkedStep.execute(stepExecution); -// -// assertFalse(reader.isRestoreFromCalled()); -// assertTrue(reader.isGetExecutionAttributesCalled()); -// } -// -// /* -// * make sure a job that has been executed before, and is therefore being -// * restarted, is restored. -// */ -// public void testRestartedJob() throws Exception { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// MockRestartableTasklet tasklet = new MockRestartableTasklet(); -// chunkedStep.setItemReader(tasklet); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// stepExecution -// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); -// chunkedStep.execute(stepExecution); -// -// assertTrue(tasklet.isRestoreFromCalled()); -// assertTrue(tasklet.isRestoreFromCalledWithSomeContext()); -// assertTrue(tasklet.isGetExecutionAttributesCalled()); -// } -// -// /* -// * 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)); -// step.setStepExecutionCount(1); -// MockRestartableTasklet tasklet = new MockRestartableTasklet(); -// stepConfiguration.setItemReader(tasklet); -// stepConfiguration.setSaveExecutionAttributes(false); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// try { -// chunkedStep.execute(stepExecution); -// } -// catch (Throwable t) { -// fail(); -// } -// -// assertFalse(tasklet.isRestoreFromCalled()); -// assertFalse(tasklet.isGetExecutionAttributesCalled()); -// } -// -// /* -// * 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)); -// step.setStepExecutionCount(1); -// stepConfiguration.setItemReader(new ItemReader() { -// public Object read() throws Exception { -// return ExitStatus.FINISHED; -// } -// }); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecution = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// -// chunkedStep.execute(stepExecution); -// } -// -//// public void testApplyConfigurationWithExceptionHandler() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// stepConfiguration.setExceptionHandler(new DefaultExceptionHandler()); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(1, list.size()); -//// } -//// -//// public void testApplyConfigurationWithZeroSkipLimit() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// stepConfiguration.setSkipLimit(0); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(0, list.size()); -//// } -//// -//// public void testApplyConfigurationWithNonZeroSkipLimit() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// stepConfiguration.setSkipLimit(1); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(1, list.size()); -//// } -// -// public void testStreamManager() throws Exception { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// stepConfiguration.setItemReader(new ItemReader() { -// public Object read() throws Exception { -// return ExitStatus.FINISHED; -// } -// }); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecution = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// -// assertEquals(false, stepExecution.getExecutionAttributes().containsKey("foo")); -// -// final Map map = new HashMap(); -// chunkedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { -// public ExecutionAttributes getExecutionAttributes(Object key) { -// // TODO Auto-generated method stub -// return new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar")); -// } -// }); -// -// chunkedStep.execute(stepExecution); -// -// // At least once in that process the statistics service was asked for -// // statistics... -// assertEquals("bar", stepExecution.getExecutionAttributes().getString("foo")); -// // ...but nothing was registered because nothing with step scoped. -// assertEquals(0, map.size()); -// } -// -// public void testStatusForInterruptedException() { -// -// StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { -// -// public void checkInterrupted(RepeatContext context) throws JobInterruptedException { -// throw new JobInterruptedException(""); -// } -// }; -// -// chunkedStep.setInterruptionPolicy(interruptionPolicy); -// -// ItemReader itemReader = new ItemReader() { -// -// public Object read() throws Exception { -// int counter = 0; -// counter++; -// -// if (counter == 1) { -// throw new RuntimeException(); -// } -// -// return ExitStatus.CONTINUABLE; -// } -// -// }; -// -// chunkedStep.setItemReader(itemReader); -// -// 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 { -// chunkedStep.execute(stepExecution); -// fail("Expected StepInterruptedException"); -// } -// catch (JobInterruptedException ex) { -// assertEquals(BatchStatus.STOPPED, stepExecution.getStatus()); -// String msg = stepExecution.getExitStatus().getExitDescription(); -// assertTrue("Message does not contain JobInterruptedException: " + msg, msg -// .contains("JobInterruptedException")); -// } -// } -// -// public void testStatusForResetFailedException() throws Exception { -// -// ItemReader itemReader = new ItemReader() { -// public Object read() throws Exception { -// // Trigger a rollback -// throw new RuntimeException("Foo"); -// } -// }; -// chunkedStep.setItemReader(itemReader); -// chunkedStep.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 { -// chunkedStep.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()); -// } -// } -// - private class MockRestartableItemReader extends ItemStreamAdapter implements ItemReader { - - private boolean getExecutionAttributesCalled = false; - - private boolean restoreFromCalled = false; - - private boolean restoreFromCalledWithSomeContext = false; - - private int counter = 0; - - public Object read() throws Exception { - StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); - counter++; - if(counter > 4){ - return "item"; - } - else{ - return null; - } - } - - public boolean isRestoreFromCalledWithSomeContext() { - return restoreFromCalledWithSomeContext; - } - - public ExecutionAttributes getExecutionAttributes() { - getExecutionAttributesCalled = true; - return new ExecutionAttributes(PropertiesConverter.stringToProperties("spam=bucket")); - } - - public void restoreFrom(ExecutionAttributes data) { - restoreFromCalled = true; - restoreFromCalledWithSomeContext = data.getProperties().size() > 0; - } - - public boolean isGetExecutionAttributesCalled() { - return getExecutionAttributesCalled; - } - - public boolean isRestoreFromCalled() { - return restoreFromCalled; - } - - public void open() throws StreamException { - } - - public void close() throws StreamException { - } - - } - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java index d01b76b6d..5bd701cea 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java @@ -135,7 +135,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In ((AbstractLineTokenizer) tokenizer).setNames(names); } } - + mark(); } @@ -180,9 +180,9 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In } /** - * This method initialises the reader for Restart. It opens the input - * file and position the buffer reader according to information provided by - * the restart data + * This method initialises the reader for Restart. It opens the input file + * and position the buffer reader according to information provided by the + * restart data * * @param data {@link ExecutionAttributes} information */ @@ -209,8 +209,8 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In } /** - * This method returns the execution attributes for the reader. It returns the - * current Line Count which can be used to reinitialise the batch job in + * This method returns the execution attributes for the reader. It returns + * the current Line Count which can be used to reinitialise the batch job in * case of restart. */ public ExecutionAttributes getExecutionAttributes() { @@ -262,7 +262,8 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In } /** - * @return next line to be tokenized and mapped (possibly skips multiple lines). + * @return next line to be tokenized and mapped (possibly skips multiple + * lines). */ protected String readLine() { String line = nextLine(); @@ -280,6 +281,9 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In try { return (String) getReader().read(); } + catch (StreamException e) { + throw e; + } catch (Exception e) { throw new IllegalStateException(e); } @@ -290,7 +294,7 @@ public class FlatFileItemReader implements ItemReader, Skippable, ItemStream, In */ protected LineReader getReader() { if (reader == null) { - open(); + throw new StreamException("ItemStream must be open before it can be read."); // reader is now not null, or else an exception is thrown } return reader; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java index eef7ec0a4..d4cede3ef 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java @@ -19,6 +19,7 @@ import org.springframework.batch.io.xml.stax.TransactionalEventReader; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; @@ -68,7 +69,7 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade */ public Object read() { if (!initialized) { - open(); + throw new StreamException("ItemStream must be open before it can be read."); } Object item = null; @@ -192,13 +193,11 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade * available records. */ public void restoreFrom(ExecutionAttributes data) { - Assert.state(!initialized); + if (data == null || data.getProperties() == null || !data.containsKey(READ_COUNT_STATISTICS_NAME)) { return; } - open(); - long restoredRecordCount = data.getLong(READ_COUNT_STATISTICS_NAME); int REASONABLE_ADHOC_COMMIT_FREQUENCY = 100; while (currentRecordCount <= restoredRecordCount) { @@ -206,11 +205,14 @@ public class StaxEventItemReader extends AbstractItemReader implements ItemReade if (currentRecordCount % REASONABLE_ADHOC_COMMIT_FREQUENCY == 0) { txReader.onCommit(); // reset the history buffer } - Assert.state(fragmentReader.hasNext(), "restore point must be before end of input"); + if (!fragmentReader.hasNext()) { + throw new StreamException("Restore point must be before end of input"); + } fragmentReader.next(); moveCursorToNextFragment(fragmentReader); } mark(); // reset the history buffer + } /** 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 a65646d77..f7e95c98f 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 @@ -342,7 +342,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing public void write(Object output) { if (!initialized) { - open(); + throw new StreamException("ItemStream must be open before it can be used."); } currentRecordCount++; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java index 5b2ee8012..82cc64bfe 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java @@ -157,6 +157,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { } catch (StreamException e) { assertTrue("Message does not contain open: "+e.getMessage(), e.getMessage().contains("open")); } + reader.open(); assertEquals("[FlatFileInputTemplate-TestData]", reader.read().toString()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java index dd1c16c56..31e3dcaa9 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java @@ -29,6 +29,7 @@ import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.separator.DefaultRecordSeparatorPolicy; import org.springframework.batch.io.file.transform.DelimitedLineTokenizer; import org.springframework.batch.io.file.transform.LineTokenizer; +import org.springframework.batch.item.exception.StreamException; import org.springframework.core.io.AbstractResource; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; @@ -139,7 +140,12 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader = new FlatFileItemReader(); itemReader.setResource(getInputResource(TEST_STRING)); itemReader.setFieldSetMapper(fieldSetMapper); - assertEquals("[FlatFileInputTemplate-TestData]", itemReader.read().toString()); + try { + itemReader.read(); + fail("Expected StreamException"); + } catch (StreamException e) { + assertTrue(e.getMessage().contains("open")); + } } public void testCloseBeforeOpen() throws Exception { @@ -147,7 +153,8 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setResource(getInputResource(TEST_STRING)); itemReader.setFieldSetMapper(fieldSetMapper); itemReader.close(); - // The open still happens automatically on a read... + // The open does not happen automatically on a read... + itemReader.open(); assertEquals("[FlatFileInputTemplate-TestData]", itemReader.read().toString()); } @@ -172,6 +179,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { itemReader.setEncoding("UTF-8"); itemReader.setResource(getInputResource(TEST_STRING)); itemReader.setFieldSetMapper(fieldSetMapper); + itemReader.open(); testRead(); } @@ -301,6 +309,7 @@ public class FlatFileItemReaderBasicTests extends TestCase { //replace the resource to simulate runtime resource creation testReader.setResource(getInputResource(TEST_STRING)); + testReader.open(); assertEquals(TEST_OUTPUT, testReader.read().toString()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java index 5b6e20a04..90d2105fa 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java @@ -15,6 +15,7 @@ import javax.xml.stream.events.XMLEvent; import junit.framework.TestCase; import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.exception.StreamException; import org.springframework.core.io.AbstractResource; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; @@ -131,23 +132,19 @@ public class StaxEventItemReaderTests extends TestCase { context.putLong(StaxEventItemReader.READ_COUNT_STATISTICS_NAME, 100000); try { source.restoreFrom(context); - fail(); + fail("Expected StreamException"); } - catch (IllegalStateException e) { - // expected - } - - source = createNewInputSouce(); - source.open(); - try { - source.restoreFrom(new ExecutionAttributes()); - fail(); - } - catch (IllegalStateException e) { + catch (StreamException e) { // expected + String message = e.getMessage(); + assertTrue("Wrong message: "+message, message.contains("must be before")); } } + public void testRestoreWorksFromClosedStream() throws Exception { + source.close(); + source.restoreFrom(new ExecutionAttributes()); + } /** * Skipping marked records after rollback. */ @@ -224,16 +221,22 @@ public class StaxEventItemReaderTests extends TestCase { newSource.setFragmentRootElementName(FRAGMENT_ROOT_ELEMENT); newSource.setFragmentDeserializer(deserializer); + newSource.open(); + Object item = newSource.read(); assertNotNull(item); assertTrue(newSource.isOpenCalled()); - newSource.close(); + newSource.close(); newSource.setOpenCalled(false); // calling read again should require re-initialization because of close - item = newSource.read(); - assertNotNull(item); - assertTrue(newSource.isOpenCalled()); + try { + item = newSource.read(); + fail("Expected StreamException"); + } + catch (StreamException e) { + // expected + } } public void testOpenBadIOInput() { @@ -293,6 +296,8 @@ public class StaxEventItemReaderTests extends TestCase { newSource.setFragmentRootElementName(FRAGMENT_ROOT_ELEMENT); newSource.setFragmentDeserializer(deserializer); + newSource.open(); + return newSource; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java index 476c56eef..e4ab425d1 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java @@ -45,6 +45,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { protected void setUp() throws Exception { resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", "xml")); writer = createItemWriter(); + writer.open(); } /** @@ -84,17 +85,18 @@ public class StaxEventWriterItemWriterTests extends TestCase { * Restart scenario - content is appended to the output file after restart. */ public void testRestart() throws Exception { - // write records + // write record writer.write(record); - writer.mark(); + // writer.mark(); ExecutionAttributes streamContext = writer.getExecutionAttributes(); + writer.close(); // create new writer from saved restart data and continue writing writer = createItemWriter(); writer.restoreFrom(streamContext); writer.write(record); writer.close(); - + // check the output is concatenation of 'before restart' and 'after // restart' writes. String outputFile = outputFileContent(); @@ -199,7 +201,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { source.setOverwriteOutput(true); source.afterPropertiesSet(); - + return source; } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AbstractTradeBatchTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AbstractTradeBatchTests.java index b4ba31075..993861206 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AbstractTradeBatchTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AbstractTradeBatchTests.java @@ -45,6 +45,7 @@ public abstract class AbstractTradeBatchTests extends TestCase { protected void setUp() throws Exception { super.setUp(); provider = new TradeItemReader(resource); + provider.open(); } protected static class TradeItemReader extends DelegatingItemReader { diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java index 9eb874fef..ceb291bd2 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java @@ -14,6 +14,7 @@ import org.springframework.batch.execution.scope.StepContextAware; import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.KeyedItemReader; +import org.springframework.batch.item.exception.StreamException; import org.springframework.batch.sample.item.writer.StagingItemWriter; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.jdbc.core.RowMapper; @@ -127,7 +128,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Key private Long doRead() { if (!initialized) { - open(); + throw new StreamException("ItemStream must be open before it can be read."); } Long key = getBuffer().next(); diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractBatchLauncherTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractBatchLauncherTests.java index 7295f5984..3920789bf 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractBatchLauncherTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractBatchLauncherTests.java @@ -29,9 +29,7 @@ import org.springframework.test.AbstractDependencyInjectionSpringContextTests; * Abstract unit test for running functional tests by getting context locations * for both the container and configuration separately and having them auto * wired in by type. This allows the two to be completely separated, and remove - * any 'configuration coupling' between the two. However, it is still purely - * theoretical until a decision is made as to how job configuration and - * container configuration files are pulled together. + * any 'configuration coupling' between the two. * * @author Lucas Ward * diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/FixedLengthImportJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/FixedLengthImportJobFunctionalTests.java index 94b9a8614..6aee96386 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/FixedLengthImportJobFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/FixedLengthImportJobFunctionalTests.java @@ -60,6 +60,8 @@ public class FixedLengthImportJobFunctionalTests extends AbstractValidatingBatch * @throws Exception */ protected void validatePostConditions() throws Exception { + + inputSource.open(); jdbcTemplate.query("SELECT ID, ISIN, QUANTITY, PRICE, CUSTOMER FROM trade ORDER BY id", new RowCallbackHandler() { diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java index 7ffbf900d..915c0d41c 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java @@ -48,6 +48,7 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin writer.write("BAR"); writer.write("SPAM"); writer.write("BUCKET"); + provider.open(); } protected void onTearDownAfterTransaction() throws Exception {