From 38c486e449a496e852a6155b7ab71400af43c369 Mon Sep 17 00:00:00 2001 From: lucasward Date: Tue, 26 Feb 2008 08:29:48 +0000 Subject: [PATCH] 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. --- .../repository/dao/JdbcStepExecutionDao.java | 1 - .../execution/scope/SimpleStepContext.java | 24 +--------- .../execution/step/ItemOrientedStep.java | 44 ++++++++----------- .../execution/step/ItemOrientedStepTests.java | 33 +++++++++----- 4 files changed, 41 insertions(+), 61 deletions(-) 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() {