From f7c5e0ff448b97dbe4ef0f6ec1e4ecfee0976b36 Mon Sep 17 00:00:00 2001 From: dsyer Date: Fri, 14 Dec 2007 01:21:29 +0000 Subject: [PATCH] Add parallel processing sample (parallel version of football job) --- .../batch/sample/domain/Player.java | 4 +- .../item/processor/StagingItemProcessor.java | 76 +++++ .../item/provider/StagingItemProvider.java | 263 ++++++++++++++++++ .../src/main/resources/batch.properties | 4 +- .../src/main/resources/jobs/adhocLoopJob.xml | 66 +++-- .../src/main/resources/jobs/parallelJob.xml | 238 ++++++++++++++++ .../sample/FootballJobFunctionalTests.java | 32 ++- .../processor/StagingItemProcessorTests.java | 45 +++ .../provider/StagingItemProviderTests.java | 163 +++++++++++ .../DerbyDataSourceFactoryBean.java | 38 +++ .../item/processor/staging-test-context.xml | 32 +++ 11 files changed, 937 insertions(+), 24 deletions(-) create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/StagingItemProcessor.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemProvider.java create mode 100644 spring-batch-samples/src/main/resources/jobs/parallelJob.xml create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/item/processor/StagingItemProcessorTests.java create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/item/provider/StagingItemProviderTests.java create mode 100644 spring-batch-samples/src/test/java/test/jdbc/datasource/DerbyDataSourceFactoryBean.java create mode 100644 spring-batch-samples/src/test/resources/org/springframework/batch/sample/item/processor/staging-test-context.xml diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/Player.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/Player.java index 2d238d921..5a9594913 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/Player.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/Player.java @@ -1,6 +1,8 @@ package org.springframework.batch.sample.domain; -public class Player { +import java.io.Serializable; + +public class Player implements Serializable { private String ID; private String lastName; diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/StagingItemProcessor.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/StagingItemProcessor.java new file mode 100644 index 000000000..4143f519a --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/StagingItemProcessor.java @@ -0,0 +1,76 @@ +package org.springframework.batch.sample.item.processor; + +import java.io.Serializable; +import java.sql.Types; + +import org.apache.commons.lang.SerializationUtils; +import org.springframework.batch.execution.scope.StepContext; +import org.springframework.batch.execution.scope.StepContextAware; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.jdbc.core.support.JdbcDaoSupport; +import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +public class StagingItemProcessor extends JdbcDaoSupport implements + StepContextAware, ItemProcessor { + + public static final String NEW = "N"; + public static final String DONE = "Y"; + public static final Object WORKING = "W"; + + private DataFieldMaxValueIncrementer incrementer; + private StepContext stepContext; + + /** + * Check mandatory properties. + * + * @see org.springframework.dao.support.DaoSupport#initDao() + */ + protected void initDao() throws Exception { + super.initDao(); + Assert + .notNull( + incrementer, + "DataFieldMaxValueIncrementer is required - set the incrementer property in the " + + ClassUtils + .getShortName(StagingItemProcessor.class)); + } + + /** + * Callback for injection of the step context. + * + * @param stepContext + * the stepContext to set + */ + public void setStepContext(StepContext stepContext) { + this.stepContext = stepContext; + } + + /** + * Setter for the key generator for the staging table. + * + * @param incrementer + * the {@link DataFieldMaxValueIncrementer} to set + */ + public void setIncrementer(DataFieldMaxValueIncrementer incrementer) { + this.incrementer = incrementer; + } + + /** + * Serialize the item to the staging table, and add a NEW processed flag. + * + * @see org.springframework.batch.item.ItemProcessor#process(java.lang.Object) + */ + public void process(Object data) throws Exception { + Long id = new Long(incrementer.nextLongValue()); + Long jobId = stepContext.getStepExecution().getJobExecution().getJobId(); + byte[] blob = SerializationUtils.serialize((Serializable) data); + getJdbcTemplate() + .update( + "INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)", + new Object[] { id, jobId, blob, NEW }, + new int[] { Types.BIGINT, Types.BIGINT, Types.BLOB, Types.CHAR}); + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemProvider.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemProvider.java new file mode 100644 index 000000000..1a101f655 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/provider/StagingItemProvider.java @@ -0,0 +1,263 @@ +package org.springframework.batch.sample.item.provider; + +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import org.apache.commons.lang.SerializationUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.execution.scope.StepContext; +import org.springframework.batch.execution.scope.StepContextAware; +import org.springframework.batch.item.ItemProvider; +import org.springframework.batch.item.ResourceLifecycle; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.sample.item.processor.StagingItemProcessor; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.support.JdbcDaoSupport; +import org.springframework.jdbc.support.lob.DefaultLobHandler; +import org.springframework.jdbc.support.lob.LobHandler; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.util.Assert; + +public class StagingItemProvider extends JdbcDaoSupport implements + ItemProvider, ResourceLifecycle, DisposableBean, + StepContextAware { + + // Key for buffer in transaction synchronization manager + private static final String BUFFER_KEY = StagingItemProvider.class.getName()+".BUFFER"; + + private static Log logger = LogFactory.getLog(StagingItemProvider.class); + + private StepContext stepContext; + + private LobHandler lobHandler = new DefaultLobHandler(); + + private Object lock = new Object(); + + private boolean initialized = false; + + private volatile Iterator keys; + + private final TransactionSynchronization synchronization = new StagingInputTransactionSynchronization(); + + /** + * + * @see org.springframework.batch.io.driving.DrivingQueryInputSource#close() + */ + public void close() { + initialized = false; + keys = null; + if (TransactionSynchronizationManager.hasResource(BUFFER_KEY)) { + TransactionSynchronizationManager.unbindResource(BUFFER_KEY); + } + } + + /** + * @throws Exception + * @see org.springframework.batch.io.driving.DrivingQueryInputSource#destroy() + */ + public void destroy() throws Exception { + close(); + } + + /** + * + * @see org.springframework.batch.io.driving.DrivingQueryInputSource#open() + */ + public void open() { + Assert.state(keys == null || initialized, + "Cannot open an already open StagingItemProvider" + + ", call close() first."); + keys = retrieveKeys().iterator(); + logger.info("keys: " + keys); + registerSynchronization(); + initialized = true; + } + + /** + * Callback for injection of the step context. + * + * @param stepContext + * the stepContext to set + */ + public void setStepContext(StepContext stepContext) { + this.stepContext = stepContext; + } + + private List retrieveKeys() { + + synchronized (lock) { + + return getJdbcTemplate() + .query( + + "SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID", + + new Object[] { + stepContext.getStepExecution() + .getJobExecution().getJobId(), + StagingItemProcessor.NEW }, + + new RowMapper() { + public Object mapRow(ResultSet rs, int rowNum) + throws SQLException { + return new Long(rs.getLong(1)); + } + } + + ); + + } + + } + + public Object getKey(Object item) { + return item; + } + + public Object next() throws Exception { + Long id = read(); + + if (id == null) { + return null; + } + Object result = getJdbcTemplate().queryForObject( + "SELECT VALUE FROM BATCH_STAGING WHERE ID=?", + new Object[] { id }, new RowMapper() { + public Object mapRow(ResultSet rs, int rowNum) + throws SQLException { + byte[] blob = lobHandler.getBlobAsBytes(rs, 1); + return SerializationUtils.deserialize(blob); + } + }); + // Update now - changes will rollback if there is a problem later. + int count = getJdbcTemplate() + .update( + "UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?", + new Object[] { StagingItemProcessor.DONE, id, + StagingItemProcessor.NEW }); + if (count != 1) { + throw new OptimisticLockingFailureException( + "The staging record with ID=" + + id + + " was updated concurrently when trying to mark as complete."); + } + return result; + } + + private Long read() { + if (!initialized) { + open(); + } + + Long key = getBuffer().next(); + if (key == null) { + synchronized (lock) { + if (keys.hasNext()) { + Long next = (Long) keys.next(); + getBuffer().add(next); + key = next; + logger.debug("Retrieved key from list: " + key); + } + } + } else { + logger.debug("Retrieved key from buffer: " + key); + } + return key; + + } + + private StagingBuffer getBuffer() { + if (!TransactionSynchronizationManager.hasResource(BUFFER_KEY)) { + TransactionSynchronizationManager.bindResource(BUFFER_KEY, + new StagingBuffer()); + } + return (StagingBuffer) TransactionSynchronizationManager + .getResource(BUFFER_KEY); + } + + public boolean recover(Object data, Throwable cause) { + return false; + } + + /** + * Register for Synchronization. This method is left protected because + * clients of this class should not be registering for synchronization, but + * rather only subclasses, at the appropriate time, i.e. when they are not + * initialized. + */ + protected void registerSynchronization() { + BatchTransactionSynchronizationManager + .registerSynchronization(synchronization); + } + + /* + * Called when a transaction has been committed. + * + * @see TransactionSynchronization#afterCompletion + */ + protected void transactionCommitted() { + getBuffer().commit(); + } + + /* + * Called when a transaction has been rolled back. + * + * @see TransactionSynchronization#afterCompletion + */ + protected void transactionRolledBack() { + getBuffer().rollback(); + } + + /** + * Encapsulates transaction events handling. + */ + private class StagingInputTransactionSynchronization extends + TransactionSynchronizationAdapter { + public void afterCompletion(int status) { + if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + transactionRolledBack(); + } else if (status == TransactionSynchronization.STATUS_COMMITTED) { + transactionCommitted(); + } + } + } + + private class StagingBuffer { + + private List list = new ArrayList(); + private Iterator iter = new ArrayList().iterator(); + + public Long next() { + if (iter.hasNext()) { + return (Long) iter.next(); + } + return null; + } + + public void add(Long next) { + list.add(next); + } + + public void rollback() { + iter = new ArrayList(list).iterator(); + } + + public void commit() { + list.clear(); + iter = new ArrayList().iterator(); + } + + public String toString() { + return "list=" + list + "; iter.hasNext()=" + iter.hasNext(); + } + } + +} diff --git a/spring-batch-samples/src/main/resources/batch.properties b/spring-batch-samples/src/main/resources/batch.properties index 8fc31f6b8..1e223008c 100644 --- a/spring-batch-samples/src/main/resources/batch.properties +++ b/spring-batch-samples/src/main/resources/batch.properties @@ -1,9 +1,9 @@ # Placeholders batch.* # for HSQLDB: batch.jdbc.driver=org.hsqldb.jdbcDriver -batch.jdbc.url=jdbc:hsqldb:mem:testdb;sql.enforce_strict_size=true +# batch.jdbc.url=jdbc:hsqldb:mem:testdb;sql.enforce_strict_size=true # use this one for a separate server process (so you can inspect the results) -# batch.jdbc.url=jdbc:hsqldb:hsql://localhost:9005/samples +batch.jdbc.url=jdbc:hsqldb:hsql://localhost:9005/samples batch.jdbc.user=sa batch.jdbc.password= batch.schema= diff --git a/spring-batch-samples/src/main/resources/jobs/adhocLoopJob.xml b/spring-batch-samples/src/main/resources/jobs/adhocLoopJob.xml index bd945ca33..c54442800 100644 --- a/spring-batch-samples/src/main/resources/jobs/adhocLoopJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/adhocLoopJob.xml @@ -10,8 +10,8 @@ - - + + @@ -44,19 +44,40 @@ - - - + + + - org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher + + org.springframework.batch.execution.bootstrap.support.ExportedJobLauncher + - + + + + + + + + + + + org.springframework.batch.sample.ExportedJobConfigurationLoader + + + + - @@ -67,15 +88,22 @@ + - - - - + + + + + + @@ -93,7 +121,8 @@ - + @@ -110,7 +139,12 @@ class="org.springframework.core.task.SimpleAsyncTaskExecutor" /> - + - + + + + + \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml new file mode 100644 index 000000000..19c6dd49b --- /dev/null +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -0,0 +1,238 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + SELECT games.player_id, games.year_no, SUM(COMPLETES), + SUM(ATTEMPTS), SUM(PASSING_YARDS), SUM(PASSING_TD), + SUM(INTERCEPTIONS), SUM(RUSHES), SUM(RUSH_YARDS), + SUM(RECEPTIONS), SUM(RECEPTIONS_YARDS), SUM(TOTAL_TD) + from games, players where players.player_id = + games.player_id group by games.player_id, games.year_no + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/FootballJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/FootballJobFunctionalTests.java index 2422316a1..e3d82d0c3 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/FootballJobFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/FootballJobFunctionalTests.java @@ -1,14 +1,36 @@ package org.springframework.batch.sample; -public class FootballJobFunctionalTests extends AbstractValidatingBatchLauncherTests { +import javax.sql.DataSource; + +import org.springframework.batch.sample.item.processor.StagingItemProcessor; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.JdbcTemplate; + +public class FootballJobFunctionalTests extends + AbstractValidatingBatchLauncherTests { + + private JdbcOperations jdbcTemplate; + + public void setDataSource(DataSource dataSource) { + this.jdbcTemplate = new JdbcTemplate(dataSource); + } protected String[] getConfigLocations() { - return new String[] {"jobs/footballJob.xml###"}; + return new String[] { "jobs/parallelJob.xml" }; } - - protected void validatePostConditions() throws Exception { - // TODO Auto-generated method stub + protected void validatePostConditions() throws Exception { + int count; + count = jdbcTemplate.queryForInt( + "SELECT COUNT(*) from BATCH_STAGING where PROCESSED=?", + new Object[] {StagingItemProcessor.NEW}); + assertEquals(0, count); + int total = jdbcTemplate.queryForInt( + "SELECT COUNT(*) from BATCH_STAGING"); + count = jdbcTemplate.queryForInt( + "SELECT COUNT(*) from BATCH_STAGING where PROCESSED=?", + new Object[] {StagingItemProcessor.DONE}); + assertEquals(total, count); } } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/processor/StagingItemProcessorTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/processor/StagingItemProcessorTests.java new file mode 100644 index 000000000..474245a02 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/processor/StagingItemProcessorTests.java @@ -0,0 +1,45 @@ +package org.springframework.batch.sample.item.processor; + +import org.springframework.batch.core.domain.JobExecution; +import org.springframework.batch.core.domain.JobInstance; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.core.runtime.SimpleJobIdentifier; +import org.springframework.batch.execution.scope.SimpleStepContext; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.util.ClassUtils; + +public class StagingItemProcessorTests extends + AbstractTransactionalDataSourceSpringContextTests { + + private StagingItemProcessor processor; + + public void setProcessor(StagingItemProcessor processor) { + this.processor = processor; + } + + protected String[] getConfigLocations() { + return new String[] { ClassUtils.addResourcePathToPackagePath( + StagingItemProcessor.class, "staging-test-context.xml") }; + } + + protected void prepareTestInstance() throws Exception { + SimpleStepContext stepScopeContext = StepSynchronizationManager + .open(); + stepScopeContext.setStepExecution(new StepExecution(new StepInstance( + new Long(11)), new JobExecution(new JobInstance( + new SimpleJobIdentifier("job"), new Long(12))))); + super.prepareTestInstance(); + } + + public void testProcessInsertsNewItem() throws Exception { + int before = getJdbcTemplate().queryForInt( + "SELECT COUNT(*) from BATCH_STAGING"); + processor.process("FOO"); + int after = getJdbcTemplate().queryForInt( + "SELECT COUNT(*) from BATCH_STAGING"); + assertEquals(before + 1, after); + } + +} diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/provider/StagingItemProviderTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/provider/StagingItemProviderTests.java new file mode 100644 index 000000000..f1b4ab64a --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/item/provider/StagingItemProviderTests.java @@ -0,0 +1,163 @@ +package org.springframework.batch.sample.item.provider; + +import org.springframework.batch.core.domain.JobExecution; +import org.springframework.batch.core.domain.JobInstance; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.core.runtime.SimpleJobIdentifier; +import org.springframework.batch.execution.scope.SimpleStepContext; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.batch.repeat.context.RepeatContextSupport; +import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; +import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; +import org.springframework.batch.sample.item.processor.StagingItemProcessor; +import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; +import org.springframework.util.ClassUtils; + +public class StagingItemProviderTests extends + AbstractTransactionalDataSourceSpringContextTests { + + private StagingItemProcessor processor; + private StagingItemProvider provider; + private Long jobId; + + public void setProcessor(StagingItemProcessor processor) { + this.processor = processor; + } + + public void setProvider(StagingItemProvider provider) { + this.provider = provider; + } + + protected String[] getConfigLocations() { + return new String[] { ClassUtils.addResourcePathToPackagePath( + StagingItemProcessor.class, "staging-test-context.xml") }; + } + + protected void prepareTestInstance() throws Exception { + SimpleStepContext stepScopeContext = StepSynchronizationManager.open(); + jobId = new Long(11); + stepScopeContext.setStepExecution(new StepExecution(new StepInstance( + new Long(12)), new JobExecution(new JobInstance( + new SimpleJobIdentifier("job"), jobId)))); + RepeatSynchronizationManager.register(new RepeatContextSupport(null)); + super.prepareTestInstance(); + } + + protected void onSetUpInTransaction() throws Exception { + processor.process("FOO"); + processor.process("BAR"); + processor.process("SPAM"); + processor.process("BUCKET"); + } + + protected void onTearDownAfterTransaction() throws Exception { + provider.close(); + getJdbcTemplate().update("DELETE FROM BATCH_STAGING"); + } + + public void testProviderUpdatesProcessIndicator() throws Exception { + + long id = getJdbcTemplate().queryForLong( + "SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", + new Object[] { jobId }); + String before = (String) getJdbcTemplate().queryForObject( + "SELECT PROCESSED from BATCH_STAGING where ID=?", + new Object[] { new Long(id) }, String.class); + assertEquals(StagingItemProcessor.NEW, before); + + Object item = provider.next(); + assertEquals("FOO", item); + + String after = (String) getJdbcTemplate().queryForObject( + "SELECT PROCESSED from BATCH_STAGING where ID=?", + new Object[] { new Long(id) }, String.class); + assertEquals(StagingItemProcessor.DONE, after); + + } + + public void testUpdateProcessIndicatorAfterCommit() throws Exception { + testProviderUpdatesProcessIndicator(); + setComplete(); + endTransaction(); + startNewTransaction(); + long id = getJdbcTemplate().queryForLong( + "SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", + new Object[] { jobId }); + String before = (String) getJdbcTemplate().queryForObject( + "SELECT PROCESSED from BATCH_STAGING where ID=?", + new Object[] { new Long(id) }, String.class); + assertEquals(StagingItemProcessor.DONE, before); + } + + public void testProviderRollsBackMultipleTimes() throws Exception { + + setComplete(); + endTransaction(); + startNewTransaction(); + // After a rollback we have to resynchronize the TX to simulate a real batch + BatchTransactionSynchronizationManager.resynchronize(); + + int count = getJdbcTemplate().queryForInt( + "SELECT COUNT(*) from BATCH_STAGING where JOB_ID=? AND PROCESSED=?", + new Object[] { jobId, StagingItemProcessor.NEW }); + assertEquals(4, count); + + Object item = provider.next(); + assertEquals("FOO", item); + item = provider.next(); + assertEquals("BAR", item); + + endTransaction(); + startNewTransaction(); + BatchTransactionSynchronizationManager.resynchronize(); + + item = provider.next(); + assertEquals("FOO", item); + item = provider.next(); + assertEquals("BAR", item); + item = provider.next(); + assertEquals("SPAM", item); + + endTransaction(); + startNewTransaction(); + BatchTransactionSynchronizationManager.resynchronize(); + + item = provider.next(); + assertEquals("FOO", item); + + } + + public void testProviderRollsBackProcessIndicator() throws Exception { + + setComplete(); + endTransaction(); + startNewTransaction(); + // After a rollback we have to resynchronize the TX to simulate a real batch + BatchTransactionSynchronizationManager.resynchronize(); + + long id = getJdbcTemplate().queryForLong( + "SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", + new Object[] { jobId }); + String before = (String) getJdbcTemplate().queryForObject( + "SELECT PROCESSED from BATCH_STAGING where ID=?", + new Object[] { new Long(id) }, String.class); + assertEquals(StagingItemProcessor.NEW, before); + + Object item = provider.next(); + assertEquals("FOO", item); + + endTransaction(); + startNewTransaction(); + // After a rollback we have to resynchronize the TX to simulate a real batch + BatchTransactionSynchronizationManager.resynchronize(); + + String after = (String) getJdbcTemplate().queryForObject( + "SELECT PROCESSED from BATCH_STAGING where ID=?", + new Object[] { new Long(id) }, String.class); + assertEquals(StagingItemProcessor.NEW, after); + + item = provider.next(); + assertEquals("FOO", item); + } +} diff --git a/spring-batch-samples/src/test/java/test/jdbc/datasource/DerbyDataSourceFactoryBean.java b/spring-batch-samples/src/test/java/test/jdbc/datasource/DerbyDataSourceFactoryBean.java new file mode 100644 index 000000000..de5db29a3 --- /dev/null +++ b/spring-batch-samples/src/test/java/test/jdbc/datasource/DerbyDataSourceFactoryBean.java @@ -0,0 +1,38 @@ +package test.jdbc.datasource; + +import java.io.File; + +import javax.sql.DataSource; + +import org.apache.derby.jdbc.EmbeddedDataSource; +import org.springframework.beans.factory.config.AbstractFactoryBean; + +public class DerbyDataSourceFactoryBean extends AbstractFactoryBean { + + private String dataDirectory = "derby-home"; + + DataSource dataSource; + + public void setDataDirectory(String dataDirectory) { + this.dataDirectory = dataDirectory; + } + + protected Object createInstance() throws Exception { + File directory = new File(dataDirectory); + System.setProperty("derby.system.home", directory.getCanonicalPath()); + System.setProperty("derby.storage.fileSyncTransactionLog", "true"); + System.setProperty("derby.storage.pageCacheSize", "100"); + + final EmbeddedDataSource ds = new EmbeddedDataSource(); + ds.setDatabaseName("derbydb"); + ds.setCreateDatabase("create"); + dataSource = ds; + + return ds; + } + + public Class getObjectType() { + return DataSource.class; + } + +} diff --git a/spring-batch-samples/src/test/resources/org/springframework/batch/sample/item/processor/staging-test-context.xml b/spring-batch-samples/src/test/resources/org/springframework/batch/sample/item/processor/staging-test-context.xml new file mode 100644 index 000000000..32792fa62 --- /dev/null +++ b/spring-batch-samples/src/test/resources/org/springframework/batch/sample/item/processor/staging-test-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file