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 37f77cd6f..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 @@ -1,64 +1,96 @@ package org.springframework.batch.sample.common; -import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.Iterator; +import java.util.List; import javax.sql.DataSource; import org.apache.commons.lang.SerializationUtils; -import org.springframework.batch.item.ExecutionContext; +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.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.database.JdbcCursorItemReader; +import org.springframework.batch.item.ReaderNotOpenException; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.jdbc.core.PreparedStatementSetter; -import org.springframework.jdbc.core.RowMapper; +import org.springframework.dao.DataAccessException; +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 implements ItemReader, ItemStream, InitializingBean { +public class StagingItemReader implements ItemReader, StepExecutionListener, InitializingBean, DisposableBean { + + private static Log logger = LogFactory.getLog(StagingItemReader.class); + + private StepExecution stepExecution; + + private final Object lock = new Object(); + + private volatile boolean initialized = false; + + private volatile Iterator keys; - private JdbcCursorItemReader delegate; - private SimpleJdbcTemplate jdbcTemplate; - private long jobId; - - public void setJobId(long jobId) { - this.jobId = jobId; - } - public void setDataSource(DataSource dataSource) { jdbcTemplate = new SimpleJdbcTemplate(dataSource); - delegate = new JdbcCursorItemReader(); - delegate.setDataSource(dataSource); - delegate.setSql("SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID"); - delegate.setPreparedStatementSetter(new PreparedStatementSetter() { - public void setValues(PreparedStatement ps) throws SQLException { - ps.setLong(1, jobId); - ps.setString(2, StagingItemWriter.NEW); - } - }); - delegate.setMapper(new RowMapper() { - public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - return rs.getLong(1); - } - }); + } + + public void destroy() throws Exception { + initialized = false; + keys = null; } public final void afterPropertiesSet() throws Exception { - delegate.afterPropertiesSet(); + Assert.notNull(jdbcTemplate, "You must provide a DataSource."); } - public T read() throws Exception { + private List retrieveKeys() { - Long id = delegate.read(); + synchronized (lock) { + + return jdbcTemplate.query( + + "SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID", + + new ParameterizedRowMapper() { + public Long mapRow(ResultSet rs, int rowNum) throws SQLException { + return rs.getLong(1); + } + }, + + stepExecution.getJobExecution().getJobId(), StagingItemWriter.NEW); + + } + + } + + public T read() throws DataAccessException { + + if (!initialized) { + throw new ReaderNotOpenException("ItemStream must be open before it can be read."); + } + + Long id = null; + synchronized (lock) { + if (keys.hasNext()) { + id = keys.next(); + } + } + logger.debug("Retrieved key from list: " + id); + + if (id == null) { + return null; + } @SuppressWarnings("unchecked") T result = (T) jdbcTemplate.queryForObject("SELECT VALUE FROM BATCH_STAGING WHERE ID=?", new ParameterizedRowMapper() { @@ -68,20 +100,55 @@ public class StagingItemReader implements ItemReader, ItemStream, Initiali } }, id); + // 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."); + } + return result; } - public void close(ExecutionContext executionContext) throws ItemStreamException { - delegate.close(executionContext); + /* + * (non-Javadoc) + * + * @see + * org.springframework.batch.core.domain.StepListener#afterStep(StepExecution + * ) + */ + public ExitStatus afterStep(StepExecution stepExecution) { + return null; } - public void open(ExecutionContext executionContext) throws ItemStreamException { - delegate.open(executionContext); + /* + * (non-Javadoc) + * + * @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org. + * springframework.batch.core.domain.StepExecution) + */ + public void beforeStep(StepExecution stepExecution) { + this.stepExecution = stepExecution; + synchronized (lock) { + if (keys == null) { + keys = retrieveKeys().iterator(); + logger.info("Keys obtained for staging."); + initialized = true; + } + } } - public void update(ExecutionContext executionContext) throws ItemStreamException { - delegate.update(executionContext); + /* + * (non-Javadoc) + * + * @see + * org.springframework.batch.core.domain.StepListener#onErrorInStep(java + * .lang.Throwable) + */ + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { + return null; } } diff --git a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml index 6effa93ac..c12461453 100644 --- a/spring-batch-samples/src/main/resources/jobs/parallelJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/parallelJob.xml @@ -46,11 +46,6 @@ - - - - - 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 94e55f835..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.core.UnexpectedJobExecutionException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; import org.springframework.test.context.ContextConfiguration; @@ -55,11 +54,12 @@ public class StagingItemReaderTests { new JobParameters(), "testJob"))); writer.beforeStep(stepExecution); writer.write(Arrays.asList(new String[] { "FOO", "BAR", "SPAM", "BUCKET" })); - reader.setJobId(jobId); + reader.beforeStep(stepExecution); } @AfterTransaction public void onTearDownAfterTransaction() throws Exception { + reader.destroy(); simpleJdbcTemplate.update("DELETE FROM BATCH_STAGING"); } @@ -118,13 +118,7 @@ public class StagingItemReaderTests { String.class, id); assertEquals(StagingItemWriter.NEW, before); - Object item; - try { - item = reader.read(); - } - catch (Exception e) { - throw new UnexpectedJobExecutionException("Reader error", e); - } + Object item = reader.read(); assertEquals("FOO", item); transactionStatus.setRollbackOnly();