Fix thread interruption synchronization issue in CI build

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java

@@ -29,7 +29,9 @@ 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.listener.CompositeStepListener;
+import org.springframework.batch.execution.step.support.DefaultStepExecutionSynchronizerSynchronizer;
 import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier;
+import org.springframework.batch.execution.step.support.StepExecutionSynchronizer;
 import org.springframework.batch.execution.step.support.StepInterruptionPolicy;
 import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy;
 import org.springframework.batch.io.exception.InfrastructureException;
@@ -48,8 +50,6 @@ import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.DefaultTransactionDefinition;
 
-import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;
-
 /**
  * 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
@@ -93,6 +93,8 @@ public class ItemOrientedStep extends AbstractStep {
 
 	private ItemHandler itemHandler;
 
+	private StepExecutionSynchronizer synchronizer = new DefaultStepExecutionSynchronizerSynchronizer();
+
 	/**
 	 * @param name
 	 */
@@ -122,7 +124,7 @@ public class ItemOrientedStep extends AbstractStep {
 	 * Public setter for the {@link ItemHandler}.
 	 * @param itemHandler the {@link ItemHandler} to set
 	 */
-	public void setItemProcessor(ItemHandler itemHandler) {
+	public void setItemHandler(ItemHandler itemHandler) {
 		this.itemHandler = itemHandler;
 	}
 
@@ -220,6 +222,16 @@ public class ItemOrientedStep extends AbstractStep {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 
+	/**
+	 * 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
+	 */
+	public void setSynchronizer(StepExecutionSynchronizer synchronizer) {
+		this.synchronizer = synchronizer;
+	}
+
 	/**
 	 * 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
@@ -266,8 +278,6 @@ public class ItemOrientedStep extends AbstractStep {
 			listener.beforeStep(stepExecution);
 			stream.open(stepExecution.getExecutionContext());
 
-			final Semaphore semaphore = new Semaphore(1);
-
 			status = stepOperations.iterate(new RepeatCallback() {
 
 				public ExitStatus doInIteration(final RepeatContext context) throws Exception {
@@ -292,7 +302,13 @@ public class ItemOrientedStep extends AbstractStep {
 						// If the step operations are asynchronous then we need
 						// to synchronize changes to the step execution (at a
 						// minimum).
-						semaphore.acquire();
+						try {
+							synchronizer.lock(stepExecution);
+						}
+						catch (InterruptedException e) {
+							stepExecution.setStatus(BatchStatus.STOPPED);
+							Thread.currentThread().interrupt();
+						}
 
 						// Apply the contribution to the step
 						// only if chunk was successful
@@ -332,9 +348,7 @@ public class ItemOrientedStep extends AbstractStep {
 						 * commit (e.g. Hibernate flush) so this catch block
 						 * comes outside the transaction.
 						 */
-						synchronized (stepExecution) {
-							stepExecution.rollback();
-						}
+						stepExecution.rollback();
 
 						try {
 							itemHandler.reset();
@@ -350,12 +364,12 @@ public class ItemOrientedStep extends AbstractStep {
 							throw (RuntimeException) t;
 						}
 						else {
-							throw new RuntimeException(t);
+							throw new RuntimeException(t);							
 						}
 
 					}
 					finally {
-						semaphore.release();
+						synchronizer.release(stepExecution);
 					}
 
 					// Check for interruption after transaction as well, so that

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java

@@ -164,7 +164,7 @@ public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAw
 		Assert.notNull(jobRepository, "JobRepository must be provided");
 		Assert.notNull(transactionManager, "TransactionManager must be provided");
 
-		step.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 		step.setTransactionManager(transactionManager);
 		step.setJobRepository(jobRepository);
 		step.setStartLimit(startLimit);

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepExecutionSynchronizerSynchronizer.java

@@ -0,0 +1,44 @@
+/*
+ * 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.support;
+
+import org.springframework.batch.core.domain.StepExecution;
+
+import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class DefaultStepExecutionSynchronizerSynchronizer 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-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java

@@ -218,7 +218,7 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
 		}
 
 		setItemProcessor(itemProcessor);
-		step.setItemProcessor(itemProcessor);
+		step.setItemHandler(itemProcessor);
 
 	}
 	

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java

@@ -130,7 +130,7 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
 		setItemWriter(itemWriter);
 
 		step.setStepListeners(stepListeners);
-		step.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 
 		step.setChunkOperations(chunkOperations);
 		step.setStepOperations(stepOperations);

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java

@@ -109,7 +109,7 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
 			StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
 					retryTemplate, retryCallback);
 
-			step.setItemProcessor(itemProcessor);
+			step.setItemHandler(itemProcessor);
 
 		}
 

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StepExecutionSynchronizer.java

@@ -0,0 +1,43 @@
+/*
+ * 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.support;
+
+import org.springframework.batch.core.domain.StepExecution;
+
+/**
+ * Strategy for blocking while a step execution is being updated.
+ *  
+ * @author Dave Syer
+ *
+ */
+public interface StepExecutionSynchronizer {
+
+	/**
+	 * Lock the step execution, blocking if it has been locked by another thread.
+	 * 
+	 * @param stepExecution the {@link StepExecution} that is in progress
+	 * @throws InterruptedException if the thread is interrupted while waiting
+	 */
+	void lock(StepExecution stepExecution) throws InterruptedException;
+
+	/**
+	 * Release the lock.  Use this in a finally block.
+	 * 
+	 * @param stepExecution
+	 */
+	void release(StepExecution stepExecution);
+
+}

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java

@@ -86,7 +86,7 @@ public class ItemOrientedStepTests extends TestCase {
 
 	private AbstractStep getStep(String[] strings) throws Exception {
 		ItemOrientedStep step = new ItemOrientedStep("stepName");
-		step.setItemProcessor(new SimpleItemHandler(getReader(strings), itemWriter));
+		step.setItemHandler(new SimpleItemHandler(getReader(strings), itemWriter));
 		step.setJobRepository(new JobRepositorySupport());
 		step.setTransactionManager(transactionManager);
 		return step;
@@ -168,7 +168,7 @@ public class ItemOrientedStepTests extends TestCase {
 
 		};
 
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
 
@@ -198,7 +198,7 @@ public class ItemOrientedStepTests extends TestCase {
 
 		};
 
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
 
@@ -217,7 +217,7 @@ public class ItemOrientedStepTests extends TestCase {
 	 */
 	public void testNonRestartedJob() throws Exception {
 		MockRestartableItemReader tasklet = new MockRestartableItemReader();
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(tasklet, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(tasklet, itemWriter));
 		itemOrientedStep.registerStream(tasklet);
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -288,7 +288,7 @@ public class ItemOrientedStepTests extends TestCase {
 	 */
 	public void testNoSaveExecutionAttributesRestartableJob() {
 		MockRestartableItemReader tasklet = new MockRestartableItemReader();
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(tasklet, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(tasklet, itemWriter));
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
 
@@ -308,7 +308,7 @@ public class ItemOrientedStepTests extends TestCase {
 	 * Restartable.
 	 */
 	public void testRestartJobOnNonRestartableTasklet() throws Exception {
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(new AbstractItemReader() {
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(new AbstractItemReader() {
 			public Object read() throws Exception {
 				return "foo";
 			}
@@ -329,7 +329,7 @@ public class ItemOrientedStepTests extends TestCase {
 				executionContext.putString("foo", "bar");
 			}
 		};
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(reader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(reader, itemWriter));
 		itemOrientedStep.registerStream(reader);
 		JobExecution jobExecution = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
@@ -422,7 +422,7 @@ public class ItemOrientedStepTests extends TestCase {
 				return null;
 			}
 		});
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(new MockRestartableItemReader() {
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(new MockRestartableItemReader() {
 			public Object read() throws Exception {
 				throw new RuntimeException("FOO");
 			}
@@ -448,7 +448,7 @@ public class ItemOrientedStepTests extends TestCase {
 				executionContext.putString("foo", "bar");
 			}
 		};
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(reader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(reader, itemWriter));
 		itemOrientedStep.setStreams(new ItemStream[] {reader});
 		JobExecution jobExecution = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
@@ -488,7 +488,7 @@ public class ItemOrientedStepTests extends TestCase {
 
 		};
 
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -516,7 +516,7 @@ public class ItemOrientedStepTests extends TestCase {
 				throw new RuntimeException("Foo");
 			}
 		};
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -543,7 +543,7 @@ public class ItemOrientedStepTests extends TestCase {
 				throw new RuntimeException("Foo");
 			}
 		};
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 		itemOrientedStep.setTransactionManager(new ResourcelessTransactionManager() {
 			protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
 				// Simulate failure on rollback when stream resets
@@ -639,7 +639,7 @@ public class ItemOrientedStepTests extends TestCase {
 				throw new RuntimeException("Bar");
 			}
 		};
-		itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
+		itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 		itemOrientedStep.registerStream(itemReader);
 
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java

@@ -31,7 +31,6 @@ import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
 import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
 import org.springframework.batch.execution.step.ItemOrientedStep;
 import org.springframework.batch.item.reader.AbstractItemReader;
-import org.springframework.batch.item.reader.ItemReaderAdapter;
 import org.springframework.batch.item.writer.AbstractItemWriter;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -45,12 +44,15 @@ public class StepExecutorInterruptionTests extends TestCase {
 
 	private AbstractItemWriter itemWriter;
 
+	private StepExecution stepExecution;
+
 	public void setUp() throws Exception {
 		MapJobInstanceDao.clear();
 		MapJobExecutionDao.clear();
 		MapStepExecutionDao.clear();
 
-		JobRepository jobRepository = new SimpleJobRepository(new MapJobInstanceDao(),  new MapJobExecutionDao(), new MapStepExecutionDao());
+		JobRepository jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
+				new MapStepExecutionDao());
 
 		JobSupport jobConfiguration = new JobSupport();
 		step = new ItemOrientedStep("interruptedStep");
@@ -63,13 +65,85 @@ public class StepExecutorInterruptionTests extends TestCase {
 			public void write(Object item) throws Exception {
 			}
 		};
-		step.setItemProcessor(new SimpleItemHandler(new ItemReaderAdapter(), itemWriter));
+		step.setItemHandler(new SimpleItemHandler(new AbstractItemReader() {
+			public Object read() throws Exception {
+				return null;
+			}
+		}, itemWriter));
+		stepExecution = new StepExecution(step, jobExecution);
 	}
 
 	public void testInterruptChunk() throws Exception {
 
-		final StepExecution stepExecution = new StepExecution(step, jobExecution);
-		step.setItemProcessor(new SimpleItemHandler(new AbstractItemReader() {
+		Thread processingThread = createThread(stepExecution);
+
+		processingThread.start();
+		Thread.sleep(100);
+		processingThread.interrupt();
+
+		int count = 0;
+		while (processingThread.isAlive() && count < 1000) {
+			Thread.sleep(20);
+			count++;
+		}
+
+		assertTrue("Timed out waiting for step to be interrupted.", count < 1000);
+		assertFalse(processingThread.isAlive());
+		assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
+
+	}
+
+	public void testInterruptStep() throws Exception {
+		RepeatTemplate template = new RepeatTemplate();
+		// N.B, If we don't set the completion policy it might run forever
+		template.setCompletionPolicy(new SimpleCompletionPolicy(2));
+		step.setChunkOperations(template);
+		testInterruptChunk();
+	}
+
+	public void testInterruptOnInterruptedException() throws Exception {
+
+		Thread processingThread = createThread(stepExecution);
+
+		step.setItemHandler(new SimpleItemHandler(new AbstractItemReader() {
+			public Object read() throws Exception {
+				return null;
+			}
+		}, 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.
+		step.setSynchronizer(new StepExecutionSynchronizer() {
+			public void lock(StepExecution stepExecution) throws InterruptedException {
+				Thread.currentThread().interrupt();
+				throw new InterruptedException();
+			}
+
+			public void release(StepExecution stepExecution) {
+			}
+		});
+
+		processingThread.start();
+		Thread.sleep(100);
+
+		int count = 0;
+		while (processingThread.isAlive() && count < 1000) {
+			Thread.sleep(20);
+			count++;
+		}
+
+		assertTrue("Timed out waiting for step to be interrupted.", count < 1000);
+		assertFalse(processingThread.isAlive());
+		assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
+
+	}
+
+	/**
+	 * @return
+	 */
+	private Thread createThread(final StepExecution stepExecution) {
+		step.setItemHandler(new SimpleItemHandler(new AbstractItemReader() {
 			public Object read() throws Exception {
 				// do something non-trivial (and not Thread.sleep())
 				double foo = 1;
@@ -96,29 +170,7 @@ public class StepExecutorInterruptionTests extends TestCase {
 				}
 			}
 		};
-
-		processingThread.start();
-
-		Thread.sleep(100);
-
-		processingThread.interrupt();
-
-		int count = 0;
-		while (processingThread.isAlive() && count < 1000) {
-			Thread.sleep(20);
-			count++;
-		}
-
-		assertFalse(processingThread.isAlive());
-		assertEquals(BatchStatus.STOPPED, stepExecution.getStatus());
-	}
-
-	public void testInterruptStep() throws Exception {
-		RepeatTemplate template = new RepeatTemplate();
-		// N.B, If we don't set the completion policy it might run forever
-		template.setCompletionPolicy(new SimpleCompletionPolicy(2));
-		step.setChunkOperations(template);
-		testInterruptChunk();
+		return processingThread;
 	}
 
 }