From 4aaa88b76de2c8951aa76645e142514398ea6d12 Mon Sep 17 00:00:00 2001 From: dsyer Date: Wed, 3 Mar 2010 12:48:58 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1522: Intermittent failure of FaultTolerantStepFactoryBean in multi-threaded test --- ...tTolerantStepFactoryBeanRollbackTests.java | 2 +- .../FaultTolerantStepFactoryBeanTests.java | 264 ++++++++++++++++++ .../src/test/resources/log4j.properties | 5 +- .../batch/core/StepExecution.java | 9 +- .../support/SimpleJobRepository.java | 5 + .../batch/core/step/tasklet/TaskletStep.java | 20 +- 6 files changed, 287 insertions(+), 18 deletions(-) create mode 100644 spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java index c64283cf1..c886c279c 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java @@ -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; } diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java new file mode 100644 index 000000000..35ac8892d --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java @@ -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 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(); + + 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 committed = new ArrayList(writer.getCommitted()); + Collections.sort(committed); + assertEquals("[1, 2, 3, 4, 5]", committed.toString()); + List processed = new ArrayList(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 { + + 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 { + + private List written = new ArrayList(); + + private Collection failures = Collections.emptySet(); + + private SimpleJdbcTemplate jdbcTemplate; + + public SkipWriterStub(DataSource dataSource) { + jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + public List getCommitted() { + return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='written'", + new ParameterizedRowMapper() { + 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 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 { + + private final Log logger = LogFactory.getLog(getClass()); + + private List processed = new ArrayList(); + + private SimpleJdbcTemplate jdbcTemplate; + + /** + * @param dataSource + */ + public SkipProcessorStub(DataSource dataSource) { + jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + public List getCommitted() { + return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'", + new ParameterizedRowMapper() { + 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; + } + } + +} diff --git a/spring-batch-core-tests/src/test/resources/log4j.properties b/spring-batch-core-tests/src/test/resources/log4j.properties index bdb054942..b1071e86e 100644 --- a/spring-batch-core-tests/src/test/resources/log4j.properties +++ b/spring-batch-core-tests/src/test/resources/log4j.properties @@ -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 diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java index 9afed1ba7..9e787121d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java @@ -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); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java index c0a157e4d..54c40ed37 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java @@ -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(); } } 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 3943f9bee..54ab7102d 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 @@ -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;