Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - issue BATCH-364: StepScope responsibilities can be assumed by Step (not ApplicationContext)

Browse Source 
http://jira.springframework.org/browse/BATCH-364

Instead of lazy open, throw exception is stream is used before open().
This commit is contained in:
dsyer
2008-02-15 09:19:25 +00:00
parent ff0f6266a9
commit 711d0c6e92
15 changed files with 167 additions and 671 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java
View File

@@ -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').
*
*

164
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -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).<br/>
* 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).<br/>
*
* 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}.<br/>
* 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}.<br/>
*
* @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.<br/>
* 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.<br/>
*
* 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()
*/

566
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java
View File

@@ -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 {
}
}
}

20
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemReader.java
View File

@@ -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;

12
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java
View File

@@ -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
}
/**

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java
View File

@@ -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++;

1
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderAdvancedTests.java
View File

@@ -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());
}

13
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemReaderBasicTests.java
View File

@@ -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());
}

37
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemReaderTests.java
View File

@@ -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;
}

10
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java
View File

@@ -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;
}
}

1
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/AbstractTradeBatchTests.java
View File

@@ -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 {

3
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
View File

@@ -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();

4
spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractBatchLauncherTests.java
View File

@@ -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
*

2
spring-batch-samples/src/test/java/org/springframework/batch/sample/FixedLengthImportJobFunctionalTests.java
View File

@@ -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() {

1
spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java
View File

@@ -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 {