diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java index 20d7daf80..eb55f81ef 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java @@ -2,234 +2,114 @@ package org.springframework.batch.sample.common; import java.sql.ResultSet; import java.sql.SQLException; -import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import javax.sql.DataSource; + import org.apache.commons.lang.SerializationUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; -import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ReaderNotOpenException; -import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; -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.TransactionSynchronizationManager; +import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.jdbc.core.simple.ParameterizedRowMapper; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; import org.springframework.util.Assert; /** * Thread-safe database {@link ItemReader} implementing the process indicator * pattern. */ -public class StagingItemReader extends JdbcDaoSupport implements ItemStream, ItemReader, StepExecutionListener { - - // Key for buffer in transaction synchronization manager - private static final String BUFFER_KEY = StagingItemReader.class.getName() + ".BUFFER"; +public class StagingItemReader implements ItemReader, StepExecutionListener, InitializingBean, DisposableBean { private static Log logger = LogFactory.getLog(StagingItemReader.class); private StepExecution stepExecution; - private LobHandler lobHandler = new DefaultLobHandler(); - private final Object lock = new Object(); private volatile boolean initialized = false; private volatile Iterator keys; - /** - * Public setter for the {@link LobHandler}. - * - * @param lobHandler the {@link LobHandler} to set (defaults to - * {@link DefaultLobHandler}). - */ - public void setLobHandler(LobHandler lobHandler) { - this.lobHandler = lobHandler; + private SimpleJdbcTemplate jdbcTemplate; + + public void setDataSource(DataSource dataSource) { + jdbcTemplate = new SimpleJdbcTemplate(dataSource); } - /** - * - * @see org.springframework.batch.item.ItemStream#close(ExecutionContext) - */ - public void close(ExecutionContext executionContext) { + public void destroy() throws Exception { initialized = false; keys = null; - if (TransactionSynchronizationManager.hasResource(BUFFER_KEY)) { - TransactionSynchronizationManager.unbindResource(BUFFER_KEY); - } } - /** - * - * @see org.springframework.batch.item.ItemStream#open(ExecutionContext) - */ - public void open(ExecutionContext executionContext) { - // Can be called from multiple threads because of lazy initialisation... - synchronized (lock) { - if (keys == null) { - keys = retrieveKeys().iterator(); - logger.info("Keys obtained for staging."); - initialized = true; - } - } + public final void afterPropertiesSet() throws Exception { + Assert.notNull(jdbcTemplate, "You must provide a DataSource."); } - @SuppressWarnings("unchecked") private List retrieveKeys() { synchronized (lock) { - return getJdbcTemplate().query( + return jdbcTemplate.query( "SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID", - new Object[] { stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW }, - - new RowMapper() { - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + new ParameterizedRowMapper() { + public Long mapRow(ResultSet rs, int rowNum) throws SQLException { return rs.getLong(1); } - } + }, - ); + stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW); } } - @SuppressWarnings("unchecked") public T read() throws DataAccessException { - Long id = doRead(); - if (id == null) { - return null; - } - T result = (T) 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[] { StagingItemWriter.DONE, id, StagingItemWriter.NEW }); - if (count != 1) { - throw new OptimisticLockingFailureException("The staging record with ID=" + id - + " was updated concurrently when trying to mark as complete (updated " + count + " records."); - } - return result; - } - - private Long doRead() { if (!initialized) { throw new ReaderNotOpenException("ItemStream must be open before it can be read."); } - Long key = getBuffer().next(); - if (key == null) { - synchronized (lock) { - if (keys.hasNext()) { - Assert.state(TransactionSynchronizationManager.isActualTransactionActive(), - "Transaction not active for this thread."); - Long next = keys.next(); - getBuffer().add(next); - key = next; - logger.debug("Retrieved key from list: " + key); - } + Long id = null; + synchronized (lock) { + if (keys.hasNext()) { + id = keys.next(); } } - else { - logger.debug("Retrieved key from buffer: " + key); - } - return key; + logger.debug("Retrieved key from list: " + id); - } - - 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; - } - - private static class StagingBuffer { - - private List list = new ArrayList(); - - private Iterator iter = new ArrayList().iterator(); - - public Long next() { - if (iter.hasNext()) { - return iter.next(); - } + if (id == null) { return null; } + @SuppressWarnings("unchecked") + T result = (T) jdbcTemplate.queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?", + new ParameterizedRowMapper() { + public Object mapRow(ResultSet rs, int rowNum) throws SQLException { + byte[] blob = rs.getBytes(1); + return SerializationUtils.deserialize(blob); + } + }, id); - public void add(Long next) { - list.add(next); + // Update now - changes will rollback if there is a problem later. + int count = jdbcTemplate.update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?", + StagingItemWriter.DONE, id, StagingItemWriter.NEW); + if (count != 1) { + throw new OptimisticLockingFailureException("The staging record with ID=" + id + + " was updated concurrently when trying to mark as complete (updated " + count + " records."); } - public void rollback() { - logger.debug("Resetting buffer on rollback: " + list); - iter = new ArrayList(list).iterator(); - } + return result; - public void commit() { - logger.debug("Clearing buffer on commit: " + list); - list.clear(); - iter = new ArrayList().iterator(); - } - - public String toString() { - return "list=" + list + "; iter.hasNext()=" + iter.hasNext(); - } - } - - /** - * Mark is supported in a multi- as well as a single-threaded environment. - * The state backing the mark is a buffer, and access is synchronized, so - * multiple threads can be accommodated. Buffers are stored as transaction - * resources (using - * {@link TransactionSynchronizationManager#bindResource(Object, Object)}), - * so they are thread bound. - */ - public void mark() { - getBuffer().commit(); - } - - /* - * (non-Javadoc) - * - * @see - * org.springframework.batch.item.ItemStream#reset(org.springframework.batch - * .item.ExecutionContext) - */ - public void reset() { - getBuffer().rollback(); - } - - /* - * (non-Javadoc) - * - * @see - * org.springframework.batch.item.ExecutionContextProvider#getExecutionContext - * () - */ - public void update(ExecutionContext executionContext) { } /* @@ -251,6 +131,13 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, */ public void beforeStep(StepExecution stepExecution) { this.stepExecution = stepExecution; + synchronized (lock) { + if (keys == null) { + keys = retrieveKeys().iterator(); + logger.info("Keys obtained for staging."); + initialized = true; + } + } } /* diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemWriter.java index 3c4d06ca7..a8a84ca11 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemWriter.java @@ -14,8 +14,6 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.jdbc.core.BatchPreparedStatementSetter; import org.springframework.jdbc.core.support.JdbcDaoSupport; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; -import org.springframework.jdbc.support.lob.DefaultLobHandler; -import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -34,17 +32,6 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutio private StepExecution stepExecution; - private LobHandler lobHandler = new DefaultLobHandler(); - - /** - * Public setter for the {@link LobHandler}. - * @param lobHandler the {@link LobHandler} to set (defaults to - * {@link DefaultLobHandler}). - */ - public void setLobHandler(LobHandler lobHandler) { - this.lobHandler = lobHandler; - } - /** * Check mandatory properties. * @@ -92,7 +79,7 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutio ps.setLong(1, id); ps.setLong(2, jobId); - lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob); + ps.setBytes(3, blob); ps.setString(4, NEW); } }); diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index b5dcb7422..ccfdc42fa 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -26,8 +26,6 @@ class="org.springframework.batch.sample.common.StagingItemWriter"> - - diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java index 8fc727a64..b1dabd5ff 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/StagingItemReaderTests.java @@ -13,7 +13,6 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; -import org.springframework.batch.item.ExecutionContext; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; import org.springframework.test.context.ContextConfiguration; @@ -35,7 +34,7 @@ public class StagingItemReaderTests { @Autowired private PlatformTransactionManager transactionManager; - + @Autowired private StagingItemWriter writer; @@ -44,7 +43,6 @@ public class StagingItemReaderTests { private Long jobId = 11L; - @Autowired public void setDataSource(DataSource dataSource) { this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource); @@ -54,23 +52,22 @@ public class StagingItemReaderTests { public void onSetUpBeforeTransaction() throws Exception { StepExecution stepExecution = new StepExecution("stepName", new JobExecution(new JobInstance(jobId, new JobParameters(), "testJob"))); - reader.beforeStep(stepExecution); writer.beforeStep(stepExecution); - writer.write(Arrays.asList(new String[] {"FOO","BAR","SPAM","BUCKET"})); - reader.open(new ExecutionContext()); + writer.write(Arrays.asList(new String[] { "FOO", "BAR", "SPAM", "BUCKET" })); + reader.beforeStep(stepExecution); } @AfterTransaction public void onTearDownAfterTransaction() throws Exception { - reader.close(null); + reader.destroy(); simpleJdbcTemplate.update("DELETE FROM BATCH_STAGING"); } - @Transactional @Test + @Transactional + @Test public void testReaderUpdatesProcessIndicator() throws Exception { - long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", - jobId); + long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", jobId); String before = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", String.class, id); assertEquals(StagingItemWriter.NEW, before); @@ -84,131 +81,55 @@ public class StagingItemReaderTests { } - @Transactional @Test + @Transactional + @Test public void testUpdateProcessIndicatorAfterCommit() throws Exception { TransactionTemplate txTemplate = new TransactionTemplate(transactionManager); txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - try { - testReaderUpdatesProcessIndicator(); - } catch (Exception e) { - fail("Unxpected Exception: " + e); - } - long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", - jobId); - String before = - simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", - String.class, id); - assertEquals(StagingItemWriter.DONE, before); - return null; - } - }); + public Object doInTransaction(TransactionStatus transactionStatus) { + try { + testReaderUpdatesProcessIndicator(); + } + catch (Exception e) { + fail("Unxpected Exception: " + e); + } + return null; + } + }); + long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", jobId); + String before = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", + String.class, id); + assertEquals(StagingItemWriter.DONE, before); } - @Transactional @Test - public void testProviderRollsBackMultipleTimes() throws Exception { + @Transactional + @Test + public void testReaderRollsBackProcessIndicator() throws Exception { TransactionTemplate txTemplate = new TransactionTemplate(transactionManager); txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); - reader.mark(); + final Long idToUse = (Long) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus transactionStatus) { - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - int count = simpleJdbcTemplate.queryForInt("SELECT COUNT(*) from BATCH_STAGING where JOB_ID=? AND PROCESSED=?", - jobId, StagingItemWriter.NEW); - assertEquals(4, count); + long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", jobId); + String before = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", + String.class, id); + assertEquals(StagingItemWriter.NEW, before); - Object item = reader.read(); - assertEquals("FOO", item); - item = reader.read(); - assertEquals("BAR", item); + Object item = reader.read(); + assertEquals("FOO", item); - transactionStatus.setRollbackOnly(); + transactionStatus.setRollbackOnly(); - return null; - } - }); + return id; + } + }); - reader.reset(); - - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - Object item = reader.read(); - assertEquals("FOO", item); - item = reader.read(); - assertEquals("BAR", item); - item = reader.read(); - assertEquals("SPAM", item); - - transactionStatus.setRollbackOnly(); - - return null; - } - }); - - reader.reset(); - - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - Object item = reader.read(); - assertEquals("FOO", item); - - transactionStatus.setRollbackOnly(); - - return null; - } - }); - - } - - @Transactional @Test - public void testProviderRollsBackProcessIndicator() throws Exception { - - TransactionTemplate txTemplate = new TransactionTemplate(transactionManager); - txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); - - reader.mark(); - // After a rollback we have to resynchronize the TX to simulate a real - // batch - final Long idToUse = (Long)txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - - long id = simpleJdbcTemplate.queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?", - jobId); - String before = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", - String.class, id); - assertEquals(StagingItemWriter.NEW, before); - - Object item = reader.read(); - assertEquals("FOO", item); - - transactionStatus.setRollbackOnly(); - - return id; - } - }); - - reader.reset(); - - // After a rollback we have to resynchronize the TX to simulate a real - // batch - txTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus transactionStatus) { - - String after = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", - String.class, idToUse); - assertEquals(StagingItemWriter.NEW, after); - - Object item = reader.read(); - assertEquals("FOO", item); - - transactionStatus.setRollbackOnly(); - - return null; - } - }); + String after = simpleJdbcTemplate.queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?", + String.class, idToUse); + assertEquals(StagingItemWriter.NEW, after); } }