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

RESOLVED - issue BATCH-1406: Avoid deadlock with database pool and multithreaded step when throttle limit is too high.

Browse Source
This commit is contained in:
dsyer
2009-09-16 16:54:45 +00:00
parent cd22ad3ffc
commit 0f5d1f2ded
6 changed files with 191 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-batch-core/pom.xml
View File

@@ -49,6 +49,12 @@
<artifactId>commons-lang</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>commons-dbcp</groupId>
<artifactId>commons-dbcp</artifactId>
<optional>true</optional>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymock</artifactId>

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

@@ -261,12 +261,15 @@ public class TaskletStep extends AbstractStep {
boolean locked = false;
Integer oldVersion = null;
boolean committed = true;
try {
try {
try {
result = tasklet.execute(contribution, chunkContext);
if(result == null) {
if (result == null) {
result = RepeatStatus.FINISHED;
}
}
@@ -275,9 +278,29 @@ public class TaskletStep extends AbstractStep {
throw e;
}
}
chunkListener.afterChunk();
}
finally {
// If the step operations are asynchronous then we need
// to synchronize changes to the step execution (at a
// minimum). Take the lock *before* changing the step
// execution.
try {
semaphore.acquire();
locked = true;
}
catch (InterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
Thread.currentThread().interrupt();
}
// In case we need to push it back to its old value
// after a commit fails...
oldVersion = stepExecution.getVersion();
// Apply the contribution to the step
// even if unsuccessful
logger.debug("Applying contribution: " + contribution);
@@ -285,26 +308,26 @@ public class TaskletStep extends AbstractStep {
}
// If the step operations are asynchronous then we need
// to synchronize changes to the step execution (at a
// minimum).
try {
semaphore.acquire();
locked = true;
}
catch (InterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
Thread.currentThread().interrupt();
}
stream.update(stepExecution.getExecutionContext());
try {
// Going to attempt a commit. If it fails this flag will
// stay false and we can use that later.
committed = false;
getJobRepository().updateExecutionContext(stepExecution);
transactionManager.commit(transaction);
stepExecution.incrementCommitCount();
logger.debug("Saving step execution after commit: " + stepExecution);
/*
* The step execution has to be saved before commit
* because otherwise there is a deadlock between the
* data source pool and the semaphore. As long as only
* one connection is used inside the section of this
* callback that is locked with the semaphore, the
* deadlock is avoided.
*/
logger.debug("Saving step execution before commit: " + stepExecution);
getJobRepository().update(stepExecution);
transactionManager.commit(transaction);
committed = true;
}
catch (Exception e) {
throw new FatalException("Fatal failure detected", e);
@@ -352,11 +375,19 @@ public class TaskletStep extends AbstractStep {
throw e;
}
finally {
if (!committed && oldVersion != null) {
// Wah! the commit failed. We need to rescue the step
// execution data.
stepExecution.setVersion(oldVersion);
}
// only release the lock if we acquired it
if (locked) {
semaphore.release();
}
locked = false;
if (!committed) {
getJobRepository().update(stepExecution);
}
}
// Check for interruption after transaction as well, so that

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java
View File

@@ -235,8 +235,8 @@ public class TaskletStepExceptionTests {
taskletStep.execute(stepExecution);
assertEquals(UNKNOWN, stepExecution.getStatus());
assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
assertTrue(stepExecution.getFailureExceptions().contains(exception));
assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
}
private static class ExceptionTasklet implements Tasklet {

128
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java Normal file
View File

@@ -0,0 +1,128 @@
/*
* 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.tasklet;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.commons.dbcp.BasicDataSource;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameter;
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.JobRepository;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
/**
* @author Dave Syer
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/core/repository/dao/sql-dao-test.xml")
public class AsyncChunkOrientedStepIntegrationTests {
private TaskletStep step;
private Job job;
private List<String> written = new ArrayList<String>();
@Autowired
private PlatformTransactionManager transactionManager;
@Autowired
private BasicDataSource dataSource;
@Autowired
private JobRepository jobRepository;
private RepeatTemplate chunkOperations;
private ItemReader<String> getReader(String[] args) {
return new ListItemReader<String>(Arrays.asList(args));
}
@Before
public void onSetUp() throws Exception {
// Force deadlock with batch waiting for DB pool and vice versa
dataSource.setMaxActive(1);
dataSource.setMaxIdle(1);
step = new TaskletStep("stepName");
step.setJobRepository(jobRepository);
step.setTransactionManager(transactionManager);
// Only process one item:
chunkOperations = new RepeatTemplate();
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
job = new JobSupport("FOO");
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setThrottleLimit(2);
repeatTemplate.setTaskExecutor(new SimpleAsyncTaskExecutor());
step.setStepOperations(repeatTemplate);
step.setTransactionManager(transactionManager);
}
@Test
public void testStatus() throws Exception {
step.setTasklet(new TestingChunkOrientedTasklet<String>(getReader(new String[] { "a", "b", "c", "a", "b", "c",
"a", "b", "c", "a", "b", "c" }), new ItemWriter<String>() {
public void write(List<? extends String> data) throws Exception {
written.addAll(data);
}
}, chunkOperations));
JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters(Collections
.singletonMap("run.id", new JobParameter(getClass().getName() + ".1"))));
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
jobRepository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step
.getName());
assertEquals(lastStepExecution, stepExecution);
assertFalse(lastStepExecution == stepExecution);
}
}

31
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java
View File

@@ -20,24 +20,20 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import javax.sql.DataSource;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameter;
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.JobRepository;
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.JobRepositoryFactoryBean;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -67,8 +63,6 @@ public class ChunkOrientedStepIntegrationTests {
private PlatformTransactionManager transactionManager;
@Autowired
private DataSource dataSource;
private JobRepository jobRepository;
private RepeatTemplate chunkOperations;
@@ -79,23 +73,10 @@ public class ChunkOrientedStepIntegrationTests {
@Before
public void onSetUp() throws Exception {
MapJobInstanceDao.clear();
MapStepExecutionDao.clear();
MapJobExecutionDao.clear();
JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
jobRepositoryFactoryBean.setDatabaseType("hsql");
jobRepositoryFactoryBean.setDataSource(dataSource);
jobRepositoryFactoryBean.setTransactionManager(transactionManager);
jobRepositoryFactoryBean.afterPropertiesSet();
jobRepository = (JobRepository) jobRepositoryFactoryBean.getObject();
step = new TaskletStep("stepName");
step.setJobRepository(jobRepository);
step.setTransactionManager(transactionManager);
RepeatTemplate template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setStepOperations(template);
// Only process one item:
chunkOperations = new RepeatTemplate();
@@ -122,7 +103,8 @@ public class ChunkOrientedStepIntegrationTests {
}
}, chunkOperations));
JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters(Collections
.singletonMap("run.id", new JobParameter(getClass().getName() + ".1"))));
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
stepExecution.setExecutionContext(new ExecutionContext() {
@@ -130,11 +112,12 @@ public class ChunkOrientedStepIntegrationTests {
put("foo", "bar");
}
});
jobRepository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus());
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step.getName());
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step
.getName());
assertEquals(lastStepExecution, stepExecution);
assertFalse(lastStepExecution == stepExecution);

4
spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml
View File

@@ -10,9 +10,11 @@
</list>
</property>
</bean>
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource">
<property name="driverClassName" value="org.hsqldb.jdbcDriver" />
<property name="url" value="jdbc:hsqldb:mem:testdb;sql.enforce_strict_size=true" />
<property name="username" value="sa" />
<property name="password" value="" />
</bean>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />