BATCH-1708: optimize TaskletStep

spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java

@@ -97,8 +97,6 @@ public class TaskletStep extends AbstractStep {
 
 	private Tasklet tasklet;
 
-	private final Semaphore semaphore = new Semaphore(1);
-
 	/**
 	 * Default constructor.
 	 */
@@ -245,6 +243,10 @@ public class TaskletStep extends AbstractStep {
 		stream.update(stepExecution.getExecutionContext());
 		getJobRepository().updateExecutionContext(stepExecution);
 
+		// Shared semaphore per step execution, so other step executions can run
+		// in parallel without needing the lock
+		final Semaphore semaphore = createSemaphore();
+
 		stepOperations.iterate(new StepContextRepeatCallback(stepExecution) {
 
 			@Override
@@ -260,7 +262,7 @@ public class TaskletStep extends AbstractStep {
 				RepeatStatus result;
 				try {
 					result = (RepeatStatus) new TransactionTemplate(transactionManager, transactionAttribute)
-							.execute(new ChunkTransactionCallback(chunkContext));
+							.execute(new ChunkTransactionCallback(chunkContext, semaphore));
 				}
 				catch (UncheckedTransactionException e) {
 					// Allow checked exceptions to be thrown inside callback
@@ -281,6 +283,16 @@ public class TaskletStep extends AbstractStep {
 
 	}
 
+	/**
+	 * Extension point mainly for test purposes so that the behaviour of the
+	 * lock can be manipulated to simulate various pathologies.
+	 * 
+	 * @return a semaphore for locking access to the JobRepository
+	 */
+	protected Semaphore createSemaphore() {
+		return new Semaphore(1);
+	}
+
 	protected void close(ExecutionContext ctx) throws Exception {
 		stream.close();
 	}
@@ -312,9 +324,12 @@ public class TaskletStep extends AbstractStep {
 
 		private boolean locked = false;
 
-		public ChunkTransactionCallback(ChunkContext chunkContext) {
+		private final Semaphore semaphore;
+
+		public ChunkTransactionCallback(ChunkContext chunkContext, Semaphore semaphore) {
 			this.chunkContext = chunkContext;
 			this.stepExecution = chunkContext.getStepContext().getStepExecution();
+			this.semaphore = semaphore;
 		}
 
 		@Override

spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java

@@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import java.lang.reflect.Field;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 
@@ -32,7 +31,10 @@ import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.JobSupport;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.core.repository.dao.MapExecutionContextDao;
 import org.springframework.batch.core.repository.dao.MapJobExecutionDao;
 import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
@@ -43,7 +45,6 @@ import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
-import org.springframework.util.ReflectionUtils;
 
 public class StepExecutorInterruptionTests {
 
@@ -56,17 +57,18 @@ public class StepExecutorInterruptionTests {
 	private StepExecution stepExecution;
 
 	private JobRepository jobRepository;
-	
-	private Field semaphore;
 
 	@Before
 	public void setUp() throws Exception {
-
 		jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
 				new MapStepExecutionDao(), new MapExecutionContextDao());
+	}
 
+	private void configureStep(TaskletStep step) throws JobExecutionAlreadyRunningException, JobRestartException,
+			JobInstanceAlreadyCompleteException {
+
+		this.step = step;
 		JobSupport job = new JobSupport();
-		step = new TaskletStep("interruptedStep");
 		job.addStep(step);
 		job.setBeanName("testJob");
 		jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
@@ -77,12 +79,13 @@ public class StepExecutorInterruptionTests {
 			}
 		};
 		stepExecution = new StepExecution(step.getName(), jobExecution);
-		semaphore = ReflectionUtils.findField(step.getClass(), "semaphore");
 	}
 
 	@Test
 	public void testInterruptStep() throws Exception {
 
+		configureStep(new TaskletStep("step"));
+
 		Thread processingThread = createThread(stepExecution);
 
 		RepeatTemplate template = new RepeatTemplate();
@@ -124,6 +127,28 @@ public class StepExecutorInterruptionTests {
 	@Test
 	public void testInterruptOnInterruptedException() throws Exception {
 
+		// This simulates the unlikely sounding, but in practice all too common
+		// in Bamboo situation where the thread is interrupted before the lock
+		// is taken.
+
+		configureStep(new TaskletStep("step") {
+			@Override
+			protected Semaphore createSemaphore() {
+				return new Semaphore(1) {
+
+					@Override
+					public void acquire() throws InterruptedException {
+						Thread.currentThread().interrupt();
+						throw new InterruptedException();
+					}
+
+					@Override
+					public void release() {
+					}
+				};
+			}
+		});
+
 		Thread processingThread = createThread(stepExecution);
 
 		step.setTasklet(new TestingChunkOrientedTasklet<Object>(new ItemReader<Object>() {
@@ -132,24 +157,6 @@ public class StepExecutorInterruptionTests {
 			}
 		}, itemWriter));
 
-		// This simulates the unlikely sounding, but in practice all too common
-		// in Bamboo situation where the thread is interrupted before the lock
-		// is taken.
-
-		ReflectionUtils.makeAccessible(semaphore);
-		ReflectionUtils.setField(semaphore, step, new Semaphore(1) {
-
-			@Override
-			public void acquire() throws InterruptedException {
-				Thread.currentThread().interrupt();
-				throw new InterruptedException();
-			}
-
-			@Override
-			public void release() {
-			}
-		});
-
 		processingThread.start();
 		Thread.sleep(100);
 
@@ -168,27 +175,31 @@ public class StepExecutorInterruptionTests {
 	@Test
 	public void testLockNotReleasedIfChunkFails() throws Exception {
 
+		configureStep(new TaskletStep("step") {
+			@Override
+			protected Semaphore createSemaphore() {
+				return new Semaphore(1) {
+					private boolean locked = false;
+
+					@Override
+					public void acquire() throws InterruptedException {
+						locked = true;
+					}
+
+					@Override
+					public void release() {
+						assertTrue("Lock released before it is acquired", locked);
+					}
+				};
+			}
+		});
+
 		step.setTasklet(new TestingChunkOrientedTasklet<Object>(new ItemReader<Object>() {
 			public Object read() throws Exception {
 				throw new RuntimeException("Planned!");
 			}
 		}, itemWriter));
 
-		ReflectionUtils.makeAccessible(semaphore);
-		ReflectionUtils.setField(semaphore, step, new Semaphore(1) {
-			private boolean locked = false;
-
-			@Override
-			public void acquire() throws InterruptedException {
-				locked = true;
-			}
-
-			@Override
-			public void release() {
-				assertTrue("Lock released before it is acquired", locked);
-			}
-		});
-
 		jobRepository.add(stepExecution);
 		step.execute(stepExecution);