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 c250e4116..29dc33b47 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 @@ -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 diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java index 794c72042..7ee7975c1 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java +++ b/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); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepExecutionSynchronizerSynchronizer.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepExecutionSynchronizerSynchronizer.java new file mode 100644 index 000000000..a087412d1 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java index 9a15a55c8..8b826899d 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java +++ b/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); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java index 1d85245af..81b7a6be3 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java +++ b/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); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java index 149eff84b..5a8683e88 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java +++ b/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); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StepExecutionSynchronizer.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StepExecutionSynchronizer.java new file mode 100644 index 000000000..1655a0c19 --- /dev/null +++ b/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); + +} 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 a0429a04d..65f479309 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 @@ -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); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java index 9d5c58895..34003a911 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java +++ b/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; } }