Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

CLOSED - BATCH-913: Remove the JdkConcurrent* special classes

Browse Source 
removed the synchronizer interface as well
This commit is contained in:
robokaso
2008-11-11 17:36:05 +00:00
parent fa5a3cafff
commit cb738dc177
4 changed files with 86 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-batch-core/src/main/java/org/springframework/batch/core/step/JdkConcurrentStepExecutionSynchronizer.java
View File

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

43
spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSynchronizer.java
View File

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

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

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

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

@@ -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<Object> 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<Object>(new ItemReader<Object>() {
@@ -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());
}