From 786f756ba22c218b2553fd63bdd81fece6c79e28 Mon Sep 17 00:00:00 2001 From: lucasward Date: Tue, 19 Feb 2008 05:55:34 +0000 Subject: [PATCH] BATCH-220: Updated sample jobs to use 'chunkedStep' --- .../execution/step/simple/ChunkedStep.java | 190 ++++++++++-------- .../execution/step/simple/ItemChunker.java | 53 +++++ .../execution/step/simple/ItemDechunker.java | 54 +++++ 3 files changed, 218 insertions(+), 79 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java index b454a5468..2c0cdfd79 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java @@ -55,6 +55,8 @@ import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.policy.NeverRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.TransactionStatus; @@ -147,6 +149,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { private ItemSkipPolicy itemSkipPolicy; private RetryTemplate retryTemplate = new RetryTemplate(); + + private RetryPolicy retryPolicy = new NeverRetryPolicy(); private int chunkSize; @@ -161,8 +165,9 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * 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; @@ -187,7 +192,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * processing. Should be set up by the caller through a factory. Defaults to * a plain {@link RepeatTemplate}. * - * @param stepOperations a {@link RepeatOperations} instance. + * @param stepOperations + * a {@link RepeatOperations} instance. */ public void setStepOperations(RepeatOperations stepOperations) { this.stepOperations = stepOperations; @@ -198,7 +204,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * check whether an external request has been made to interrupt the job * execution. * - * @param interruptionPolicy a {@link StepInterruptionPolicy} + * @param interruptionPolicy + * a {@link StepInterruptionPolicy} */ public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { this.interruptionPolicy = interruptionPolicy; @@ -210,7 +217,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * * @param exceptionClassifier */ - public void setExceptionClassifier(ExitStatusExceptionClassifier exceptionClassifier) { + public void setExceptionClassifier( + ExitStatusExceptionClassifier exceptionClassifier) { this.exceptionClassifier = exceptionClassifier; } @@ -220,6 +228,10 @@ public class ChunkedStep extends StepSupport implements InitializingBean { public void setItemReader(ItemReader itemReader) { this.itemReader = itemReader; } + + public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; + } /** * @param itemWriter @@ -273,6 +285,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { } Assert.notNull(jobRepository, "JobRepository must not be null"); + + this.retryTemplate.setRetryPolicy(retryPolicy); } /** @@ -285,57 +299,71 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * execution, which would normally be available to the caller somehow * through the step's {@link StepContext}.
* - * @throws JobInterruptedException if the step or a chunk is interrupted - * @throws RuntimeException if there is an exception during a chunk - * execution + * @throws JobInterruptedException + * if the step or a chunk is interrupted + * @throws RuntimeException + * if there is an exception during a chunk execution * @see StepExecutor#execute(StepExecution) */ - public void execute(final StepExecution stepExecution) throws BatchCriticalException, JobInterruptedException { + public void execute(final StepExecution stepExecution) + throws BatchCriticalException, JobInterruptedException { final StepInstance stepInstance = stepExecution.getStep(); Assert.notNull(stepInstance); - boolean isRestart = stepInstance.getStepExecutionCount() > 0 ? true : false; + boolean isRestart = stepInstance.getStepExecutionCount() > 0 ? true + : false; ExitStatus status = ExitStatus.FAILED; final int chunkSize = this.chunkSize; - StepContext parentStepContext = StepSynchronizationManager.getContext(); - final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager); - StepSynchronizationManager.register(stepContext); - possiblyRegisterStreams(stepExecution); - // 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); - - if (saveExecutionContext && isRestart && stepInstance.getLastExecution() != null) { - stepExecution.setExecutionContext(stepInstance.getLastExecution().getExecutionContext()); - streamManager.restoreFrom(stepExecution, stepExecution.getExecutionContext()); - } - try { - stepExecution.setStartTime(new Date(System.currentTimeMillis())); - stepInstance.setLastExecution(stepExecution); + // We need to save the step execution right away, before we start + // using its ID. It would be better to make the creation atomic in + // the caller. updateStatus(stepExecution, BatchStatus.STARTED); + StepContext parentStepContext = StepSynchronizationManager + .getContext(); + final StepContext stepContext = new SimpleStepContext( + stepExecution, parentStepContext, streamManager); + StepSynchronizationManager.register(stepContext); + possiblyRegisterStreams(stepExecution); + // 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); + + if (saveExecutionContext && isRestart + && stepInstance.getLastExecution() != null) { + stepExecution.setExecutionContext(stepInstance + .getLastExecution().getExecutionContext()); + streamManager.restoreFrom(stepExecution, stepExecution + .getExecutionContext()); + } + status = stepOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { + public ExitStatus doInIteration(final RepeatContext context) + throws Exception { // Before starting a new transaction, check for // interruption. interruptionPolicy.checkInterrupted(context); - ExitStatus result = (ExitStatus)retryTemplate.execute(new RetryCallback() { + ExitStatus result = (ExitStatus) retryTemplate + .execute(new RetryCallback() { - public Object doWithRetry(RetryContext context) throws Throwable { - return processChunk(stepExecution, stepContext); - } - }); + public Object doWithRetry(RetryContext context) + throws Throwable { + return processChunk(stepExecution, + stepContext); + } + }); // Check for interruption after transaction as well, so that // the interrupted exception is correctly propagated up to @@ -348,35 +376,36 @@ public class ChunkedStep extends StepSupport 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 { + streamManager.close(stepExecution); + } catch (Exception e) { + logger + .error( + "Failed to update step execution: probably fatal, so there is already an exception on the stack.", + e); + } finally { // clear any registered synchronizations StepSynchronizationManager.close(); - streamManager.close(stepExecution); + } } @@ -386,35 +415,36 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * */ private void possiblyRegisterStreams(Object key) { - if (itemReader instanceof ItemStream) { - ItemStream stream = (ItemStream) itemReader; - streamManager.register(key, stream); - } - if (itemWriter instanceof ItemStream) { - ItemStream stream = (ItemStream) itemWriter; - streamManager.register(key, stream); - } + + streamManager.register(key, chunker); + streamManager.register(key, dechunker); } /** * Execute a bunch of identical business logic operations all within a * transaction. * - * @param stepExecution the current execution in which to process the chunk - * in. - * @param chunk to be processed. - * @param stepContext the current step context. + * @param stepExecution + * the current execution in which to process the chunk in. + * @param chunk + * to be processed. + * @param stepContext + * the current step context. * @return true if there is more data to process. */ - ExitStatus processChunk(final StepExecution stepExecution, StepContext stepContext) { + ExitStatus processChunk(final StepExecution stepExecution, + StepContext stepContext) { - TransactionStatus transaction = streamManager.getTransaction(stepExecution); + TransactionStatus transaction = streamManager + .getTransaction(stepExecution); - final StepContribution contribution = stepExecution.createStepContribution(); + final StepContribution contribution = stepExecution + .createStepContribution(); try { - - ChunkingResult chunkingResult = chunker.chunk(chunkSize, stepExecution); + + ChunkingResult chunkingResult = chunker.chunk(chunkSize, + stepExecution); if (chunkingResult == null) { return ExitStatus.FINISHED; @@ -423,13 +453,15 @@ public class ChunkedStep extends StepSupport implements InitializingBean { final Chunk chunk = chunkingResult.getChunk(); failureLog.handle(chunkingResult.getExceptions()); - DechunkingResult chunkResult = dechunker.dechunk(chunk, stepExecution); - failureLog.handle(chunkResult.getExceptions()); + DechunkingResult dechunkingResult = dechunker.dechunk(chunk, + stepExecution); + failureLog.handle(dechunkingResult.getExceptions()); // TODO: check that stepExecution can // aggregate these contributions if they // come in asynchronously. - ExecutionContext statistics = streamManager.getExecutionContext(stepExecution); + ExecutionContext statistics = streamManager + .getExecutionContext(stepExecution); contribution.setExecutionContext(statistics); contribution.incrementCommitCount(); @@ -450,11 +482,10 @@ public class ChunkedStep extends StepSupport implements InitializingBean { } streamManager.commit(transaction); - + return ExitStatus.CONTINUABLE; - } - catch (Throwable t) { + } catch (Throwable t) { /* * Any exception thrown within the transaction template will * automatically cause the transaction to rollback. We need to @@ -466,23 +497,24 @@ public class ChunkedStep extends StepSupport implements InitializingBean { } 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); - 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); + .error( + "Encountered reset error on rollback: " + + "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); } if (t instanceof RuntimeException) { throw (RuntimeException) t; - } - else { + } else { throw new RuntimeException(t); } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java index 6e3f8aa72..df6b5da4d 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java @@ -24,7 +24,12 @@ import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.ReadFailureException; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.exception.MarkFailedException; +import org.springframework.batch.item.exception.ResetFailedException; +import org.springframework.batch.item.exception.StreamException; import org.springframework.util.Assert; /** @@ -95,4 +100,52 @@ public class ItemChunker implements Chunker { return new Long(chunkCounter++); } + //These methods are temporary hacks until something can be done + //about the ItemStream interface. + public void close() throws StreamException { + if(itemReader instanceof ItemStream){ + ((ItemStream)itemReader).close(); + } + } + + public boolean isMarkSupported() { + if(itemReader instanceof ItemStream){ + return ((ItemStream)itemReader).isMarkSupported(); + } + + return false; + } + + public void mark() throws MarkFailedException { + if(itemReader instanceof ItemStream){ + ((ItemStream)itemReader).mark(); + } + } + + public void open() throws StreamException { + if(itemReader instanceof ItemStream){ + ((ItemStream)itemReader).open(); + } + } + + public void reset() throws ResetFailedException { + if(itemReader instanceof ItemStream){ + ((ItemStream)itemReader).reset(); + } + } + + public void restoreFrom(ExecutionContext context) { + if(itemReader instanceof ItemStream){ + ((ItemStream)itemReader).restoreFrom(context); + } + } + + public ExecutionContext getExecutionContext() { + if(itemReader instanceof ItemStream){ + return ((ItemStream)itemReader).getExecutionContext(); + } + + return new ExecutionContext(); + } + } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java index c658430c7..d90bcd4a7 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java @@ -25,7 +25,12 @@ import org.springframework.batch.core.domain.Dechunker; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.WriteFailureException; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStream; 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.util.Assert; /** @@ -84,5 +89,54 @@ public class ItemDechunker implements Dechunker { throw new RuntimeException("Error encountered while dechunking", ex); } } + + //This a hack until the ItemStream interface can be updated. + public void close() throws StreamException { + if(itemWriter instanceof ItemStream){ + ((ItemStream)itemWriter).close(); + } + } + + public boolean isMarkSupported() { + if(itemWriter instanceof ItemStream){ + return ((ItemStream)itemWriter).isMarkSupported(); + } + else{ + return false; + } + } + + public void mark() throws MarkFailedException { + if(itemWriter instanceof ItemStream){ + ((ItemStream)itemWriter).mark(); + } + } + + public void open() throws StreamException { + if(itemWriter instanceof ItemStream){ + ((ItemStream)itemWriter).open(); + } + } + + public void reset() throws ResetFailedException { + if(itemWriter instanceof ItemStream){ + ((ItemStream)itemWriter).reset(); + } + } + + public void restoreFrom(ExecutionContext context) { + if(itemWriter instanceof ItemStream){ + ((ItemStream)itemWriter).restoreFrom(context); + } + } + + public ExecutionContext getExecutionContext() { + if(itemWriter instanceof ItemStream){ + return ((ItemStream)itemWriter).getExecutionContext(); + } + else{ + return new ExecutionContext(); + } + } }