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

BATCH-365: There should now be one ExecutionContext per step. All itemStreams will be opened with an execution context, and will be notified before it is saved, to ensure they have all state in the context. Most ItemReader/Writers should now have the logic for whether or not to put their state in the context, but a few have likely been missed.

Browse Source
This commit is contained in:
lucasward
2008-02-26 08:29:48 +00:00
parent 64506040ff
commit 38c486e449
4 changed files with 41 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java
View File

@@ -16,7 +16,6 @@ import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepSupport;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.ExitStatus;

24
spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java
View File

@@ -25,7 +25,6 @@ import java.util.Set;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.stream.StreamManager;
import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor;
/**
@@ -42,29 +41,19 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
private StepExecution stepExecution;
private StreamManager streamManager;
/**
* Default constructor.
*/
public SimpleStepContext(StepExecution stepExecution) {
this(stepExecution, null, null);
}
/**
* Default constructor.
*/
public SimpleStepContext(StepExecution stepExecution, StepContext parent) {
this(stepExecution, parent, null);
this(stepExecution, null);
}
/**
* @param object
*/
public SimpleStepContext(StepExecution stepExecution, StepContext parent, StreamManager streamManager) {
public SimpleStepContext(StepExecution stepExecution, StepContext parent) {
super();
this.parent = parent;
this.streamManager = streamManager;
this.stepExecution = stepExecution;
}
@@ -117,15 +106,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements
List errors = new ArrayList();
try {
if (streamManager != null) {
streamManager.close(this);
}
}
catch (Exception t) {
errors.add(t);
}
Set copy;
synchronized (callbacks) {

44
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
View File

@@ -100,6 +100,8 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
private ItemReaderRetryCallback retryCallback;
private int commitInterval = 0;
private boolean saveExecutionContext = false;
/**
* The {@link RepeatOperations} to use for the outer loop of the batch
@@ -247,20 +249,20 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
updateStatus(stepExecution, BatchStatus.STARTED);
StepContext parentStepContext = StepSynchronizationManager.getContext();
final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager);
final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext);
StepSynchronizationManager.register(stepContext);
possiblyRegisterStreams(stepExecution);
possiblyRegisterStreams();
// Add the job identifier so that it can be used to identify
// the conversation in StepScope
stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId());
final boolean saveExecutionContext = isSaveExecutionContext();
streamManager.open(stepExecution);
streamManager.open(stepExecution.getExecutionContext());
if (saveExecutionContext && isRestart && lastStepExecution != null) {
stepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
streamManager.restoreFrom(stepExecution, stepExecution.getExecutionContext());
}
else{
stepExecution.setExecutionContext(new ExecutionContext());
}
status = stepOperations.iterate(new RepeatCallback() {
@@ -268,30 +270,21 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
final StepContribution contribution = stepExecution.createStepContribution();
contribution.setExecutionContext(stepExecution.getExecutionContext());
// Before starting a new transaction, check for
// interruption.
interruptionPolicy.checkInterrupted(context);
ExitStatus result;
streamManager.open(stepExecution.getExecutionContext());
TransactionStatus transaction = streamManager.getTransaction(stepExecution);
TransactionStatus transaction = streamManager.getTransaction();
try {
itemReader.mark();
result = processChunk(contribution);
// TODO: check that stepExecution can
// aggregate these contributions if they
// come in asynchronously.
ExecutionContext statistics;
if(isSaveExecutionContext()){
statistics = streamManager.getExecutionContext(stepExecution);
contribution.setExecutionContext(statistics);
}
else{
statistics = new ExecutionContext();
}
contribution.incrementCommitCount();
// If the step operations are asynchronous then we need
@@ -303,9 +296,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
// only if chunk was successful
stepExecution.apply(contribution);
if (saveExecutionContext) {
stepExecution.setExecutionContext(statistics);
}
streamManager.beforeSave();
jobRepository.saveOrUpdate(stepExecution);
}
@@ -384,7 +375,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
stepExecution.setEndTime(new Date(System.currentTimeMillis()));
try {
jobRepository.saveOrUpdate(stepExecution);
streamManager.close(stepExecution);
streamManager.close();
}
catch (Exception e) {
logger
@@ -404,14 +395,14 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
/**
*
*/
private void possiblyRegisterStreams(Object key) {
private void possiblyRegisterStreams() {
if (itemReader instanceof ItemStream) {
ItemStream stream = (ItemStream) itemReader;
streamManager.register(key, stream);
streamManager.register(stream);
}
if (itemWriter instanceof ItemStream) {
ItemStream stream = (ItemStream) itemWriter;
streamManager.register(key, stream);
streamManager.register(stream);
}
}
@@ -543,6 +534,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
((Skippable) this.itemWriter).skip();
}
}
/**
* Convenience method to update the status in all relevant places.

33
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
View File

@@ -45,6 +45,7 @@ import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.item.reader.AbstractItemReader;
@@ -112,7 +113,6 @@ public class ItemOrientedStepTests extends TestCase {
jobInstance = new JobInstance(new Long(0), new JobParameters(), new JobSupport("FOO"));
SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager);
streamManager.setUseClassNameAsPrefix(false);
itemOrientedStep.setStreamManager(streamManager);
}
@@ -337,7 +337,6 @@ public class ItemOrientedStepTests extends TestCase {
}
assertFalse(tasklet.isRestoreFromCalled());
assertFalse(tasklet.isGetExecutionAttributesCalled());
}
/*
@@ -413,9 +412,15 @@ public class ItemOrientedStepTests extends TestCase {
final Map map = new HashMap();
itemOrientedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) {
public ExecutionContext getExecutionContext(Object key) {
ExecutionContext executionContext;
public void beforeSave() {
// TODO Auto-generated method stub
return new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"));
executionContext.putString("foo", "bar");
}
public void open(ExecutionContext executionContext)
throws StreamException {
this.executionContext = executionContext;
}
});
@@ -435,6 +440,8 @@ public class ItemOrientedStepTests extends TestCase {
private boolean restoreFromCalled = false;
private boolean restoreFromCalledWithSomeContext = false;
private ExecutionContext executionContext;
public Object read() throws Exception {
StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this);
@@ -445,14 +452,9 @@ public class ItemOrientedStepTests extends TestCase {
return restoreFromCalledWithSomeContext;
}
public ExecutionContext getExecutionContext() {
public void beforeSave() {
getExecutionAttributesCalled = true;
return new ExecutionContext(PropertiesConverter.stringToProperties("spam=bucket"));
}
public void restoreFrom(ExecutionContext data) {
restoreFromCalled = true;
restoreFromCalledWithSomeContext = data.getProperties().size() > 0;
executionContext.putString("spam", "bucket");
}
public boolean isGetExecutionAttributesCalled() {
@@ -463,12 +465,19 @@ public class ItemOrientedStepTests extends TestCase {
return restoreFromCalled;
}
public void open() throws StreamException {
public void open(ExecutionContext executionContext) throws StreamException {
this.executionContext = executionContext;
}
public void close() throws StreamException {
}
public void mark() throws MarkFailedException {
}
public void reset() throws ResetFailedException {
}
}
public void testStatusForInterruptedException() {