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

RESOLVED - issue BATCH-1522: Intermittent failure of FaultTolerantStepFactoryBean in multi-threaded test

Browse Source
This commit is contained in:
dsyer
2010-03-03 12:48:58 +00:00
parent 17daf7ab07
commit 4aaa88b76d
6 changed files with 287 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -270,7 +270,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
public String process(String item) throws Exception {
processed.add(item);
logger.info("Processed item: "+item);
logger.debug("Processed item: "+item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}

264
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java Normal file
View File

@@ -0,0 +1,264 @@
package org.springframework.batch.core.test.step;
import static org.junit.Assert.assertEquals;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.jdbc.SimpleJdbcTestUtils;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Tests for {@link FaultTolerantStepFactoryBean}.
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanTests {
private static final int MAX_COUNT = 1000;
private final Log logger = LogFactory.getLog(getClass());
private FaultTolerantStepFactoryBean<String, String> factory;
private SkipReaderStub reader;
private SkipProcessorStub processor;
private SkipWriterStub writer;
private JobExecution jobExecution;
private StepExecution stepExecution;
@Autowired
private DataSource dataSource;
@Autowired
private JobRepository repository;
@Autowired
private PlatformTransactionManager transactionManager;
@Before
public void setUp() throws Exception {
reader = new SkipReaderStub();
writer = new SkipWriterStub(dataSource);
processor = new SkipProcessorStub(dataSource);
factory = new FaultTolerantStepFactoryBean<String, String>();
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setJobRepository(repository);
factory.setCommitInterval(3);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
}
@Test
public void testUpdatesNoRollback() throws Exception {
SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource);
writer.write(Arrays.asList("foo", "bar"));
processor.process("spam");
assertEquals(3, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
writer.clear();
processor.clear();
assertEquals(0, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
}
@Test
public void testMultithreadedSunnyDay() throws Throwable {
jobExecution = repository.createJobExecution("vanillaJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource);
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
assertEquals(0, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG"));
try {
Step step = (Step) factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
List<String> committed = new ArrayList<String>(writer.getCommitted());
Collections.sort(committed);
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
List<String> processed = new ArrayList<String>(processor.getCommitted());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
catch (Throwable e) {
logger.info("Failed on iteration " + i + " of " + MAX_COUNT);
throw e;
}
}
}
private static class SkipReaderStub implements ItemReader<String> {
private String[] items;
private int counter = -1;
public SkipReaderStub() throws Exception {
super();
}
public void setItems(String... items) {
Assert.isTrue(counter < 0, "Items cannot be set once reading has started");
this.items = items;
}
public void clear() {
counter = -1;
}
public synchronized String read() throws Exception, UnexpectedInputException, ParseException {
counter++;
if (counter >= items.length) {
return null;
}
String item = items[counter];
return item;
}
}
private static class SkipWriterStub implements ItemWriter<String> {
private List<String> written = new ArrayList<String>();
private Collection<String> failures = Collections.emptySet();
private SimpleJdbcTemplate jdbcTemplate;
public SkipWriterStub(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'",
new ParameterizedRowMapper<String>() {
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
written.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='written'");
}
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
written.add(item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "written");
checkFailure(item);
}
}
private void checkFailure(String item) {
if (failures.contains(item)) {
throw new RuntimeException("Planned failure");
}
}
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Log logger = LogFactory.getLog(getClass());
private List<String> processed = new ArrayList<String>();
private SimpleJdbcTemplate jdbcTemplate;
/**
* @param dataSource
*/
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public List<String> getCommitted() {
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
new ParameterizedRowMapper<String>() {
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString(1);
}
});
}
public void clear() {
processed.clear();
jdbcTemplate.update("DELETE FROM ERROR_LOG where STEP_NAME='processed'");
}
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}
}
}

5
spring-batch-core-tests/src/test/resources/log4j.properties
View File

@@ -9,9 +9,6 @@ log4j.category.org.apache.activemq=ERROR
log4j.category.org.springframework.jdbc=INFO
log4j.category.org.springframework.jms=INFO
log4j.category.org.springframework.batch=INFO
#log4j.category.org.springframework.batch.core.scope=DEBUG
log4j.category.org.springframework.batch.core.step.item=INFO
log4j.category.org.springframework.batch.core.step=DEBUG
log4j.category.org.springframework.batch.core.test=DEBUG
# log4j.category.org.springframework.batch.core.test=DEBUG
log4j.category.org.springframework.retry=INFO
# log4j.category.org.springframework.beans.factory.config=TRACE

9
spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java
View File

@@ -27,8 +27,8 @@ import org.springframework.util.Assert;
/**
* Batch domain object representation the execution of a step. Unlike
* {@link JobExecution}, there are additional properties related the
* processing of items such as commit count, etc.
* {@link JobExecution}, there are additional properties related the processing
* of items such as commit count, etc.
*
* @author Lucas Ward
* @author Dave Syer
@@ -507,8 +507,9 @@ public class StepExecution extends Entity {
return super.toString()
+ String.format(
", name=%s, status=%s, exitStatus=%s, readCount=%d, filterCount=%d, writeCount=%d readSkipCount=%d, writeSkipCount=%d"
+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", stepName, status, exitStatus.getExitCode(),
readCount, filterCount, writeCount, readSkipCount, writeSkipCount, processSkipCount, commitCount, rollbackCount);
+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", stepName, status,
exitStatus.getExitCode(), readCount, filterCount, writeCount, readSkipCount, writeSkipCount,
processSkipCount, commitCount, rollbackCount);
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java
View File

@@ -20,6 +20,8 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
@@ -55,6 +57,8 @@ import org.springframework.util.Assert;
*/
public class SimpleJobRepository implements JobRepository {
private static final Log logger = LogFactory.getLog(SimpleJobRepository.class);
private JobInstanceDao jobInstanceDao;
private JobExecutionDao jobExecutionDao;
@@ -238,6 +242,7 @@ public class SimpleJobRepository implements JobRepository {
JobExecution jobExecution = stepExecution.getJobExecution();
jobExecutionDao.synchronizeStatus(jobExecution);
if (jobExecution.isStopping()) {
logger.info("Parent JobExecution is stopped, so passing message on to StepExecution");
stepExecution.setTerminateOnly();
}
}

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

@@ -307,6 +307,8 @@ public class TaskletStep extends AbstractStep {
private Integer oldVersion;
private boolean locked = false;
public ChunkTransactionCallback(ChunkContext chunkContext) {
this.chunkContext = chunkContext;
this.stepExecution = chunkContext.getStepContext().getStepExecution();
@@ -322,10 +324,16 @@ public class TaskletStep extends AbstractStep {
}
}
if (status == TransactionSynchronization.STATUS_UNKNOWN) {
logger.error("Rolling back with transaction in unknown state");
rollback(stepExecution);
stepExecution.upgradeStatus(BatchStatus.UNKNOWN);
stepExecution.setTerminateOnly();
}
// Only release the lock if we acquired it, and release as late as possible
if (locked) {
semaphore.release();
}
locked = false;
}
public Object doInTransaction(TransactionStatus status) {
@@ -338,8 +346,6 @@ public class TaskletStep extends AbstractStep {
chunkListener.beforeChunk();
boolean locked = false;
try {
try {
@@ -367,9 +373,11 @@ public class TaskletStep extends AbstractStep {
locked = true;
}
catch (InterruptedException e) {
logger.error("Thread interrupted while locking for repository update");
stepExecution.setStatus(BatchStatus.STOPPED);
stepExecution.setTerminateOnly();
Thread.currentThread().interrupt();
throw e;
}
// In case we need to push it back to its old value
@@ -396,6 +404,7 @@ public class TaskletStep extends AbstractStep {
catch (Exception e) {
// If we get to here there was a problem saving the step
// execution and we have to fail.
logger.error("JobRepository failure forcing exit with unknown status", e);
stepExecution.upgradeStatus(BatchStatus.UNKNOWN);
stepExecution.setTerminateOnly();
throw e;
@@ -419,13 +428,6 @@ public class TaskletStep extends AbstractStep {
// Allow checked exceptions
throw new TransactionException(e);
}
finally {
// only release the lock if we acquired it
if (locked) {
semaphore.release();
}
locked = false;
}
return result;