diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java index 9f4376396..d8f643dc4 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepExecutionDao.java @@ -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; diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java index 2686b33f9..e99819ef0 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java @@ -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) { diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java index cc044fbf2..cac7f6146 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java @@ -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. diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java index e0235f8e4..9a56e4bd3 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java @@ -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() {