From 42ba209b1185375e8809f124a13352e443e6abba Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 26 Feb 2008 14:29:27 +0000 Subject: [PATCH] Incomplete - task 84: Fix samples Fix skip policy so it responds to subclasses --- .../batch/execution/step/AbstractStep.java | 10 - .../execution/step/ItemOrientedStep.java | 1099 ++++++++-------- .../support/LimitCheckingItemSkipPolicy.java | 26 +- .../execution/launch/SimpleJobTests.java | 385 +++--- .../repository/dao/AbstractStepDaoTests.java | 452 ++++--- .../execution/step/ItemOrientedStepTests.java | 1128 ++++++++--------- 6 files changed, 1533 insertions(+), 1567 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java index 08c349a81..536a47cc0 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/AbstractStep.java @@ -42,8 +42,6 @@ import org.springframework.util.Assert; */ public abstract class AbstractStep extends StepSupport implements InitializingBean { - private int skipLimit = 0; - protected ExceptionHandler exceptionHandler; protected RetryPolicy retryPolicy; @@ -102,14 +100,6 @@ public abstract class AbstractStep extends StepSupport implements InitializingBe this.exceptionHandler = exceptionHandler; } - public void setSkipLimit(int skipLimit) { - this.skipLimit = skipLimit; - } - - public int getSkipLimit() { - return skipLimit; - } - /** * Public setter for {@link JobRepository}. * 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 04adc66b6..2fae057b6 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 @@ -1,556 +1,543 @@ -/* - * 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; - -import java.util.Date; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.batch.core.domain.BatchStatus; -import org.springframework.batch.core.domain.JobInstance; -import org.springframework.batch.core.domain.JobInterruptedException; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; -import org.springframework.batch.core.tasklet.Tasklet; -import org.springframework.batch.execution.scope.SimpleStepContext; -import org.springframework.batch.execution.scope.StepContext; -import org.springframework.batch.execution.scope.StepScope; -import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier; -import org.springframework.batch.execution.step.support.StepInterruptionPolicy; -import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy; -import org.springframework.batch.io.Skippable; -import org.springframework.batch.io.exception.BatchCriticalException; -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemRecoverer; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.KeyedItemReader; -import org.springframework.batch.item.exception.ResetFailedException; -import org.springframework.batch.item.stream.SimpleStreamManager; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.exception.handler.ExceptionHandler; -import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.callback.ItemReaderRetryCallback; -import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; -import org.springframework.beans.factory.InitializingBean; -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).
- * - * 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 - * @author Ben Hale - */ -public class ItemOrientedStep extends AbstractStep implements InitializingBean { - - private static final Log logger = LogFactory.getLog(ItemOrientedStep.class); - - private RepeatOperations chunkOperations = new RepeatTemplate(); - - private RepeatOperations stepOperations = new RepeatTemplate(); - - private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier(); - - // default to checking current thread for interruption. - private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); - - private RetryPolicy retryPolicy = null; - - private RetryTemplate template = new RetryTemplate(); - - private ItemReaderRetryCallback retryCallback; - - private int commitInterval = 0; - - /** - * 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. - */ - public void setStepOperations(RepeatOperations stepOperations) { - this.stepOperations = stepOperations; - } - - /** - * 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. - */ - public void setChunkOperations(RepeatOperations chunkoperations) { - this.chunkOperations = chunkoperations; - } - - /** - * 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} - */ - public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { - this.interruptionPolicy = interruptionPolicy; - } - - /** - * Setter for the {@link ExitStatusExceptionClassifier} that will be used to - * classify any exception that causes a job to fail. - * - * @param exceptionClassifier - */ - public void setExceptionClassifier(ExitStatusExceptionClassifier exceptionClassifier) { - this.exceptionClassifier = exceptionClassifier; - } - - /** - * Public setter for the retryPolicy. - * - * @param retyPolicy the retryPolicy to set - */ - public void setRetryPolicy(RetryPolicy retryPolicy) { - this.retryPolicy = retryPolicy; - } - - public void setCommitInterval(int commitInterval) { - this.commitInterval = commitInterval; - } - - /** - * Check mandatory properties (reader and writer). - * - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() - */ - public void afterPropertiesSet() throws Exception { - Assert.notNull(itemReader, "ItemReader must be provided"); - Assert.notNull(itemWriter, "ItemWriter must be provided"); - - ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); - template.setRetryPolicy(itemProviderRetryPolicy); - - if (retryPolicy != null) { - Assert.state(itemReader instanceof KeyedItemReader, - "ItemReader must be instance of KeyedItemReader to use the retry policy"); - retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); - } - - if (streamManager == null && transactionManager != null) { - streamManager = new SimpleStreamManager(transactionManager); - } - else if (streamManager == null && transactionManager == null) { - throw new IllegalArgumentException("Either StreamManager or TransactionManager must be set"); - } - - if (commitInterval > 0) { - ((RepeatTemplate) chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); - } - - if (exceptionHandler != null) { - ((RepeatTemplate) chunkOperations).setExceptionHandler(exceptionHandler); - } - } - - /** - * Apply the configuration by inspecting it to see if it has any relevant - * policy information. - * - * @param step a step - */ - void applyConfiguration(AbstractStep step) { - - ExceptionHandler exceptionHandler = step.getExceptionHandler(); - - if (step.getSkipLimit() > 0 && exceptionHandler == null) { - SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler(); - handler.setLimit(step.getSkipLimit()); - exceptionHandler = handler; - } - - if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) { - RepeatTemplate template = (RepeatTemplate) this.stepOperations; - template.setExceptionHandler(exceptionHandler); - } - - } - - /** - * 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}.
- * - * @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 { - - JobInstance jobInstance = stepExecution.getJobExecution().getJobInstance(); - StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, this); - - boolean isRestart = jobRepository.getStepExecutionCount(jobInstance, this) > 0 ? true : false; - - ExitStatus status = ExitStatus.FAILED; - - try { - - stepExecution.setStartTime(new Date(System.currentTimeMillis())); - // 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); - StepSynchronizationManager.register(stepContext); - 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()); - - if (isSaveExecutionContext() && isRestart && lastStepExecution != null) { - stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); - } - else { - stepExecution.setExecutionContext(new ExecutionContext()); - } - - // Open the stream manager *after* the execution context is fixed in - // the step, otherwise it will not be the same reference that is - // updated by the streams. TODO: this is a little fragile - maybe - // StreamManager.update() should accept the context as a parameter. - streamManager.open(stepExecution.getExecutionContext()); - - status = stepOperations.iterate(new RepeatCallback() { - - 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; - - TransactionStatus transaction = streamManager.getTransaction(); - - try { - itemReader.mark(); - result = processChunk(contribution); - - contribution.incrementCommitCount(); - - // If the step operations are asynchronous then we need - // to synchronize changes to the step execution (at a - // minimum). - synchronized (stepExecution) { - - // Apply the contribution to the step - // only if chunk was successful - stepExecution.apply(contribution); - - streamManager.update(); - jobRepository.saveOrUpdate(stepExecution); - - } - - itemReader.mark(); - itemWriter.flush(); - streamManager.commit(transaction); - - } - 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. - */ - synchronized (stepExecution) { - stepExecution.rollback(); - } - try { - itemReader.reset(); - itemWriter.clear(); - streamManager.rollback(transaction); - } - 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); - } - if (t instanceof RuntimeException) { - throw (RuntimeException) t; - } - else { - throw new RuntimeException(t); - } - } - - // Check for interruption after transaction as well, so that - // the interrupted exception is correctly propagated up to - // caller - interruptionPolicy.checkInterrupted(context); - - return result; - - } - }); - - updateStatus(stepExecution, BatchStatus.COMPLETED); - } - 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) { - updateStatus(stepExecution, BatchStatus.UNKNOWN); - throw (ResetFailedException) e; - } - else { - updateStatus(stepExecution, BatchStatus.FAILED); - throw e; - } - - } - finally { - stepExecution.setExitStatus(status); - stepExecution.setEndTime(new Date(System.currentTimeMillis())); - try { - jobRepository.saveOrUpdate(stepExecution); - streamManager.close(); - } - 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(); - } - } - - } - - /** - * - */ - private void possiblyRegisterStreams() { - if (itemReader instanceof ItemStream) { - ItemStream stream = (ItemStream) itemReader; - streamManager.register(stream); - } - if (itemWriter instanceof ItemStream) { - ItemStream stream = (ItemStream) itemWriter; - streamManager.register(stream); - } - } - - /** - * 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. - * @return true if there is more data to process. - */ - ExitStatus processChunk(final StepContribution contribution) { - ExitStatus result = chunkOperations.iterate(new RepeatCallback() { - public ExitStatus doInIteration(final RepeatContext context) throws Exception { - if (contribution.isTerminateOnly()) { - context.setTerminateOnly(); - } - // check for interruption before each item as well - interruptionPolicy.checkInterrupted(context); - ExitStatus exitStatus = doProcessing(contribution); - contribution.incrementTaskCount(); - // check for interruption after each item as well - interruptionPolicy.checkInterrupted(context); - return exitStatus; - } - }); - return result; - } - - /** - * 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.
- * - * 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 - * @return boolean if there is more processing to do - * @throws Exception if there is an error - */ - private ExitStatus doProcessing(StepContribution contribution) throws Exception { - ExitStatus exitStatus = ExitStatus.CONTINUABLE; - - try { - - exitStatus = execute(); - - } - catch (Exception e) { - if (getItemSkipPolicy().shouldSkip(e, contribution)) { - skip(); - } - else { - // Rethrow so that outer transaction is rolled back properly - throw e; - } - } - - return exitStatus; - } - - /** - * 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() - */ - private ExitStatus execute() throws Exception { - - if (retryCallback == null) { - Object item = null; - try { - item = itemReader.read(); - } - catch (Exception ex) { - getItemFailureHandler().handleReadFailure(ex); - throw ex; - } - if (item == null) { - return ExitStatus.FINISHED; - } - try { - itemWriter.write(item); - } - catch (Exception e) { - - getItemFailureHandler().handleWriteFailure(item, e); - // Re-throw the exception so that the surrounding transaction - // rolls back if there is one - throw e; - } - return ExitStatus.CONTINUABLE; - } - - return new ExitStatus(template.execute(retryCallback) != null); - - } - - /** - * 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() - */ - private void skip() { - if (retryCallback != null) { - // No need to skip because the recoverer will take any action - // necessary. - return; - } - if (this.itemReader instanceof Skippable) { - ((Skippable) this.itemReader).skip(); - } - if (this.itemWriter instanceof Skippable) { - ((Skippable) this.itemWriter).skip(); - } - } - - /** - * Convenience method to update the status in all relevant places. - * - * @param stepInstance the current step - * @param stepExecution the current stepExecution - * @param status the status to set - */ - private void updateStatus(StepExecution stepExecution, BatchStatus status) { - stepExecution.setStatus(status); - try { - jobRepository.saveOrUpdate(stepExecution); - } - catch (Exception e) { - logger.error("Failed to update step execution with status: probably fatal.", e); - } - - } -} +/* + * 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; + +import java.util.Date; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.domain.BatchStatus; +import org.springframework.batch.core.domain.JobInstance; +import org.springframework.batch.core.domain.JobInterruptedException; +import org.springframework.batch.core.domain.StepContribution; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; +import org.springframework.batch.core.tasklet.Tasklet; +import org.springframework.batch.execution.scope.SimpleStepContext; +import org.springframework.batch.execution.scope.StepContext; +import org.springframework.batch.execution.scope.StepScope; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier; +import org.springframework.batch.execution.step.support.StepInterruptionPolicy; +import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy; +import org.springframework.batch.io.Skippable; +import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.KeyedItemReader; +import org.springframework.batch.item.exception.ResetFailedException; +import org.springframework.batch.item.stream.SimpleStreamManager; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.callback.ItemReaderRetryCallback; +import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.beans.factory.InitializingBean; +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).
+ * + * 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 + * @author Ben Hale + */ +public class ItemOrientedStep extends AbstractStep implements InitializingBean { + + private static final Log logger = LogFactory.getLog(ItemOrientedStep.class); + + private RepeatOperations chunkOperations = new RepeatTemplate(); + + private RepeatOperations stepOperations = new RepeatTemplate(); + + private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier(); + + // default to checking current thread for interruption. + private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); + + private RetryPolicy retryPolicy = null; + + private RetryTemplate template = new RetryTemplate(); + + private ItemReaderRetryCallback retryCallback; + + private int commitInterval = 0; + + /** + * 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. + */ + public void setStepOperations(RepeatOperations stepOperations) { + this.stepOperations = stepOperations; + } + + /** + * 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. + */ + public void setChunkOperations(RepeatOperations chunkoperations) { + this.chunkOperations = chunkoperations; + } + + /** + * 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} + */ + public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { + this.interruptionPolicy = interruptionPolicy; + } + + /** + * Setter for the {@link ExitStatusExceptionClassifier} that will be used to + * classify any exception that causes a job to fail. + * + * @param exceptionClassifier + */ + public void setExceptionClassifier(ExitStatusExceptionClassifier exceptionClassifier) { + this.exceptionClassifier = exceptionClassifier; + } + + /** + * Public setter for the retryPolicy. + * + * @param retyPolicy the retryPolicy to set + */ + public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; + } + + public void setCommitInterval(int commitInterval) { + this.commitInterval = commitInterval; + } + + /** + * Check mandatory properties (reader and writer). + * + * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(itemReader, "ItemReader must be provided"); + Assert.notNull(itemWriter, "ItemWriter must be provided"); + + applyConfiguration(); + } + + /** + * Apply the configuration by inspecting it to see if it has any relevant + * policy information. + * + * @param step a step + */ + void applyConfiguration() { + + ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); + template.setRetryPolicy(itemProviderRetryPolicy); + + if (retryPolicy != null) { + Assert.state(itemReader instanceof KeyedItemReader, + "ItemReader must be instance of KeyedItemReader to use the retry policy"); + retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); + } + + if (streamManager == null && transactionManager != null) { + streamManager = new SimpleStreamManager(transactionManager); + } + else if (streamManager == null && transactionManager == null) { + throw new IllegalArgumentException("Either StreamManager or TransactionManager must be set"); + } + + if (this.chunkOperations instanceof RepeatTemplate && commitInterval > 0) { + ((RepeatTemplate) chunkOperations).setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); + } + + if (this.chunkOperations instanceof RepeatTemplate && exceptionHandler != null) { + ((RepeatTemplate) chunkOperations).setExceptionHandler(exceptionHandler); + } + + } + + /** + * 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}.
+ * + * @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 { + + JobInstance jobInstance = stepExecution.getJobExecution().getJobInstance(); + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, this); + + boolean isRestart = jobRepository.getStepExecutionCount(jobInstance, this) > 0 ? true : false; + + ExitStatus status = ExitStatus.FAILED; + + try { + + stepExecution.setStartTime(new Date(System.currentTimeMillis())); + // 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); + StepSynchronizationManager.register(stepContext); + 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()); + + if (isSaveExecutionContext() && isRestart && lastStepExecution != null) { + stepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); + } + else { + stepExecution.setExecutionContext(new ExecutionContext()); + } + + // Open the stream manager *after* the execution context is fixed in + // the step, otherwise it will not be the same reference that is + // updated by the streams. TODO: this is a little fragile - maybe + // StreamManager.update() should accept the context as a parameter. + streamManager.open(stepExecution.getExecutionContext()); + + status = stepOperations.iterate(new RepeatCallback() { + + 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; + + TransactionStatus transaction = streamManager.getTransaction(); + + try { + itemReader.mark(); + result = processChunk(contribution); + + contribution.incrementCommitCount(); + + // If the step operations are asynchronous then we need + // to synchronize changes to the step execution (at a + // minimum). + synchronized (stepExecution) { + + // Apply the contribution to the step + // only if chunk was successful + stepExecution.apply(contribution); + + streamManager.update(); + jobRepository.saveOrUpdate(stepExecution); + + } + + itemReader.mark(); + itemWriter.flush(); + streamManager.commit(transaction); + + } + 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. + */ + synchronized (stepExecution) { + stepExecution.rollback(); + } + try { + itemReader.reset(); + itemWriter.clear(); + streamManager.rollback(transaction); + } + 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); + } + if (t instanceof RuntimeException) { + throw (RuntimeException) t; + } + else { + throw new RuntimeException(t); + } + } + + // Check for interruption after transaction as well, so that + // the interrupted exception is correctly propagated up to + // caller + interruptionPolicy.checkInterrupted(context); + + return result; + + } + }); + + updateStatus(stepExecution, BatchStatus.COMPLETED); + } + 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) { + updateStatus(stepExecution, BatchStatus.UNKNOWN); + throw (ResetFailedException) e; + } + else { + updateStatus(stepExecution, BatchStatus.FAILED); + throw e; + } + + } + finally { + stepExecution.setExitStatus(status); + stepExecution.setEndTime(new Date(System.currentTimeMillis())); + try { + jobRepository.saveOrUpdate(stepExecution); + streamManager.close(); + } + 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(); + } + } + + } + + /** + * + */ + private void possiblyRegisterStreams() { + if (itemReader instanceof ItemStream) { + ItemStream stream = (ItemStream) itemReader; + streamManager.register(stream); + } + if (itemWriter instanceof ItemStream) { + ItemStream stream = (ItemStream) itemWriter; + streamManager.register(stream); + } + } + + /** + * 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. + * @return true if there is more data to process. + */ + ExitStatus processChunk(final StepContribution contribution) { + ExitStatus result = chunkOperations.iterate(new RepeatCallback() { + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + if (contribution.isTerminateOnly()) { + context.setTerminateOnly(); + } + // check for interruption before each item as well + interruptionPolicy.checkInterrupted(context); + ExitStatus exitStatus = doProcessing(contribution); + contribution.incrementTaskCount(); + // check for interruption after each item as well + interruptionPolicy.checkInterrupted(context); + return exitStatus; + } + }); + return result; + } + + /** + * 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.
+ * + * 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 + * @return boolean if there is more processing to do + * @throws Exception if there is an error + */ + private ExitStatus doProcessing(StepContribution contribution) throws Exception { + ExitStatus exitStatus = ExitStatus.CONTINUABLE; + + try { + + exitStatus = execute(); + + } + catch (Exception e) { + if (getItemSkipPolicy().shouldSkip(e, contribution)) { + skip(); + } + else { + // Rethrow so that outer transaction is rolled back properly + throw e; + } + } + + return exitStatus; + } + + /** + * 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() + */ + private ExitStatus execute() throws Exception { + + if (retryCallback == null) { + Object item = null; + try { + item = itemReader.read(); + } + catch (Exception ex) { + getItemFailureHandler().handleReadFailure(ex); + throw ex; + } + if (item == null) { + return ExitStatus.FINISHED; + } + try { + itemWriter.write(item); + } + catch (Exception e) { + + getItemFailureHandler().handleWriteFailure(item, e); + // Re-throw the exception so that the surrounding transaction + // rolls back if there is one + throw e; + } + return ExitStatus.CONTINUABLE; + } + + return new ExitStatus(template.execute(retryCallback) != null); + + } + + /** + * 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() + */ + private void skip() { + if (retryCallback != null) { + // No need to skip because the recoverer will take any action + // necessary. + return; + } + if (this.itemReader instanceof Skippable) { + ((Skippable) this.itemReader).skip(); + } + if (this.itemWriter instanceof Skippable) { + ((Skippable) this.itemWriter).skip(); + } + } + + /** + * Convenience method to update the status in all relevant places. + * + * @param stepInstance the current step + * @param stepExecution the current stepExecution + * @param status the status to set + */ + private void updateStatus(StepExecution stepExecution, BatchStatus status) { + stepExecution.setStatus(status); + try { + jobRepository.saveOrUpdate(stepExecution); + } + catch (Exception e) { + logger.error("Failed to update step execution with status: probably fatal.", e); + } + + } +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java index 2039a0dca..a214da05e 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/LimitCheckingItemSkipPolicy.java @@ -17,9 +17,15 @@ package org.springframework.batch.execution.step.support; import java.io.FileNotFoundException; import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; import java.util.List; +import java.util.Map; +import org.springframework.batch.common.ExceptionClassifier; +import org.springframework.batch.common.SubclassExceptionClassifier; import org.springframework.batch.core.domain.ItemSkipPolicy; +import org.springframework.batch.core.domain.Step; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.FlatFileParsingException; @@ -44,9 +50,14 @@ import org.springframework.batch.io.exception.FlatFileParsingException; */ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { - private final int skipLimit; + /** + * Label for classifying skippable exceptions. + */ + private static final String SKIP = "skip"; - private final List skippableExceptions; + private final int skipLimit; + + private ExceptionClassifier exceptionClassifier; public LimitCheckingItemSkipPolicy(int skipLimit) { this(skipLimit, new ArrayList(0)); @@ -54,7 +65,14 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { public LimitCheckingItemSkipPolicy(int skipLimit, List skippableExceptions) { this.skipLimit = skipLimit; - this.skippableExceptions = skippableExceptions; + SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier(); + Map typeMap = new HashMap(); + for (Iterator iterator = skippableExceptions.iterator(); iterator.hasNext();) { + Class throwable = (Class) iterator.next(); + typeMap.put(throwable, SKIP); + } + exceptionClassifier.setTypeMap(typeMap); + this.exceptionClassifier = exceptionClassifier; } /** @@ -66,7 +84,7 @@ public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { * will be thrown. */ public boolean shouldSkip(Exception ex, StepContribution stepContribution){ - if(skippableExceptions.contains(ex.getClass())){ + if(exceptionClassifier.classify(ex).equals(SKIP)){ if(stepContribution.getSkipCount() < skipLimit){ stepContribution.incrementSkipCount(); return true; diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java index af5dcdadc..13dc0f98b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java @@ -1,193 +1,192 @@ -/* - * 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.launch; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; - -import junit.framework.TestCase; - -import org.springframework.batch.core.domain.BatchStatus; -import org.springframework.batch.core.domain.ItemFailureHandler; -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.execution.job.simple.SimpleJob; -import org.springframework.batch.execution.repository.SimpleJobRepository; -import org.springframework.batch.execution.repository.dao.MapJobDao; -import org.springframework.batch.execution.repository.dao.MapStepDao; -import org.springframework.batch.execution.step.AbstractStep; -import org.springframework.batch.execution.step.ItemOrientedStep; -import org.springframework.batch.execution.step.support.NeverSkipItemSkipPolicy; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.reader.ListItemReader; -import org.springframework.batch.item.writer.AbstractItemWriter; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.exception.handler.ExceptionHandler; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; - -public class SimpleJobTests extends TestCase { - - private List recovered = new ArrayList(); - - private SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapJobDao(), new MapStepDao()); - - private List processed = new ArrayList(); - - private ItemWriter processor = new AbstractItemWriter() { - public void write(Object data) throws Exception { - processed.add((String) data); - } - }; - - private ItemReader provider; - - private SimpleJob job = new SimpleJob();; - - protected void setUp() throws Exception { - super.setUp(); - job.setJobRepository(repository); - } - - private AbstractStep getStep(String arg) throws Exception { - return getStep(new String[] { arg }); - } - - private AbstractStep getStep(String arg0, String arg1) throws Exception { - return getStep(new String[] { arg0, arg1 }); - } - - private ItemOrientedStep getStep(String[] args) throws Exception { - ItemOrientedStep step = new ItemOrientedStep(); - List items = TransactionAwareProxyFactory.createTransactionalList(); - items.addAll(Arrays.asList(args)); - provider = new ListItemReader(items); -// step.setItemRecoverer(new ItemRecoverer() { -// public boolean recover(Object item, Throwable cause) { -// recovered.add(item); -// assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); -// return true; -// } -// }); - step.setItemReader(provider); - step.setItemWriter(processor); - step.setJobRepository(repository); - step.setTransactionManager(new ResourcelessTransactionManager()); - step.setName("stepName"); - step.afterPropertiesSet(); - return step; - } - - public void testSimpleJob() throws Exception { - - job.setSteps(new ArrayList()); - AbstractStep step = getStep("foo", "bar"); - job.addStep(step); - step = getStep("spam"); - job.addStep(step); - - JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance(); - - JobExecution jobExecutionContext = new JobExecution(jobInstance); - - job.execute(jobExecutionContext); - assertEquals(BatchStatus.COMPLETED, jobExecutionContext.getStatus()); - assertEquals(3, processed.size()); - assertTrue(processed.contains("foo")); - } - - public void testSimpleJobWithRecovery() throws Exception { - - final List throwables = new ArrayList(); - - RepeatTemplate chunkOperations = new RepeatTemplate(); - // Always handle the exception a check it is the right one... - chunkOperations.setExceptionHandler(new ExceptionHandler() { - public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException { - throwables.add(throwable); - assertEquals("Error!", throwable.getMessage()); - } - }); - - /* - * Each message fails once and the chunk (size=1) "rolls back"; then it - * is recovered ("skipped") on the second attempt (see retry policy - * definition above)... - */ - ItemOrientedStep step = getStep(new String[] { "foo", "bar", "spam" }); - - -// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); -// RepeatOperationsStep step = new RepeatOperationsStep(); - step.setChunkOperations(chunkOperations); - step.setItemWriter(new AbstractItemWriter() { - public void write(Object data) throws Exception { - throw new RuntimeException("Error!"); - } - }); - step.setItemFailureHandler(new ItemFailureHandler(){ - - public void handleReadFailure(Exception ex) { - recovered.add(ex); - } - - public void handleWriteFailure(Object item, Exception ex) { - recovered.add(ex); - } - - }); - step.afterPropertiesSet(); - job.setSteps(Collections.singletonList(step)); - - JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); - job.execute(jobExecution); - - assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - assertEquals(0, processed.size()); - // provider should be exhausted - assertEquals(null, provider.read()); - assertEquals(3, recovered.size()); - } - - public void testExceptionTerminates() throws Exception { -// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); - AbstractStep step = getStep(new String[] { "foo", "bar", "spam" }); - step.setItemWriter(new AbstractItemWriter() { - public void write(Object data) throws Exception { - throw new RuntimeException("Foo"); - } - }); - step.afterPropertiesSet(); - job.setSteps(Collections.singletonList(step)); - - JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); - try { - job.execute(jobExecution); - fail("Expected RuntimeException"); - } - catch (RuntimeException e) { - assertEquals("Foo", e.getMessage()); - // expected - } - assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); - } -} +/* + * 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.launch; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import junit.framework.TestCase; + +import org.springframework.batch.core.domain.BatchStatus; +import org.springframework.batch.core.domain.ItemFailureHandler; +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.execution.job.simple.SimpleJob; +import org.springframework.batch.execution.repository.SimpleJobRepository; +import org.springframework.batch.execution.repository.dao.MapJobDao; +import org.springframework.batch.execution.repository.dao.MapStepDao; +import org.springframework.batch.execution.step.AbstractStep; +import org.springframework.batch.execution.step.ItemOrientedStep; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.reader.ListItemReader; +import org.springframework.batch.item.writer.AbstractItemWriter; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.exception.handler.ExceptionHandler; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; + +public class SimpleJobTests extends TestCase { + + private List recovered = new ArrayList(); + + private SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapJobDao(), new MapStepDao()); + + private List processed = new ArrayList(); + + private ItemWriter processor = new AbstractItemWriter() { + public void write(Object data) throws Exception { + processed.add((String) data); + } + }; + + private ItemReader provider; + + private SimpleJob job = new SimpleJob();; + + protected void setUp() throws Exception { + super.setUp(); + job.setJobRepository(repository); + } + + private AbstractStep getStep(String arg) throws Exception { + return getStep(new String[] { arg }); + } + + private AbstractStep getStep(String arg0, String arg1) throws Exception { + return getStep(new String[] { arg0, arg1 }); + } + + private ItemOrientedStep getStep(String[] args) throws Exception { + ItemOrientedStep step = new ItemOrientedStep(); + List items = TransactionAwareProxyFactory.createTransactionalList(); + items.addAll(Arrays.asList(args)); + provider = new ListItemReader(items); +// step.setItemRecoverer(new ItemRecoverer() { +// public boolean recover(Object item, Throwable cause) { +// recovered.add(item); +// assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); +// return true; +// } +// }); + step.setItemReader(provider); + step.setItemWriter(processor); + step.setJobRepository(repository); + step.setTransactionManager(new ResourcelessTransactionManager()); + step.setName("stepName"); + step.afterPropertiesSet(); + return step; + } + + public void testSimpleJob() throws Exception { + + job.setSteps(new ArrayList()); + AbstractStep step = getStep("foo", "bar"); + job.addStep(step); + step = getStep("spam"); + job.addStep(step); + + JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance(); + + JobExecution jobExecutionContext = new JobExecution(jobInstance); + + job.execute(jobExecutionContext); + assertEquals(BatchStatus.COMPLETED, jobExecutionContext.getStatus()); + assertEquals(3, processed.size()); + assertTrue(processed.contains("foo")); + } + + public void testSimpleJobWithRecovery() throws Exception { + + final List throwables = new ArrayList(); + + RepeatTemplate chunkOperations = new RepeatTemplate(); + // Always handle the exception a check it is the right one... + chunkOperations.setExceptionHandler(new ExceptionHandler() { + public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException { + throwables.add(throwable); + assertEquals("Error!", throwable.getMessage()); + } + }); + + /* + * Each message fails once and the chunk (size=1) "rolls back"; then it + * is recovered ("skipped") on the second attempt (see retry policy + * definition above)... + */ + ItemOrientedStep step = getStep(new String[] { "foo", "bar", "spam" }); + + +// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); +// RepeatOperationsStep step = new RepeatOperationsStep(); + step.setChunkOperations(chunkOperations); + step.setItemWriter(new AbstractItemWriter() { + public void write(Object data) throws Exception { + throw new RuntimeException("Error!"); + } + }); + step.setItemFailureHandler(new ItemFailureHandler(){ + + public void handleReadFailure(Exception ex) { + recovered.add(ex); + } + + public void handleWriteFailure(Object item, Exception ex) { + recovered.add(ex); + } + + }); + step.afterPropertiesSet(); + job.setSteps(Collections.singletonList(step)); + + JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); + job.execute(jobExecution); + + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + assertEquals(0, processed.size()); + // provider should be exhausted + assertEquals(null, provider.read()); + assertEquals(3, recovered.size()); + } + + public void testExceptionTerminates() throws Exception { +// Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" }); + AbstractStep step = getStep(new String[] { "foo", "bar", "spam" }); + step.setItemWriter(new AbstractItemWriter() { + public void write(Object data) throws Exception { + throw new RuntimeException("Foo"); + } + }); + step.afterPropertiesSet(); + job.setSteps(Collections.singletonList(step)); + + JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); + try { + job.execute(jobExecution); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + assertEquals("Foo", e.getMessage()); + // expected + } + assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); + } +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java index 3ef400175..6539e4537 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java @@ -1,228 +1,224 @@ -/* - * 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.repository.dao; - -import java.io.Serializable; -import java.util.ArrayList; -import java.util.Date; -import java.util.List; - -import org.springframework.batch.core.domain.BatchStatus; -import org.springframework.batch.core.domain.Job; -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.Step; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.dao.OptimisticLockingFailureException; -import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; -import org.springframework.util.ClassUtils; - -/** - * Tests for step persistence (StepInstanceDao and StepExecutionDao). Because it is very reasonable to assume that there is a - * foreign key constraint on the JobId of a step, the JobDao is used to create - * jobs, to have an id for creating steps. - * - * @author Lucas Ward - * @author Dave Syer - */ -public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSourceSpringContextTests { - - protected JobInstanceDao jobInstanceDao; - - protected StepExecutionDao stepExecutionDao; - - protected JobExecutionDao jobExecutionDao; - - protected JobInstance jobInstance; - - protected Step step1; - - protected Step step2; - - protected StepExecution stepExecution; - - protected JobExecution jobExecution; - - protected JobParameters jobParameters = new JobParameters(); - - protected ExecutionContext executionContext; - - public void setJobInstanceDao(JobInstanceDao jobInstanceDao) { - this.jobInstanceDao = jobInstanceDao; - } - - public void setStepExecutionDao(StepExecutionDao stepExecutionDao) { - this.stepExecutionDao = stepExecutionDao; - } - - public void setJobExecutionDao(JobExecutionDao jobExecutionDao) { - this.jobExecutionDao = jobExecutionDao; - } - - /* - * (non-Javadoc) - * @see org.springframework.test.AbstractSingleSpringContextTests#getConfigLocations() - */ - protected String[] getConfigLocations() { - return new String[] { ClassUtils.addResourcePathToPackagePath(getClass(), "sql-dao-test.xml") }; - } - - /* - * (non-Javadoc) - * @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpInTransaction() - */ - protected void onSetUpInTransaction() throws Exception { - Job job = new JobSupport("TestJob"); - jobInstance = jobInstanceDao.createJobInstance(job, jobParameters); - step1 = new StepSupport("TestStep1"); - step2 = new StepSupport("TestStep2"); - jobExecution = new JobExecution(jobInstance); - jobExecutionDao.saveJobExecution(jobExecution); - - stepExecution = new StepExecution(step1, jobExecution, new Long(1)); - stepExecution.setStatus(BatchStatus.STARTED); - stepExecution.setStartTime(new Date(System.currentTimeMillis())); - stepExecutionDao.saveStepExecution(stepExecution); - - executionContext = new ExecutionContext(); - executionContext.putString("1", "testString1"); - executionContext.putString("2", "testString2"); - executionContext.putLong("3", 3); - executionContext.putDouble("4", 4.4); - - - } - - public void testVersionIsNotNullForStepExecution() throws Exception { - int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_EXECUTION where STEP_EXECUTION_ID=" - + stepExecution.getId()); - assertEquals(0, version); - } - - public void testUpdateStepWithExecutionContext() { - stepExecution.setExecutionContext(executionContext); - stepExecutionDao.saveExecutionContext(stepExecution); - ExecutionContext tempAttributes = stepExecutionDao.findExecutionContext(stepExecution); - assertEquals(executionContext, tempAttributes); - } - - public void testSaveStepExecution() { - StepExecution execution = new StepExecution(step2, jobExecution, null); - execution.setStatus(BatchStatus.STARTED); - execution.setStartTime(new Date(System.currentTimeMillis())); - execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, - "java.lang.Exception")); - stepExecutionDao.saveStepExecution(execution); - StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2); - assertNotNull(retrievedExecution); - assertEquals(execution, retrievedExecution); - assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus()); - } - - public void testSaveStepExecutionAndExecutionContext() { - StepExecution execution = new StepExecution(step2, jobExecution, null); - execution.setStatus(BatchStatus.STARTED); - execution.setStartTime(new Date(System.currentTimeMillis())); - execution.setExecutionContext(executionContext); - execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, - "java.lang.Exception")); - stepExecutionDao.saveStepExecution(execution); - stepExecutionDao.saveExecutionContext(execution); - StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2); - assertNotNull(retrievedExecution); - assertEquals(execution, retrievedExecution); - assertEquals(execution.getExecutionContext().getString("1"), retrievedExecution.getExecutionContext().getString("1")); - assertEquals(execution.getExecutionContext().getLong("3"), retrievedExecution.getExecutionContext().getLong("3")); - assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus()); - } - - public void testUpdateStepExecution() { - - stepExecution.setStatus(BatchStatus.COMPLETED); - stepExecution.setEndTime(new Date(System.currentTimeMillis())); - stepExecution.setCommitCount(5); - stepExecution.setTaskCount(5); - stepExecution.setExecutionContext(new ExecutionContext()); - stepExecution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, - "java.lang.Exception")); - stepExecutionDao.updateStepExecution(stepExecution); - StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step1); - assertNotNull(retrievedExecution); - assertEquals(stepExecution, retrievedExecution); - assertEquals(stepExecution.getExitStatus(), retrievedExecution.getExitStatus()); - } - - public void testUpdateStepExecutionWithNullId() { - StepExecution stepExecution = new StepExecution(new StepSupport("testStep"), null, null); - try { - stepExecutionDao.updateStepExecution(stepExecution); - fail("Expected IllegalArgumentException"); - } - catch (IllegalArgumentException ex) { - // expected - } - } - - public void testUpdateStepExecutionVersion() throws Exception { - int before = stepExecution.getVersion().intValue(); - stepExecutionDao.updateStepExecution(stepExecution); - int after = stepExecution.getVersion().intValue(); - assertEquals("StepExecution version not updated", before + 1, after); - } - - public void testUpdateStepExecutionOptimisticLocking() throws Exception { - stepExecution.incrementVersion(); // not really allowed outside dao - // code - try { - stepExecutionDao.updateStepExecution(stepExecution); - fail("Expected OptimisticLockingFailureException"); - } - catch (OptimisticLockingFailureException e) { - // expected - assertTrue("Exception message should contain step execution id: " + e.getMessage(), e.getMessage().indexOf( - "" + stepExecution.getId()) >= 0); - assertTrue("Exception message should contain step execution version: " + e.getMessage(), e.getMessage() - .indexOf("" + stepExecution.getVersion()) >= 0); - } - } - - public void testSaveExecutionContext(){ - - stepExecution.setExecutionContext(executionContext); - stepExecutionDao.saveExecutionContext(stepExecution); - ExecutionContext attributes = stepExecutionDao.findExecutionContext(stepExecution); - assertEquals(executionContext, attributes); - executionContext.putString("newString", "newString"); - executionContext.putLong("newLong", 1); - executionContext.putDouble("newDouble", 2.5); - executionContext.put("newSerializable", "serializableValue"); - stepExecutionDao.updateExecutionContext(stepExecution); - attributes = stepExecutionDao.findExecutionContext(stepExecution); - assertEquals(executionContext, attributes); - } - - public void testGetStepExecution() { - assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, step1)); - } - -} +/* + * 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.repository.dao; + +import java.util.Date; +import org.springframework.batch.core.domain.BatchStatus; +import org.springframework.batch.core.domain.Job; +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.Step; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepSupport; +import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.repeat.ExitStatus; +import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.util.ClassUtils; + +/** + * Tests for step persistence (StepInstanceDao and StepExecutionDao). Because it is very reasonable to assume that there is a + * foreign key constraint on the JobId of a step, the JobDao is used to create + * jobs, to have an id for creating steps. + * + * @author Lucas Ward + * @author Dave Syer + */ +public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSourceSpringContextTests { + + protected JobInstanceDao jobInstanceDao; + + protected StepExecutionDao stepExecutionDao; + + protected JobExecutionDao jobExecutionDao; + + protected JobInstance jobInstance; + + protected Step step1; + + protected Step step2; + + protected StepExecution stepExecution; + + protected JobExecution jobExecution; + + protected JobParameters jobParameters = new JobParameters(); + + protected ExecutionContext executionContext; + + public void setJobInstanceDao(JobInstanceDao jobInstanceDao) { + this.jobInstanceDao = jobInstanceDao; + } + + public void setStepExecutionDao(StepExecutionDao stepExecutionDao) { + this.stepExecutionDao = stepExecutionDao; + } + + public void setJobExecutionDao(JobExecutionDao jobExecutionDao) { + this.jobExecutionDao = jobExecutionDao; + } + + /* + * (non-Javadoc) + * @see org.springframework.test.AbstractSingleSpringContextTests#getConfigLocations() + */ + protected String[] getConfigLocations() { + return new String[] { ClassUtils.addResourcePathToPackagePath(getClass(), "sql-dao-test.xml") }; + } + + /* + * (non-Javadoc) + * @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpInTransaction() + */ + protected void onSetUpInTransaction() throws Exception { + Job job = new JobSupport("TestJob"); + jobInstance = jobInstanceDao.createJobInstance(job, jobParameters); + step1 = new StepSupport("TestStep1"); + step2 = new StepSupport("TestStep2"); + jobExecution = new JobExecution(jobInstance); + jobExecutionDao.saveJobExecution(jobExecution); + + stepExecution = new StepExecution(step1, jobExecution, new Long(1)); + stepExecution.setStatus(BatchStatus.STARTED); + stepExecution.setStartTime(new Date(System.currentTimeMillis())); + stepExecutionDao.saveStepExecution(stepExecution); + + executionContext = new ExecutionContext(); + executionContext.putString("1", "testString1"); + executionContext.putString("2", "testString2"); + executionContext.putLong("3", 3); + executionContext.putDouble("4", 4.4); + + + } + + public void testVersionIsNotNullForStepExecution() throws Exception { + int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_EXECUTION where STEP_EXECUTION_ID=" + + stepExecution.getId()); + assertEquals(0, version); + } + + public void testUpdateStepWithExecutionContext() { + stepExecution.setExecutionContext(executionContext); + stepExecutionDao.saveExecutionContext(stepExecution); + ExecutionContext tempAttributes = stepExecutionDao.findExecutionContext(stepExecution); + assertEquals(executionContext, tempAttributes); + } + + public void testSaveStepExecution() { + StepExecution execution = new StepExecution(step2, jobExecution, null); + execution.setStatus(BatchStatus.STARTED); + execution.setStartTime(new Date(System.currentTimeMillis())); + execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, + "java.lang.Exception")); + stepExecutionDao.saveStepExecution(execution); + StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2); + assertNotNull(retrievedExecution); + assertEquals(execution, retrievedExecution); + assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus()); + } + + public void testSaveStepExecutionAndExecutionContext() { + StepExecution execution = new StepExecution(step2, jobExecution, null); + execution.setStatus(BatchStatus.STARTED); + execution.setStartTime(new Date(System.currentTimeMillis())); + execution.setExecutionContext(executionContext); + execution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, + "java.lang.Exception")); + stepExecutionDao.saveStepExecution(execution); + stepExecutionDao.saveExecutionContext(execution); + StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step2); + assertNotNull(retrievedExecution); + assertEquals(execution, retrievedExecution); + assertEquals(execution.getExecutionContext().getString("1"), retrievedExecution.getExecutionContext().getString("1")); + assertEquals(execution.getExecutionContext().getLong("3"), retrievedExecution.getExecutionContext().getLong("3")); + assertEquals(execution.getExitStatus(), retrievedExecution.getExitStatus()); + } + + public void testUpdateStepExecution() { + + stepExecution.setStatus(BatchStatus.COMPLETED); + stepExecution.setEndTime(new Date(System.currentTimeMillis())); + stepExecution.setCommitCount(5); + stepExecution.setTaskCount(5); + stepExecution.setExecutionContext(new ExecutionContext()); + stepExecution.setExitStatus(new ExitStatus(false, ExitStatusExceptionClassifier.FATAL_EXCEPTION, + "java.lang.Exception")); + stepExecutionDao.updateStepExecution(stepExecution); + StepExecution retrievedExecution = stepExecutionDao.getStepExecution(jobExecution, step1); + assertNotNull(retrievedExecution); + assertEquals(stepExecution, retrievedExecution); + assertEquals(stepExecution.getExitStatus(), retrievedExecution.getExitStatus()); + } + + public void testUpdateStepExecutionWithNullId() { + StepExecution stepExecution = new StepExecution(new StepSupport("testStep"), null, null); + try { + stepExecutionDao.updateStepExecution(stepExecution); + fail("Expected IllegalArgumentException"); + } + catch (IllegalArgumentException ex) { + // expected + } + } + + public void testUpdateStepExecutionVersion() throws Exception { + int before = stepExecution.getVersion().intValue(); + stepExecutionDao.updateStepExecution(stepExecution); + int after = stepExecution.getVersion().intValue(); + assertEquals("StepExecution version not updated", before + 1, after); + } + + public void testUpdateStepExecutionOptimisticLocking() throws Exception { + stepExecution.incrementVersion(); // not really allowed outside dao + // code + try { + stepExecutionDao.updateStepExecution(stepExecution); + fail("Expected OptimisticLockingFailureException"); + } + catch (OptimisticLockingFailureException e) { + // expected + assertTrue("Exception message should contain step execution id: " + e.getMessage(), e.getMessage().indexOf( + "" + stepExecution.getId()) >= 0); + assertTrue("Exception message should contain step execution version: " + e.getMessage(), e.getMessage() + .indexOf("" + stepExecution.getVersion()) >= 0); + } + } + + public void testSaveExecutionContext(){ + + stepExecution.setExecutionContext(executionContext); + stepExecutionDao.saveExecutionContext(stepExecution); + ExecutionContext attributes = stepExecutionDao.findExecutionContext(stepExecution); + assertEquals(executionContext, attributes); + executionContext.putString("newString", "newString"); + executionContext.putLong("newLong", 1); + executionContext.putDouble("newDouble", 2.5); + executionContext.put("newSerializable", "serializableValue"); + stepExecutionDao.updateExecutionContext(stepExecution); + attributes = stepExecutionDao.findExecutionContext(stepExecution); + assertEquals(executionContext, attributes); + } + + public void testGetStepExecution() { + assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, step1)); + } + +} 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 b3aff87ee..081a4b2be 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 @@ -1,576 +1,552 @@ -/* - * 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; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -import junit.framework.TestCase; - -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.JobInterruptedException; -import org.springframework.batch.core.domain.JobParameters; -import org.springframework.batch.core.domain.JobSupport; -import org.springframework.batch.core.domain.Step; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.execution.repository.SimpleJobRepository; -import org.springframework.batch.execution.repository.dao.MapJobDao; -import org.springframework.batch.execution.repository.dao.MapStepDao; -import org.springframework.batch.execution.scope.StepScope; -import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.execution.step.support.JobRepositorySupport; -import org.springframework.batch.execution.step.support.StepInterruptionPolicy; -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; -import org.springframework.batch.item.reader.ListItemReader; -import org.springframework.batch.item.stream.ItemStreamSupport; -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.exception.handler.DefaultExceptionHandler; -import org.springframework.batch.repeat.exception.handler.ExceptionHandler; -import org.springframework.batch.repeat.interceptor.RepeatListenerSupport; -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; -import org.springframework.transaction.TransactionStatus; - -public class ItemOrientedStepTests extends TestCase { - - ArrayList processed = new ArrayList(); - - ItemWriter processor = new AbstractItemWriter() { - public void write(Object data) throws Exception { - processed.add((String) data); - } - }; - - private ItemOrientedStep itemOrientedStep; - - private RepeatTemplate template; - - private JobInstance jobInstance; - - private ResourcelessTransactionManager transactionManager; - - private ItemReader getReader(String[] args) { - return new ListItemReader(Arrays.asList(args)); - } - - - - private AbstractStep getStep(String[] strings) throws Exception { - ItemOrientedStep step = new ItemOrientedStep(); - step.setItemWriter(processor); - step.setItemReader(getReader(strings)); - step.setJobRepository(new JobRepositorySupport()); - step.setTransactionManager(transactionManager); - step.afterPropertiesSet(); - return step; - } - - protected void setUp() throws Exception { - transactionManager = new ResourcelessTransactionManager(); - - itemOrientedStep = (ItemOrientedStep) getStep(new String[] { "foo", "bar", "spam" }); - template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - itemOrientedStep.setStepOperations(template); - // Only process one item: - template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - itemOrientedStep.setChunkOperations(template); - - jobInstance = new JobInstance(new Long(0), new JobParameters(), new JobSupport("FOO")); - - SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); - itemOrientedStep.setStreamManager(streamManager); - - } - - public void testStepExecutor() throws Exception { - - Step step = new StepSupport("stepName"); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - itemOrientedStep.execute(stepExecution); - assertEquals(1, processed.size()); - assertEquals(1, stepExecution.getTaskCount().intValue()); - } - - public void testChunkExecutor() throws Exception { - - template = new RepeatTemplate(); - - // Only process one item: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - itemOrientedStep.setChunkOperations(template); - - Step step = new StepSupport("stepName"); - JobExecution jobExecution = new JobExecution(jobInstance); - - StepExecution stepExecution = new StepExecution(step, jobExecution); - StepContribution contribution = stepExecution.createStepContribution(); - itemOrientedStep.processChunk(contribution); - assertEquals(1, processed.size()); - assertEquals(0, stepExecution.getTaskCount().intValue()); - assertEquals(1, contribution.getTaskCount()); - - } - - public void testStepContextInitialized() throws Exception { - - template = new RepeatTemplate(); - - // Only process one item: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - itemOrientedStep.setChunkOperations(template); - - final Step step = new StepSupport("stepName"); - final JobExecution jobExecution = new JobExecution(jobInstance); - final StepExecution stepExecution = new StepExecution(step, jobExecution); - - itemOrientedStep.setItemReader(new AbstractItemReader() { - public Object read() throws Exception { - assertEquals(step.getName(), stepExecution.getStepName()); - assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); - return "foo"; - } - }); - - itemOrientedStep.execute(stepExecution); - assertEquals(1, processed.size()); - - } - - public void testStepContextInitializedBeforeTasklet() throws Exception { - - template = new RepeatTemplate(); - - // Only process one chunk: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - itemOrientedStep.setStepOperations(template); - - final Step step = new StepSupport("stepName"); - final JobExecution jobExecution = new JobExecution(jobInstance); - jobExecution.setId(new Long(1)); - final StepExecution stepExecution = new StepExecution(step, jobExecution); - - template.setListener(new RepeatListenerSupport() { - 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)); - } - }); - - itemOrientedStep.execute(stepExecution); - assertEquals(1, processed.size()); - - } - - public void testRepository() throws Exception { - - SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapJobDao(), new MapStepDao()); - itemOrientedStep.setJobRepository(repository); - - Step step = new StepSupport("stepName"); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - itemOrientedStep.execute(stepExecution); - assertEquals(1, processed.size()); - } - - public void testIncrementRollbackCount() { - - ItemReader itemReader = new AbstractItemReader() { - - public Object read() throws Exception { - int counter = 0; - counter++; - - if (counter == 1) { - throw new RuntimeException(); - } - - return ExitStatus.CONTINUABLE; - } - - }; - - Step step = new StepSupport("stepName"); - itemOrientedStep.setItemReader(itemReader); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - try { - itemOrientedStep.execute(stepExecution); - } - catch (Exception ex) { - assertEquals(stepExecution.getRollbackCount(), new Integer(1)); - } - - } - - public void testExitCodeDefaultClassification() throws Exception { - - ItemReader itemReader = new AbstractItemReader() { - - public Object read() throws Exception { - int counter = 0; - counter++; - - if (counter == 1) { - throw new RuntimeException(); - } - - return ExitStatus.CONTINUABLE; - } - - }; - - Step step = new StepSupport("stepName"); - itemOrientedStep.setItemReader(itemReader); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - try { - itemOrientedStep.execute(stepExecution); - } - catch (Exception ex) { - ExitStatus status = stepExecution.getExitStatus(); - 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. - */ - public void testNonRestartedJob() throws Exception { - Step step = new StepSupport("stepName"); - MockRestartableItemReader tasklet = new MockRestartableItemReader(); - itemOrientedStep.setItemReader(tasklet); - itemOrientedStep.setSaveExecutionContext(true); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - itemOrientedStep.execute(stepExecution); - - assertFalse(tasklet.isRestoreFromCalled()); - assertTrue(tasklet.isGetExecutionAttributesCalled()); - } - - /* - * make sure a job that has been executed before, and is therefore being - * restarted, is restored. - */ -// public void testRestartedJob() throws Exception { -// String step = "stepName"; -//// step.setStepExecutionCount(1); -// MockRestartableItemReader tasklet = new MockRestartableItemReader(); -// stepExecutor.setItemReader(tasklet); -// stepConfiguration.setSaveExecutionContext(true); -// JobExecution jobExecution = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// -// stepExecution -// .setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); -//// step.setLastExecution(stepExecution); -// stepExecutor.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() { - Step step = new StepSupport("stepName"); - MockRestartableItemReader tasklet = new MockRestartableItemReader(); - itemOrientedStep.setItemReader(tasklet); - itemOrientedStep.setSaveExecutionContext(false); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - try { - itemOrientedStep.execute(stepExecution); - } - catch (Throwable t) { - fail(); - } - - assertFalse(tasklet.isRestoreFromCalled()); - } - - /* - * 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 { - Step step = new StepSupport("stepName"); - itemOrientedStep.setItemReader(new AbstractItemReader() { - public Object read() throws Exception { - return "foo"; - } - }); - itemOrientedStep.setSaveExecutionContext(true); - JobExecution jobExecution = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecution); - - itemOrientedStep.execute(stepExecution); - } - - public void testApplyConfigurationWithExceptionHandler() throws Exception { - AbstractStep stepConfiguration = new StubStep("foo"); - final List list = new ArrayList(); - itemOrientedStep.setStepOperations(new RepeatTemplate() { - public void setExceptionHandler(ExceptionHandler exceptionHandler) { - list.add(exceptionHandler); - } - }); - stepConfiguration.setExceptionHandler(new DefaultExceptionHandler()); - itemOrientedStep.applyConfiguration(stepConfiguration); - assertEquals(1, list.size()); - } - - public void testApplyConfigurationWithZeroSkipLimit() throws Exception { - AbstractStep stepConfiguration = new StubStep("foo"); - stepConfiguration.setSkipLimit(0); - final List list = new ArrayList(); - itemOrientedStep.setStepOperations(new RepeatTemplate() { - public void setExceptionHandler(ExceptionHandler exceptionHandler) { - list.add(exceptionHandler); - } - }); - itemOrientedStep.applyConfiguration(stepConfiguration); - assertEquals(0, list.size()); - } - - public void testApplyConfigurationWithNonZeroSkipLimit() throws Exception { - AbstractStep stepConfiguration = new StubStep("foo"); - stepConfiguration.setSkipLimit(1); - final List list = new ArrayList(); - itemOrientedStep.setStepOperations(new RepeatTemplate() { - public void setExceptionHandler(ExceptionHandler exceptionHandler) { - list.add(exceptionHandler); - } - }); - itemOrientedStep.applyConfiguration(stepConfiguration); - assertEquals(1, list.size()); - } - - public void testStreamManager() throws Exception { - Step step = new StepSupport("stepName"); - itemOrientedStep.setItemReader(new AbstractItemReader() { - public Object read() throws Exception { - return "foo"; - } - }); - itemOrientedStep.setSaveExecutionContext(true); - JobExecution jobExecution = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecution); - - assertEquals(false, stepExecution.getExecutionContext().containsKey("foo")); - - itemOrientedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { - ExecutionContext executionContext; - public void update() { - // TODO Auto-generated method stub - executionContext.putString("foo", "bar"); - } - - public void open(ExecutionContext executionContext) - throws StreamException { - this.executionContext = executionContext; - } - }); - - itemOrientedStep.execute(stepExecution); - - // At least once in that process the statistics service was asked for - // statistics... - assertEquals("bar", stepExecution.getExecutionContext().getString("foo")); - } - - private class MockRestartableItemReader extends ItemStreamSupport implements ItemReader { - - private boolean getExecutionAttributesCalled = false; - - private boolean restoreFromCalled = false; - - private boolean restoreFromCalledWithSomeContext = false; - - private ExecutionContext executionContext; - - public Object read() throws Exception { - StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); - return "item"; - } - - public boolean isRestoreFromCalledWithSomeContext() { - return restoreFromCalledWithSomeContext; - } - - public void update() { - getExecutionAttributesCalled = true; - executionContext.putString("spam", "bucket"); - } - - public boolean isGetExecutionAttributesCalled() { - return getExecutionAttributesCalled; - } - - public boolean isRestoreFromCalled() { - return restoreFromCalled; - } - - 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() { - - StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { - - public void checkInterrupted(RepeatContext context) throws JobInterruptedException { - throw new JobInterruptedException(""); - } - }; - - itemOrientedStep.setInterruptionPolicy(interruptionPolicy); - - ItemReader itemReader = new AbstractItemReader() { - - public Object read() throws Exception { - int counter = 0; - counter++; - - if (counter == 1) { - throw new RuntimeException(); - } - - return ExitStatus.CONTINUABLE; - } - - }; - - itemOrientedStep.setItemReader(itemReader); - - Step step = new StepSupport("stepName"); - JobExecution jobExecutionContext = new JobExecution(jobInstance); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - stepExecution - .setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); - - try { - itemOrientedStep.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 AbstractItemReader() { - public Object read() throws Exception { - // Trigger a rollback - throw new RuntimeException("Foo"); - } - }; - itemOrientedStep.setItemReader(itemReader); - itemOrientedStep.setStreamManager(new SimpleStreamManager(transactionManager) { - public void rollback(TransactionStatus status) { - super.rollback(status); - // Simulate failure on rollback when stream resets - throw new ResetFailedException("Bar"); - } - }); - - Step step = new StepSupport("stepName"); - JobExecution jobExecutionContext = jobInstance.createJobExecution(); - StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - stepExecution - .setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); - - try { - itemOrientedStep.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 StubStep extends AbstractStep{ - - public StubStep(String name) { - super(name); - } - - public void execute(StepExecution stepExecution) - throws JobInterruptedException, BatchCriticalException { - } - - } - -} +/* + * 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; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import junit.framework.TestCase; + +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.JobInterruptedException; +import org.springframework.batch.core.domain.JobParameters; +import org.springframework.batch.core.domain.JobSupport; +import org.springframework.batch.core.domain.Step; +import org.springframework.batch.core.domain.StepContribution; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepSupport; +import org.springframework.batch.execution.repository.SimpleJobRepository; +import org.springframework.batch.execution.repository.dao.MapJobDao; +import org.springframework.batch.execution.repository.dao.MapStepDao; +import org.springframework.batch.execution.scope.StepScope; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.batch.execution.step.support.JobRepositorySupport; +import org.springframework.batch.execution.step.support.LimitCheckingItemSkipPolicy; +import org.springframework.batch.execution.step.support.StepInterruptionPolicy; +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; +import org.springframework.batch.item.reader.ListItemReader; +import org.springframework.batch.item.stream.ItemStreamSupport; +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.exception.handler.DefaultExceptionHandler; +import org.springframework.batch.repeat.exception.handler.ExceptionHandler; +import org.springframework.batch.repeat.interceptor.RepeatListenerSupport; +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; +import org.springframework.transaction.TransactionStatus; + +public class ItemOrientedStepTests extends TestCase { + + ArrayList processed = new ArrayList(); + + ItemWriter processor = new AbstractItemWriter() { + public void write(Object data) throws Exception { + processed.add((String) data); + } + }; + + private ItemOrientedStep itemOrientedStep; + + private RepeatTemplate template; + + private JobInstance jobInstance; + + private ResourcelessTransactionManager transactionManager; + + private ItemReader getReader(String[] args) { + return new ListItemReader(Arrays.asList(args)); + } + + private AbstractStep getStep(String[] strings) throws Exception { + ItemOrientedStep step = new ItemOrientedStep(); + step.setItemWriter(processor); + step.setItemReader(getReader(strings)); + step.setJobRepository(new JobRepositorySupport()); + step.setTransactionManager(transactionManager); + step.afterPropertiesSet(); + return step; + } + + protected void setUp() throws Exception { + transactionManager = new ResourcelessTransactionManager(); + + itemOrientedStep = (ItemOrientedStep) getStep(new String[] { "foo", "bar", "spam" }); + template = new RepeatTemplate(); + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + itemOrientedStep.setStepOperations(template); + // Only process one item: + template = new RepeatTemplate(); + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + itemOrientedStep.setChunkOperations(template); + + jobInstance = new JobInstance(new Long(0), new JobParameters(), new JobSupport("FOO")); + + SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); + itemOrientedStep.setStreamManager(streamManager); + + } + + public void testStepExecutor() throws Exception { + + Step step = new StepSupport("stepName"); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + itemOrientedStep.execute(stepExecution); + assertEquals(1, processed.size()); + assertEquals(1, stepExecution.getTaskCount().intValue()); + } + + public void testChunkExecutor() throws Exception { + + template = new RepeatTemplate(); + + // Only process one item: + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + itemOrientedStep.setChunkOperations(template); + + Step step = new StepSupport("stepName"); + JobExecution jobExecution = new JobExecution(jobInstance); + + StepExecution stepExecution = new StepExecution(step, jobExecution); + StepContribution contribution = stepExecution.createStepContribution(); + itemOrientedStep.processChunk(contribution); + assertEquals(1, processed.size()); + assertEquals(0, stepExecution.getTaskCount().intValue()); + assertEquals(1, contribution.getTaskCount()); + + } + + public void testStepContextInitialized() throws Exception { + + template = new RepeatTemplate(); + + // Only process one item: + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + itemOrientedStep.setChunkOperations(template); + + final Step step = new StepSupport("stepName"); + final JobExecution jobExecution = new JobExecution(jobInstance); + final StepExecution stepExecution = new StepExecution(step, jobExecution); + + itemOrientedStep.setItemReader(new AbstractItemReader() { + public Object read() throws Exception { + assertEquals(step.getName(), stepExecution.getStepName()); + assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); + return "foo"; + } + }); + + itemOrientedStep.execute(stepExecution); + assertEquals(1, processed.size()); + + } + + public void testStepContextInitializedBeforeTasklet() throws Exception { + + template = new RepeatTemplate(); + + // Only process one chunk: + template.setCompletionPolicy(new SimpleCompletionPolicy(1)); + itemOrientedStep.setStepOperations(template); + + final Step step = new StepSupport("stepName"); + final JobExecution jobExecution = new JobExecution(jobInstance); + jobExecution.setId(new Long(1)); + final StepExecution stepExecution = new StepExecution(step, jobExecution); + + template.setListener(new RepeatListenerSupport() { + 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)); + } + }); + + itemOrientedStep.execute(stepExecution); + assertEquals(1, processed.size()); + + } + + public void testRepository() throws Exception { + + SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapJobDao(), new MapStepDao()); + itemOrientedStep.setJobRepository(repository); + + Step step = new StepSupport("stepName"); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + itemOrientedStep.execute(stepExecution); + assertEquals(1, processed.size()); + } + + public void testIncrementRollbackCount() { + + ItemReader itemReader = new AbstractItemReader() { + + public Object read() throws Exception { + int counter = 0; + counter++; + + if (counter == 1) { + throw new RuntimeException(); + } + + return ExitStatus.CONTINUABLE; + } + + }; + + Step step = new StepSupport("stepName"); + itemOrientedStep.setItemReader(itemReader); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + try { + itemOrientedStep.execute(stepExecution); + } + catch (Exception ex) { + assertEquals(stepExecution.getRollbackCount(), new Integer(1)); + } + + } + + public void testExitCodeDefaultClassification() throws Exception { + + ItemReader itemReader = new AbstractItemReader() { + + public Object read() throws Exception { + int counter = 0; + counter++; + + if (counter == 1) { + throw new RuntimeException(); + } + + return ExitStatus.CONTINUABLE; + } + + }; + + Step step = new StepSupport("stepName"); + itemOrientedStep.setItemReader(itemReader); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + try { + itemOrientedStep.execute(stepExecution); + } + catch (Exception ex) { + ExitStatus status = stepExecution.getExitStatus(); + 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. + */ + public void testNonRestartedJob() throws Exception { + Step step = new StepSupport("stepName"); + MockRestartableItemReader tasklet = new MockRestartableItemReader(); + itemOrientedStep.setItemReader(tasklet); + itemOrientedStep.setSaveExecutionContext(true); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + itemOrientedStep.execute(stepExecution); + + assertFalse(tasklet.isRestoreFromCalled()); + assertTrue(tasklet.isGetExecutionAttributesCalled()); + } + + /* + * make sure a job that has been executed before, and is therefore being + * restarted, is restored. + */ + // public void testRestartedJob() throws Exception { + // String step = "stepName"; + // // step.setStepExecutionCount(1); + // MockRestartableItemReader tasklet = new MockRestartableItemReader(); + // stepExecutor.setItemReader(tasklet); + // stepConfiguration.setSaveExecutionContext(true); + // JobExecution jobExecution = new JobExecution(jobInstance); + // StepExecution stepExecution = new StepExecution(step, jobExecution); + // + // stepExecution + // .setExecutionContext(new + // ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); + // // step.setLastExecution(stepExecution); + // stepExecutor.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() { + Step step = new StepSupport("stepName"); + MockRestartableItemReader tasklet = new MockRestartableItemReader(); + itemOrientedStep.setItemReader(tasklet); + itemOrientedStep.setSaveExecutionContext(false); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + try { + itemOrientedStep.execute(stepExecution); + } + catch (Throwable t) { + fail(); + } + + assertFalse(tasklet.isRestoreFromCalled()); + } + + /* + * 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 { + Step step = new StepSupport("stepName"); + itemOrientedStep.setItemReader(new AbstractItemReader() { + public Object read() throws Exception { + return "foo"; + } + }); + itemOrientedStep.setSaveExecutionContext(true); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecution); + + itemOrientedStep.execute(stepExecution); + } + + public void testApplyConfigurationWithExceptionHandler() throws Exception { + final List list = new ArrayList(); + itemOrientedStep.setChunkOperations(new RepeatTemplate() { + public void setExceptionHandler(ExceptionHandler exceptionHandler) { + list.add(exceptionHandler); + } + }); + itemOrientedStep.setExceptionHandler(new DefaultExceptionHandler()); + itemOrientedStep.applyConfiguration(); + assertEquals(1, list.size()); + } + + public void testApplyConfigurationWithZeroSkipLimit() throws Exception { + itemOrientedStep.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(0)); + itemOrientedStep.applyConfiguration(); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + assertEquals(false, itemOrientedStep.getItemSkipPolicy().shouldSkip(new RuntimeException(), + stepExecution.createStepContribution())); + } + + public void testApplyConfigurationWithNonZeroSkipLimit() throws Exception { + itemOrientedStep.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(1, Collections.singletonList(Exception.class))); + itemOrientedStep.applyConfiguration(); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution); + assertEquals(true, itemOrientedStep.getItemSkipPolicy().shouldSkip(new RuntimeException(), + stepExecution.createStepContribution())); + } + + public void testStreamManager() throws Exception { + Step step = new StepSupport("stepName"); + itemOrientedStep.setItemReader(new AbstractItemReader() { + public Object read() throws Exception { + return "foo"; + } + }); + itemOrientedStep.setSaveExecutionContext(true); + JobExecution jobExecution = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecution); + + assertEquals(false, stepExecution.getExecutionContext().containsKey("foo")); + + itemOrientedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { + ExecutionContext executionContext; + + public void update() { + // TODO Auto-generated method stub + executionContext.putString("foo", "bar"); + } + + public void open(ExecutionContext executionContext) throws StreamException { + this.executionContext = executionContext; + } + }); + + itemOrientedStep.execute(stepExecution); + + // At least once in that process the statistics service was asked for + // statistics... + assertEquals("bar", stepExecution.getExecutionContext().getString("foo")); + } + + private class MockRestartableItemReader extends ItemStreamSupport implements ItemReader { + + private boolean getExecutionAttributesCalled = false; + + private boolean restoreFromCalled = false; + + private boolean restoreFromCalledWithSomeContext = false; + + private ExecutionContext executionContext; + + public Object read() throws Exception { + StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); + return "item"; + } + + public boolean isRestoreFromCalledWithSomeContext() { + return restoreFromCalledWithSomeContext; + } + + public void update() { + getExecutionAttributesCalled = true; + executionContext.putString("spam", "bucket"); + } + + public boolean isGetExecutionAttributesCalled() { + return getExecutionAttributesCalled; + } + + public boolean isRestoreFromCalled() { + return restoreFromCalled; + } + + 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() { + + StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { + + public void checkInterrupted(RepeatContext context) throws JobInterruptedException { + throw new JobInterruptedException(""); + } + }; + + itemOrientedStep.setInterruptionPolicy(interruptionPolicy); + + ItemReader itemReader = new AbstractItemReader() { + + public Object read() throws Exception { + int counter = 0; + counter++; + + if (counter == 1) { + throw new RuntimeException(); + } + + return ExitStatus.CONTINUABLE; + } + + }; + + itemOrientedStep.setItemReader(itemReader); + + Step step = new StepSupport("stepName"); + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); + // step.setLastExecution(stepExecution); + + try { + itemOrientedStep.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 AbstractItemReader() { + public Object read() throws Exception { + // Trigger a rollback + throw new RuntimeException("Foo"); + } + }; + itemOrientedStep.setItemReader(itemReader); + itemOrientedStep.setStreamManager(new SimpleStreamManager(transactionManager) { + public void rollback(TransactionStatus status) { + super.rollback(status); + // Simulate failure on rollback when stream resets + throw new ResetFailedException("Bar"); + } + }); + + Step step = new StepSupport("stepName"); + JobExecution jobExecutionContext = jobInstance.createJobExecution(); + StepExecution stepExecution = new StepExecution(step, jobExecutionContext); + + stepExecution.setExecutionContext(new ExecutionContext(PropertiesConverter.stringToProperties("foo=bar"))); + // step.setLastExecution(stepExecution); + + try { + itemOrientedStep.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()); + } + } + +}