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());
+ }
+ }
+
+}