[BATCH-447] Introduced a StepExecutionSynchronizerFactory that returned the proper implementation based on Java5/Backport Concurrent.

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

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).<br/>
+ * 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).<br/>
  * 
- * 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}.<br/>
+	 * 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}.<br/>
 	 * 
 	 * @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;
 		}

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

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