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 @@
@@ -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