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 deleted file mode 100644 index ca286b35c..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/JdkConcurrentStepExecutionSynchronizer.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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 - */ -public 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/StepExecutionSynchronizer.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSynchronizer.java deleted file mode 100644 index 29c7ec3cd..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSynchronizer.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.core.step; - -import org.springframework.batch.core.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-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index bf72bc24f..5d81d0974 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -15,6 +15,8 @@ */ package org.springframework.batch.core.step.tasklet; +import java.util.concurrent.Semaphore; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; @@ -26,8 +28,6 @@ import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.scope.StepContext; import org.springframework.batch.core.scope.StepContextRepeatCallback; import org.springframework.batch.core.step.AbstractStep; -import org.springframework.batch.core.step.JdkConcurrentStepExecutionSynchronizer; -import org.springframework.batch.core.step.StepExecutionSynchronizer; import org.springframework.batch.core.step.StepInterruptionPolicy; import org.springframework.batch.core.step.ThreadStepInterruptionPolicy; import org.springframework.batch.item.ExecutionContext; @@ -86,7 +86,7 @@ public class TaskletStep extends AbstractStep { private Tasklet tasklet; - private StepExecutionSynchronizer synchronizer = new JdkConcurrentStepExecutionSynchronizer(); + private Semaphore semaphore = new Semaphore(1); /** * Default constructor. @@ -105,30 +105,25 @@ public class TaskletStep extends AbstractStep { /** * Public setter for the {@link PlatformTransactionManager}. * - * @param transactionManager - * the transaction manager to set + * @param transactionManager the transaction manager to set */ - public void setTransactionManager( - PlatformTransactionManager transactionManager) { + public void setTransactionManager(PlatformTransactionManager transactionManager) { this.transactionManager = transactionManager; } /** * Public setter for the {@link TransactionAttribute}. * - * @param transactionAttribute - * the {@link TransactionAttribute} to set + * @param transactionAttribute the {@link TransactionAttribute} to set */ - public void setTransactionAttribute( - TransactionAttribute transactionAttribute) { + public void setTransactionAttribute(TransactionAttribute transactionAttribute) { this.transactionAttribute = transactionAttribute; } /** * Public setter for the {@link Tasklet}. * - * @param tasklet - * the {@link Tasklet} to set + * @param tasklet the {@link Tasklet} to set */ public void setTasklet(Tasklet tasklet) { this.tasklet = tasklet; @@ -144,8 +139,7 @@ public class TaskletStep extends AbstractStep { * 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. + * @param listeners an array of listener objects of known types. */ public void setStepExecutionListeners(StepExecutionListener[] listeners) { for (int i = 0; i < listeners.length; i++) { @@ -162,8 +156,7 @@ public class TaskletStep extends AbstractStep { * which itself is a {@link ItemStream}, you need to register the delegate * here. * - * @param streams - * an array of {@link ItemStream} objects. + * @param streams an array of {@link ItemStream} objects. */ public void setStreams(ItemStream[] streams) { for (int i = 0; i < streams.length; i++) { @@ -186,8 +179,7 @@ public class TaskletStep extends AbstractStep { * processing. Should be set up by the caller through a factory. Defaults to * a plain {@link RepeatTemplate}. * - * @param stepOperations - * a {@link RepeatOperations} instance. + * @param stepOperations a {@link RepeatOperations} instance. */ public void setStepOperations(RepeatOperations stepOperations) { this.stepOperations = stepOperations; @@ -198,25 +190,12 @@ public class TaskletStep extends AbstractStep { * check whether an external request has been made to interrupt the job * execution. * - * @param interruptionPolicy - * a {@link StepInterruptionPolicy} + * @param interruptionPolicy a {@link StepInterruptionPolicy} */ public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { this.interruptionPolicy = interruptionPolicy; } - /** - * 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 @@ -226,10 +205,9 @@ public class TaskletStep extends AbstractStep { * the current context governing the step execution, which would normally be * available to the caller through the step's {@link ExecutionContext}.
* - * @throws JobInterruptedException - * if the step or a chunk is interrupted - * @throws RuntimeException - * if there is an exception during a chunk execution + * @throws JobInterruptedException if the step or a chunk is interrupted + * @throws RuntimeException if there is an exception during a chunk + * execution * */ @Override @@ -241,14 +219,12 @@ public class TaskletStep extends AbstractStep { stepOperations.iterate(new StepContextRepeatCallback(stepExecution) { @Override - public RepeatStatus doInStepContext(RepeatContext repeatContext, - StepContext stepContext) throws Exception { + public RepeatStatus doInStepContext(RepeatContext repeatContext, StepContext stepContext) throws Exception { StepExecution stepExecution = stepContext.getStepExecution(); ExceptionHolder fatalException = new ExceptionHolder(); - StepContribution contribution = stepExecution - .createStepContribution(); + StepContribution contribution = stepExecution.createStepContribution(); stepExecution.getExecutionContext().clearDirtyFlag(); // Before starting a new transaction, check for @@ -257,8 +233,7 @@ public class TaskletStep extends AbstractStep { RepeatStatus result = RepeatStatus.CONTINUABLE; - TransactionStatus transaction = transactionManager - .getTransaction(transactionAttribute); + TransactionStatus transaction = transactionManager.getTransaction(transactionAttribute); boolean locked = false; @@ -266,7 +241,8 @@ public class TaskletStep extends AbstractStep { try { result = tasklet.execute(contribution, stepContext); - } finally { + } + finally { // Apply the contribution to the step // even if unsuccessful logger.debug("Applying contribution: " + contribution); @@ -280,9 +256,10 @@ public class TaskletStep extends AbstractStep { // to synchronize changes to the step execution (at a // minimum). try { - synchronizer.lock(stepExecution); + semaphore.acquire(); locked = true; - } catch (InterruptedException e) { + } + catch (InterruptedException e) { stepExecution.setStatus(BatchStatus.STOPPED); Thread.currentThread().interrupt(); } @@ -298,48 +275,47 @@ public class TaskletStep extends AbstractStep { stream.update(stepExecution.getExecutionContext()); try { - getJobRepository() - .updateExecutionContext(stepExecution); - } catch (Exception e) { + getJobRepository().updateExecutionContext(stepExecution); + } + catch (Exception e) { fatalException.setException(e); stepExecution.setStatus(BatchStatus.UNKNOWN); - throw new FatalException( - "Fatal error detected during save of step execution context", - e); + throw new FatalException("Fatal error detected during save of step execution context", e); } try { transactionManager.commit(transaction); - } catch (Exception e) { + } + catch (Exception e) { fatalException.setException(e); stepExecution.setStatus(BatchStatus.UNKNOWN); logger.error("Fatal error detected during commit."); - throw new FatalException( - "Fatal error detected during commit", e); + throw new FatalException("Fatal error detected during commit", e); } try { - logger.debug("Saving step execution after commit: " - + stepExecution); + logger.debug("Saving step execution after commit: " + stepExecution); getJobRepository().update(stepExecution); - } catch (Exception e) { + } + catch (Exception e) { fatalException.setException(e); stepExecution.setStatus(BatchStatus.UNKNOWN); - throw new FatalException( - "Fatal error detected during update of step execution", - e); + throw new FatalException("Fatal error detected during update of step execution", e); } - } catch (Error e) { + } + catch (Error e) { processRollback(stepExecution, fatalException, transaction); throw e; - } catch (Exception e) { + } + catch (Exception e) { processRollback(stepExecution, fatalException, transaction); throw e; - } finally { + } + finally { // only release the lock if we acquired it if (locked) { - synchronizer.release(stepExecution); + semaphore.release(); } locked = false; } @@ -361,8 +337,8 @@ public class TaskletStep extends AbstractStep { * @param fatalException * @param transaction */ - private void processRollback(final StepExecution stepExecution, - final ExceptionHolder fatalException, TransactionStatus transaction) { + private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException, + TransactionStatus transaction) { /* * Any exception thrown within the transaction should automatically @@ -372,7 +348,8 @@ public class TaskletStep extends AbstractStep { try { 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 diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java index 5b67f8361..7a8e0cd98 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java @@ -16,7 +16,9 @@ package org.springframework.batch.core.step.tasklet; +import java.lang.reflect.Field; import java.util.List; +import java.util.concurrent.Semaphore; import junit.framework.TestCase; @@ -32,13 +34,12 @@ import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; -import org.springframework.batch.core.step.StepExecutionSynchronizer; -import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.item.ItemReader; 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 extends TestCase { @@ -47,10 +48,12 @@ public class StepExecutorInterruptionTests extends TestCase { private JobExecution jobExecution; private ItemWriter itemWriter; - + private StepExecution stepExecution; - + private JobRepository jobRepository; + + private Field semaphore; public void setUp() throws Exception { MapJobInstanceDao.clear(); @@ -72,10 +75,11 @@ public class StepExecutorInterruptionTests extends TestCase { } }; stepExecution = new StepExecution(step.getName(), jobExecution); + semaphore = ReflectionUtils.findField(step.getClass(), "semaphore"); } public void testInterruptStep() throws Exception { - + Thread processingThread = createThread(stepExecution); RepeatTemplate template = new RepeatTemplate(); @@ -88,7 +92,7 @@ public class StepExecutorInterruptionTests extends TestCase { for (int i = 2; i < 250; i++) { foo = foo * i; } - + if (foo != 1) { return new Double(foo); } @@ -127,16 +131,34 @@ public class StepExecutorInterruptionTests extends TestCase { // 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 { + + ReflectionUtils.makeAccessible(semaphore); + ReflectionUtils.setField(semaphore, step, new Semaphore(1) { + + @Override + public void acquire() throws InterruptedException { Thread.currentThread().interrupt(); throw new InterruptedException(); } - public void release(StepExecution stepExecution) { + @Override + public void release() { } }); + // step.synchronizer = new Semaphore(1) { + // + // @Override + // public void acquire() throws InterruptedException { + // Thread.currentThread().interrupt(); + // throw new InterruptedException(); + // } + // + // @Override + // public void release() { + // } + // }; + processingThread.start(); Thread.sleep(100); @@ -152,7 +174,6 @@ public class StepExecutorInterruptionTests extends TestCase { } - public void testLockNotReleasedIfChunkFails() throws Exception { step.setTasklet(new TestingChunkOrientedTasklet(new ItemReader() { @@ -161,19 +182,24 @@ public class StepExecutorInterruptionTests extends TestCase { } }, itemWriter)); - step.setSynchronizer(new StepExecutionSynchronizer() { + ReflectionUtils.makeAccessible(semaphore); + ReflectionUtils.setField(semaphore, step, new Semaphore(1) { private boolean locked = false; - public void lock(StepExecution stepExecution) throws InterruptedException { + + @Override + public void acquire() throws InterruptedException { locked = true; } - public void release(StepExecution stepExecution) { + + @Override + public void release() { assertTrue("Lock released before it is acquired", locked); } }); - + jobRepository.add(stepExecution); step.execute(stepExecution); - + assertEquals("Planned!", stepExecution.getFailureExceptions().get(0).getMessage()); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); }