diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepExecutionSynchronizer.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/BackportConcurrentStepExecutionSynchronizer.java
similarity index 71%
rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepExecutionSynchronizer.java
rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/BackportConcurrentStepExecutionSynchronizer.java
index ba8b61e88..4c552bcf5 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepExecutionSynchronizer.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/BackportConcurrentStepExecutionSynchronizer.java
@@ -20,23 +20,19 @@ import org.springframework.batch.core.StepExecution;
import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;
/**
+ * An implementation of the {@link StepExecutionSynchronizer} that uses Backport Concurrent Utilties.
+ *
* @author Dave Syer
- *
+ * @author Ben Hale
*/
-public class DefaultStepExecutionSynchronizer implements StepExecutionSynchronizer {
+class BackportConcurrentStepExecutionSynchronizer implements StepExecutionSynchronizer {
private Semaphore semaphore = new Semaphore(1);
-
- /* (non-Javadoc)
- * @see org.springframework.batch.execution.step.support.StepExecutionSynchronizer#lock(org.springframework.batch.core.domain.StepExecution)
- */
+
public void lock(StepExecution stepExecution) throws InterruptedException {
semaphore.acquire();
}
- /* (non-Javadoc)
- * @see org.springframework.batch.execution.step.support.StepExecutionSynchronizer#release(org.springframework.batch.core.domain.StepExecution)
- */
public void release(StepExecution stepExecution) {
semaphore.release();
}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemOrientedStep.java
index 8d80bb7ea..e53845de0 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemOrientedStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/ItemOrientedStep.java
@@ -42,20 +42,16 @@ import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
- * 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).
+ * 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.
+ * 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
@@ -84,13 +80,14 @@ public class ItemOrientedStep extends AbstractStep {
private ItemHandler itemHandler;
- private StepExecutionSynchronizer synchronizer = new DefaultStepExecutionSynchronizer();
+ private StepExecutionSynchronizer synchronizer;
/**
* @param name
*/
public ItemOrientedStep(String name) {
super(name);
+ synchronizer = new StepExecutionSyncronizerFactory().getStepExecutionSynchronizer();
}
/**
@@ -121,13 +118,10 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Register each of the streams for callbacks at the appropriate time in the
- * step. The {@link ItemReader} and {@link ItemWriter} are automatically
- * registered, but it doesn't hurt to also register them here. Injected
- * dependencies of the reader and writer are not automatically registered,
- * so if you implement {@link ItemWriter} using delegation to another object
- * which itself is a {@link ItemStream}, you need to register the delegate
- * here.
+ * Register each of the streams for callbacks at the appropriate time in the step. The {@link ItemReader} and
+ * {@link ItemWriter} are automatically registered, but it doesn't hurt to also register them here. Injected
+ * dependencies of the reader and writer are not automatically registered, so if you implement {@link ItemWriter}
+ * using delegation to another object which itself is a {@link ItemStream}, you need to register the delegate here.
*
* @param streams an array of {@link ItemStream} objects.
*/
@@ -138,8 +132,7 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Register a single {@link ItemStream} for callbacks to the stream
- * interface.
+ * Register a single {@link ItemStream} for callbacks to the stream interface.
*
* @param stream
*/
@@ -148,11 +141,9 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Register each of the objects as listeners. If the {@link ItemReader} or
- * {@link ItemWriter} themselves implements this interface they will be
- * registered automatically, but their injected dependencies will not be.
- * This is a good way to get access to job parameters and execution context
- * if the tasklet is parameterised.
+ * Register each of the objects as listeners. If the {@link ItemReader} or {@link ItemWriter} themselves implements
+ * this interface they will be registered automatically, but their injected dependencies will not be. This is a good
+ * way to get access to job parameters and execution context if the tasklet is parameterised.
*
* @param listeners an array of listener objects of known types.
*/
@@ -163,8 +154,7 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Register a step listener for callbacks at the appropriate stages in a
- * step execution.
+ * Register a step listener for callbacks at the appropriate stages in a step execution.
*
* @param listener a {@link StepListener}
*/
@@ -173,9 +163,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * 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}.
+ * 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.
*/
@@ -184,9 +173,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * 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}.
+ * 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.
*/
@@ -195,9 +183,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Setter for the {@link StepInterruptionPolicy}. The policy is used to
- * check whether an external request has been made to interrupt the job
- * execution.
+ * 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}
*/
@@ -206,8 +193,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Setter for the {@link ExitStatusExceptionClassifier} that will be used to
- * classify any exception that causes a job to fail.
+ * Setter for the {@link ExitStatusExceptionClassifier} that will be used to classify any exception that causes a
+ * job to fail.
*
* @param exceptionClassifier
*/
@@ -216,9 +203,8 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * Mostly useful for testing, but could be used to remove dependence on
- * backport concurrency utilities. Public setter for the
- * {@link StepExecutionSynchronizer}.
+ * Mostly useful for testing, but could be used to remove dependence on backport concurrency utilities. Public
+ * setter for the {@link StepExecutionSynchronizer}.
*
* @param synchronizer the {@link StepExecutionSynchronizer} to set
*/
@@ -227,22 +213,18 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * 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}.
+ * 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
+ * @throws RuntimeException if there is an exception during a chunk execution
* @see StepExecutor#execute(StepExecution)
*/
public void execute(final StepExecution stepExecution) throws UnexpectedJobExecutionException,
- JobInterruptedException {
+ JobInterruptedException {
ExitStatus status = ExitStatus.FAILED;
final ExceptionHolder fatalException = new ExceptionHolder();
@@ -274,7 +256,7 @@ public class ItemOrientedStep extends AbstractStep {
ExitStatus result = ExitStatus.CONTINUABLE;
TransactionStatus transaction = transactionManager
- .getTransaction(new DefaultTransactionDefinition());
+ .getTransaction(new DefaultTransactionDefinition());
try {
@@ -287,8 +269,7 @@ public class ItemOrientedStep extends AbstractStep {
// minimum).
try {
synchronizer.lock(stepExecution);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
Thread.currentThread().interrupt();
}
@@ -304,29 +285,26 @@ public class ItemOrientedStep extends AbstractStep {
stream.update(stepExecution.getExecutionContext());
try {
jobRepository.saveOrUpdateExecutionContext(stepExecution);
- }
- catch (Exception e) {
+ } catch (Exception e) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
throw new CommitFailedException(
- "Fatal error detected during save of step execution context", e);
+ "Fatal error detected during save of step execution context", e);
}
try {
itemHandler.mark();
transactionManager.commit(transaction);
- }
- catch (Exception e) {
+ } catch (Exception e) {
fatalException.setException(e);
stepExecution.setStatus(BatchStatus.UNKNOWN);
throw new CommitFailedException("Fatal error detected during commit", e);
}
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
/*
- * Any exception thrown within the transaction should
- * automatically cause the transaction to rollback.
+ * Any exception thrown within the transaction should automatically cause the transaction to
+ * rollback.
*/
stepExecution.rollback();
@@ -334,12 +312,10 @@ public class ItemOrientedStep extends AbstractStep {
itemHandler.reset();
itemHandler.clear();
transactionManager.rollback(transaction);
- }
- catch (Exception e) {
+ } catch (Exception e) {
/*
- * If we already failed to commit, it doesn't help
- * to do this again - it's better to allow the
- * CommitFailedException to propagate
+ * If we already failed to commit, it doesn't help to do this again - it's better to allow
+ * the CommitFailedException to propagate
*/
if (!fatalException.hasException()) {
fatalException.setException(e);
@@ -349,13 +325,11 @@ public class ItemOrientedStep extends AbstractStep {
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
- }
- else {
+ } else {
throw new RuntimeException(t);
}
- }
- finally {
+ } finally {
synchronizer.release(stepExecution);
}
@@ -370,12 +344,10 @@ public class ItemOrientedStep extends AbstractStep {
});
fatalException.setException(updateStatus(stepExecution, BatchStatus.COMPLETED));
- }
- catch (CommitFailedException e) {
+ } catch (CommitFailedException e) {
logger.error("Fatal error detected during commit.");
throw e;
- }
- catch (RuntimeException e) {
+ } catch (RuntimeException e) {
// classify exception so an exit code can be stored.
status = exceptionClassifier.classifyForExitCode(e);
@@ -383,29 +355,24 @@ public class ItemOrientedStep extends AbstractStep {
if (e.getCause() instanceof JobInterruptedException) {
updateStatus(stepExecution, BatchStatus.STOPPED);
throw (JobInterruptedException) e.getCause();
- }
- else if (!fatalException.hasException()) {
+ } else if (!fatalException.hasException()) {
try {
status = status.and(listener.onErrorInStep(stepExecution, e));
- }
- catch (RuntimeException ex) {
+ } catch (RuntimeException ex) {
logger.error("Unexpected error in listener on error in step.", ex);
}
updateStatus(stepExecution, BatchStatus.FAILED);
throw e;
- }
- else {
+ } else {
logger.error("Fatal error detected during rollback caused by underlying exception: ", e);
throw e;
}
- }
- finally {
+ } finally {
try {
status = status.and(listener.afterStep(stepExecution));
- }
- catch (RuntimeException e) {
+ } catch (RuntimeException e) {
logger.error("Unexpected error in listener after step.", e);
}
@@ -414,8 +381,7 @@ public class ItemOrientedStep extends AbstractStep {
try {
jobRepository.saveOrUpdate(stepExecution);
- }
- catch (RuntimeException e) {
+ } catch (RuntimeException e) {
String msg = "Fatal error detected during final save of meta data";
logger.error(msg, e);
if (!fatalException.hasException()) {
@@ -426,10 +392,9 @@ public class ItemOrientedStep extends AbstractStep {
try {
stream.close(stepExecution.getExecutionContext());
- }
- catch (RuntimeException e) {
+ } catch (RuntimeException e) {
String msg = "Fatal error detected during close of streams. "
- + "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
+ + "The job execution completed (possibly unsuccessfully but with consistent meta-data).";
logger.error(msg, e);
if (!fatalException.hasException()) {
fatalException.setException(e);
@@ -439,7 +404,7 @@ public class ItemOrientedStep extends AbstractStep {
if (fatalException.hasException()) {
throw new UnexpectedJobExecutionException("Encountered an error saving batch meta data.",
- fatalException.getException());
+ fatalException.getException());
}
}
@@ -447,13 +412,11 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
- * 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
+ * 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.
+ * @param step the current step containing the {@link Tasklet} with the business logic.
* @return true if there is more data to process.
*/
protected ExitStatus processChunk(final StepContribution contribution) {
@@ -486,17 +449,12 @@ public class ItemOrientedStep extends AbstractStep {
try {
jobRepository.saveOrUpdate(stepExecution);
return null;
- }
- catch (Exception e) {
+ } catch (Exception e) {
return e;
}
}
- /**
- * @author Dave Syer
- *
- */
private static class ExceptionHolder {
private Exception exception;
@@ -505,16 +463,10 @@ public class ItemOrientedStep extends AbstractStep {
return exception != null;
}
- /**
- * @param exception
- */
public void setException(Exception exception) {
this.exception = exception;
}
- /**
- * @return
- */
public Exception getException() {
return this.exception;
}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/JdkConcurrentStepExecutionSynchronizer.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/JdkConcurrentStepExecutionSynchronizer.java
new file mode 100644
index 000000000..442103107
--- /dev/null
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/JdkConcurrentStepExecutionSynchronizer.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2002-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.core.step;
+
+import java.util.concurrent.Semaphore;
+
+import org.springframework.batch.core.StepExecution;
+
+/**
+ * An implementation of the {@link StepExecutionSynchronizer} that uses the Java 5 Concurrent Utilities.
+ *
+ * @author Ben Hale
+ */
+class JdkConcurrentStepExecutionSynchronizer implements StepExecutionSynchronizer {
+
+ private final Semaphore semaphore = new Semaphore(1);
+
+ public void lock(StepExecution stepExecution) throws InterruptedException {
+ semaphore.acquire();
+ }
+
+ public void release(StepExecution stepExecution) {
+ semaphore.release();
+ }
+
+}
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSyncronizerFactory.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSyncronizerFactory.java
new file mode 100644
index 000000000..9f457dce7
--- /dev/null
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSyncronizerFactory.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2002-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.core.step;
+
+import org.springframework.core.JdkVersion;
+import org.springframework.util.ClassUtils;
+
+/**
+ * A factory that properly determines which version of the {@link StepExecutionSynchronizer} to return based on the
+ * availabilty of Java 5 or Backport Concurrent.
+ *
+ * @author Ben Hale
+ */
+class StepExecutionSyncronizerFactory {
+
+ /** Whether the backport-concurrent library is present on the classpath */
+ private static final boolean backportConcurrentAvailable = ClassUtils.isPresent(
+ "edu.emory.mathcs.backport.java.util.concurrent.Semaphore", StepExecutionSyncronizerFactory.class
+ .getClassLoader());
+
+ private final StepExecutionSynchronizer synchronizer;
+
+ StepExecutionSyncronizerFactory() {
+ if (JdkVersion.isAtLeastJava15()) {
+ synchronizer = new JdkConcurrentStepExecutionSynchronizer();
+ } else if (backportConcurrentAvailable) {
+ synchronizer = new BackportConcurrentStepExecutionSynchronizer();
+ } else {
+ throw new IllegalStateException("Cannot create StepExecutionSycnronizer - "
+ + "neither JDK 1.5 nor backport-concurrent available on the classpath");
+ }
+ }
+
+ public StepExecutionSynchronizer getStepExecutionSynchronizer() {
+ return synchronizer;
+ }
+}