diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java index e5d35ddb9..09fe63217 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java @@ -26,6 +26,8 @@ import java.io.Serializable; * A BatchStatus can be safely serialized, however, it should be noted that the pattern can break down if different * class loaders load the enumeration. * + * This class is immutable and therefore thread-safe. + * * @author Lucas Ward * @author Greg Kick */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobParameters.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobParameters.java index edea65970..37270e4bf 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobParameters.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobParameters.java @@ -22,6 +22,8 @@ import org.apache.commons.lang.builder.HashCodeBuilder; * equals another. Furthermore, because these parameters will need to be * persisted, it is vital that the types added are restricted. * + * This class is immutable and therefore thread-safe. + * * @author Lucas Ward * @since 1.0 */ diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java index 309c54848..a536f62b5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractBufferedItemReaderItemStream.java @@ -12,6 +12,8 @@ import org.springframework.util.Assert; * {@link ExecutionContext} (therefore requires item ordering to be preserved * between runs). * + * Subclasses are inherently *not* thread-safe. + * * @author Robert Kasanicky */ public abstract class AbstractBufferedItemReaderItemStream implements ItemReader, ItemStream { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractItemWriter.java index b40e37915..e7bc126f1 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/AbstractItemWriter.java @@ -15,20 +15,17 @@ */ package org.springframework.batch.item; - /** - * Abstract {@link ItemWriter} that allows for base classes to only - * implement the close method if they need it. Because it is likely - * that the flush and clear methods may not need to be implemented, - * they are provided in this class. + * Abstract {@link ItemWriter} that allows for base classes to only implement + * the {@link #flush()} and {@link #clear()} methods if they need it. * * @author Lucas Ward */ -public abstract class AbstractItemWriter implements ItemWriter{ - +public abstract class AbstractItemWriter implements ItemWriter { + public void flush() throws FlushFailedException { } - + public void clear() throws ClearFailedException { } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java index 2c5672756..06f1b26b6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemReader.java @@ -21,9 +21,10 @@ package org.springframework.batch.item; * * Implementations are expected to be stateful and will be called multiple times * for each batch, with each call to {@link #read()} returning a different value - * and finally returning null when all input data is exhausted.
+ * and finally returning null when all input data is + * exhausted.
* - * Implementations need to be thread safe and clients of a {@link ItemReader} + * Implementations need *not* be thread safe and clients of a {@link ItemReader} * need to be aware that this is the case.
* * A richer interface (e.g. with a look ahead or peek) is not feasible because @@ -48,14 +49,12 @@ public interface ItemReader { Object read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException; /** - * Mark the stream so that it can be reset later and the items backed out.
+ * Mark the stream so that it can be reset later and the items backed + * out.
* * Mark is called before reading a new chunk of items - in case of rollback * mark will not be called again before re-processing the chunk.
* - * In a multi-threaded setting implementations have to ensure that only the - * state from the current thread is saved. - * * @throws MarkFailedException if there is a problem with the mark. If a * mark fails inside a transaction, it would be worrying, but not normally * fatal. @@ -67,9 +66,6 @@ public interface ItemReader { * such that changes (items read or written) since the last call to mark * will not be visible after a call to close.
* - * In a multi-threaded setting implementations have to ensure that only the - * state from the current thread is reset. - * * @throws ResetFailedException if there is a problem with the reset. If a * reset fails inside a transaction, it would normally be fatal, and would * leave the stream in an inconsistent state. So while this is an unchecked diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java index 0afb57ae0..87f7c54e0 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java @@ -60,6 +60,8 @@ import org.springframework.util.Assert; * database for all the keys would be too resource intensive. *

* + * The implementation is *not* thread-safe. + * * * @author Lucas Ward */ @@ -190,7 +192,9 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBean, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext) + * @see + * org.springframework.batch.io.support.AbstractTransactionalIoSource#reset + * (org.springframework.batch.item.ExecutionContext) */ public void reset() { keysIterator = keys.listIterator(lastCommitIndex); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java index 0f22284a5..f6f8ddc52 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java @@ -46,6 +46,8 @@ import org.springframework.util.ClassUtils; * Reset(rollback) functionality is implemented by item buffering allowing the * cursor used to be forward-only. * + * The implementation is *not* thread-safe. + * * @author Robert Kasanicky * @author Dave Syer */ diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisDrivingQueryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisDrivingQueryItemReader.java index 9e172ab74..a234577f9 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisDrivingQueryItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/IbatisDrivingQueryItemReader.java @@ -21,9 +21,12 @@ import org.springframework.orm.ibatis.SqlMapClientTemplate; import com.ibatis.sqlmap.client.SqlMapClient; /** - * Extension of {@link DrivingQueryItemReader} that maps keys to - * objects. An iBatis query id must be set to map and return each 'detail record'. - * + * Extension of {@link DrivingQueryItemReader} that maps keys to objects. An + * iBatis query id must be set to map and return each 'detail record'. + * + * The writer is thread safe after its properties are set (normal singleton + * behaviour). + * * @author Lucas Ward * @see IbatisKeyCollector */ @@ -34,34 +37,34 @@ public class IbatisDrivingQueryItemReader extends DrivingQueryItemReader { private SqlMapClientTemplate sqlMapClientTemplate; /** - * Overridden read() that uses the returned key as arguments to the details query. - * + * Overridden read() that uses the returned key as arguments to the details + * query. + * * @see org.springframework.batch.item.database.DrivingQueryItemReader#read() */ public Object read() { Object key = super.read(); - if (key==null) { + if (key == null) { return null; } return sqlMapClientTemplate.queryForObject(detailsQueryId, key); } /** - * @param detailsQueryId id of the iBATIS select statement that will used - * to retrieve an object for a single primary key from the list - * returned by driving query + * @param detailsQueryId id of the iBATIS select statement that will used to + * retrieve an object for a single primary key from the list returned by + * driving query */ public void setDetailsQueryId(String detailsQueryId) { this.detailsQueryId = detailsQueryId; } - + /** * Set the {@link SqlMapClientTemplate} to use for this input source. * * @param sqlMapClient */ - public void setSqlMapClient( - SqlMapClient sqlMapClient) { + public void setSqlMapClient(SqlMapClient sqlMapClient) { this.sqlMapClientTemplate = new SqlMapClientTemplate(sqlMapClient); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/IbatisKeyCollector.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/IbatisKeyCollector.java index 136be21b5..e2b4e7b80 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/IbatisKeyCollector.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/IbatisKeyCollector.java @@ -15,9 +15,12 @@ import com.ibatis.sqlmap.client.SqlMapClient; /** * {@link KeyCollector} based on iBATIS ORM framework. It is functionally * similar to {@link SingleColumnJdbcKeyCollector} but does not make assumptions - * about the primary key structure. A separate restart query is necessary to - * ensure that only the required keys remaining for processing are returned, rather - * than the entire original list.

+ * about the primary key structure. A separate restart query is necessary to + * ensure that only the required keys remaining for processing are returned, + * rather than the entire original list.

+ * + * The writer is thread safe after its properties are set (normal singleton + * behaviour). * * @author Robert Kasanicky * @author Lucas Ward @@ -54,6 +57,7 @@ public class IbatisKeyCollector extends ExecutionContextUserSupport implements K /* * (non-Javadoc) + * * @see KeyCollector#saveState(Object, ExecutionContext) */ public void updateContext(Object key, ExecutionContext executionContext) { @@ -64,7 +68,9 @@ public class IbatisKeyCollector extends ExecutionContextUserSupport implements K /* * (non-Javadoc) - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + * + * @see + * org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { Assert.notNull(sqlMapClientTemplate, "SqlMaperClientTemplate must not be null."); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/MultipleColumnJdbcKeyCollector.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/MultipleColumnJdbcKeyCollector.java index fd5fecf4f..a55e5984b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/MultipleColumnJdbcKeyCollector.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/MultipleColumnJdbcKeyCollector.java @@ -40,18 +40,23 @@ import org.springframework.util.StringUtils; * row in the result set to an Object must be set in order to work correctly. *

* + * The implementation is thread-safe as long as the + * {@link #setPreparedStatementSetter(ItemPreparedStatementSetter)} and + * {@link #setKeyMapper(RowMapper)} are thread-safe (true for default values). + * * @author Lucas Ward + * * @see DrivingQueryItemReader * @see ItemPreparedStatementSetter */ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport implements KeyCollector { private static final String CURRENT_KEY = "current.key"; - + private JdbcTemplate jdbcTemplate; private RowMapper keyMapper = new ColumnMapRowMapper(); - + private ItemPreparedStatementSetter preparedStatementSetter = new ColumnMapItemPreparedStatementSetter(); private String sql; @@ -66,8 +71,7 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport * Construct a new ItemReader. * * @param jdbcTemplate - * @param sql - SQL statement that returns all keys to process. - * object. + * @param sql - SQL statement that returns all keys to process. object. */ public MultipleColumnJdbcKeyCollector(JdbcTemplate jdbcTemplate, String sql) { this(); @@ -79,17 +83,21 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport /* * (non-Javadoc) - * @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryItemReader#retrieveKeys() + * + * @see + * org.springframework.batch.io.sql.scratch.AbstractDrivingQueryItemReader + * #retrieveKeys() */ public List retrieveKeys(ExecutionContext executionContext) { Assert.state(keyMapper != null, "KeyMapper must not be null."); Assert.state(StringUtils.hasText(restartSql), "The RestartQuery must not be null or empty" + " in order to restart."); - + if (executionContext.size() > 0) { Object key = executionContext.get(getKey(CURRENT_KEY)); - return jdbcTemplate.query(restartSql, new PreparedStatementSetterKeyWrapper(key, preparedStatementSetter), keyMapper); + return jdbcTemplate.query(restartSql, new PreparedStatementSetterKeyWrapper(key, preparedStatementSetter), + keyMapper); } else { return jdbcTemplate.query(sql, keyMapper); @@ -98,7 +106,10 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport /* * (non-Javadoc) - * @see org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext(java.lang.Object) + * + * @see + * org.springframework.batch.io.driving.KeyGenerator#getKeyAsExecutionContext + * (java.lang.Object) */ public void updateContext(Object key, ExecutionContext executionContext) { Assert.notNull(key, "The key must not be null"); @@ -118,7 +129,9 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport /* * (non-Javadoc) - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() + * + * @see + * org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { Assert.notNull(jdbcTemplate, "The JdbcTemplate must not be null."); @@ -127,8 +140,7 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport } /** - * Set the {@link RowMapper} to be used to map a result set - * to keys. + * Set the {@link RowMapper} to be used to map a result set to keys. * * @param keyMapper */ @@ -148,17 +160,17 @@ public class MultipleColumnJdbcKeyCollector extends ExecutionContextUserSupport public void setJdbcTemplate(JdbcTemplate jdbcTemplate) { this.jdbcTemplate = jdbcTemplate; } - - public void setPreparedStatementSetter( - ItemPreparedStatementSetter preparedStatementSetter) { + + public void setPreparedStatementSetter(ItemPreparedStatementSetter preparedStatementSetter) { this.preparedStatementSetter = preparedStatementSetter; } - - private static class PreparedStatementSetterKeyWrapper implements PreparedStatementSetter{ - + + private static class PreparedStatementSetterKeyWrapper implements PreparedStatementSetter { + private Object key; + private ItemPreparedStatementSetter pss; - + public PreparedStatementSetterKeyWrapper(Object key, ItemPreparedStatementSetter pss) { this.key = key; this.pss = pss; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java index 6dd12c840..fc9be6228 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java @@ -55,12 +55,15 @@ import org.springframework.util.ClassUtils; * header, skip given number of lines at the beginning of the file. *

* + * The implementation is *not* thread-safe. + * * @author Waseem Malik * @author Tomas Slanina * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemReader extends AbstractBufferedItemReaderItemStream implements ResourceAwareItemReaderItemStream, InitializingBean { +public class FlatFileItemReader extends AbstractBufferedItemReaderItemStream implements + ResourceAwareItemReaderItemStream, InitializingBean { private static Log log = LogFactory.getLog(FlatFileItemReader.class); @@ -257,7 +260,7 @@ public class FlatFileItemReader extends AbstractBufferedItemReaderItemStream imp ((AbstractLineTokenizer) tokenizer).setNames(names); } } - + } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 22fe5bf57..3bb686aca 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -59,6 +59,8 @@ import org.springframework.util.ClassUtils; * actual writing to file occurs. *

* + * The implementation is *not* thread-safe. + * * @author Waseem Malik * @author Tomas Slanina * @author Robert Kasanicky diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java index edc7f50e8..e8b2f6f8e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java @@ -39,6 +39,9 @@ import org.springframework.util.Assert; * {@link #read()}. If a recovery step is needed, set the error destination and * the item will be sent there if processing fails in a stateful retry. * + * The implementation is thread safe after its properties are set (normal + * singleton behaviour). + * * @author Dave Syer * */ diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java index cedbffae6..d0a18f73e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemWriter.java @@ -10,6 +10,8 @@ import org.springframework.batch.item.ItemWriter; /** * Calls a collection of ItemWriters in fixed-order sequence. * + * The implementation is thread-safe if all delegates are thread-safe. + * * @author Robert Kasanicky */ public class CompositeItemWriter implements ItemWriter { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java index 58dc0c570..6b5d92c51 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java @@ -22,10 +22,12 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; /** - * Simple wrapper around {@link ItemReader}. The input source is expected to + * Simple wrapper around {@link ItemReader}. The item reader is expected to * take care of open and close operations. If necessary it should be registered * as a step scoped bean to ensure that the lifecycle methods are called. * + * The implementation is thread-safe if the delegate is thread-safe. + * * @author Dave Syer */ public class DelegatingItemReader extends AbstractItemReader implements InitializingBean { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java index 2323c22ef..74beb343f 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java @@ -9,6 +9,8 @@ import org.springframework.util.Assert; /** * Simple wrapper around {@link ItemWriter}. * + * The implementation is thread-safe if the delegate is thread-safe. + * * @author Dave Syer * @author Robert Kasanicky */ diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java index 05e3d8aef..603caf077 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java @@ -26,6 +26,8 @@ import org.springframework.util.ClassUtils; * events so that the fragments can be further processed like standalone XML * documents. * + * The implementation is *not* thread-safe. + * * @author Robert Kasanicky */ public class StaxEventItemReader extends AbstractBufferedItemReaderItemStream implements diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index ebc42047f..2b17f5696 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -41,6 +41,8 @@ import org.springframework.util.CollectionUtils; * Output is buffered until {@link #flush()} is called - only then the actual * writing to file takes place. * + * The implementation is *not* thread-safe. + * * @author Peter Zozom * @author Robert Kasanicky * diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java index 626b8a476..2b19e8164 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/ExitStatus.java @@ -23,6 +23,8 @@ import org.springframework.util.StringUtils; * Value object used to carry information about the status of a * {@link RepeatOperations}. * + * ExitStatus is immutable and therefore thread-safe. + * * @author Dave Syer * */ @@ -170,8 +172,8 @@ public class ExitStatus implements Serializable { } /** - * Add an exit code to an existing {@link ExitStatus}. If there is already - * a code present tit will be replaced. + * Add an exit code to an existing {@link ExitStatus}. If there is already a + * code present tit will be replaced. * * @param code the code to add * @return a new {@link ExitStatus} with the same properties but a new exit diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java index 7d764fa70..261a4b959 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java @@ -25,6 +25,10 @@ import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.transaction.support.TransactionSynchronizationManager; 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 @@ -45,7 +49,8 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /** * Public setter for the {@link LobHandler}. * - * @param lobHandler the {@link LobHandler} to set (defaults to {@link DefaultLobHandler}). + * @param lobHandler the {@link LobHandler} to set (defaults to + * {@link DefaultLobHandler}). */ public void setLobHandler(LobHandler lobHandler) { this.lobHandler = lobHandler; @@ -107,18 +112,18 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite 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); - } - }); + 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 }); + 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."); + + " was updated concurrently when trying to mark as complete (updated " + count + " records."); } return result; } @@ -133,14 +138,15 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite synchronized (lock) { if (keys.hasNext()) { Assert.state(TransactionSynchronizationManager.isActualTransactionActive(), - "Transaction not active for this thread."); + "Transaction not active for this thread."); Long next = (Long) keys.next(); getBuffer().add(next); key = next; logger.debug("Retrieved key from list: " + key); } } - } else { + } + else { logger.debug("Retrieved key from buffer: " + key); } return key; @@ -192,9 +198,12 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite } /** - * 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. + * 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. * * @see org.springframework.batch.item.ItemReader#mark() */ @@ -205,7 +214,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext) + * @see + * org.springframework.batch.item.ItemStream#reset(org.springframework.batch + * .item.ExecutionContext) */ public void reset() { getBuffer().rollback(); @@ -214,7 +225,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.item.ExecutionContextProvider#getExecutionContext() + * @see + * org.springframework.batch.item.ExecutionContextProvider#getExecutionContext + * () */ public void update(ExecutionContext executionContext) { } @@ -222,7 +235,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.core.domain.StepListener#afterStep(StepExecution) + * @see + * org.springframework.batch.core.domain.StepListener#afterStep(StepExecution + * ) */ public ExitStatus afterStep(StepExecution stepExecution) { return null; @@ -231,7 +246,8 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution) + * @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org. + * springframework.batch.core.domain.StepExecution) */ public void beforeStep(StepExecution stepExecution) { this.stepExecution = stepExecution; @@ -240,7 +256,9 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemStream, Ite /* * (non-Javadoc) * - * @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable) + * @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/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java index 729d52a2d..f3f623154 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/StagingItemWriter.java @@ -19,6 +19,9 @@ import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; +/** + * Database {@link ItemWriter} implementing the process indicator pattern. + */ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionListener, ItemWriter { public static final String NEW = "N"; @@ -71,16 +74,15 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionLi final long id = incrementer.nextLongValue(); final long jobId = stepExecution.getJobExecution().getJobId().longValue(); final byte[] blob = SerializationUtils.serialize((Serializable) data); - getJdbcTemplate() - .update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)", - new PreparedStatementSetter() { - public void setValues(PreparedStatement ps) throws SQLException { - ps.setLong(1, id); - ps.setLong(2, jobId); - lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob); - ps.setString(4, NEW); - } - + getJdbcTemplate().update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)", + new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + ps.setLong(1, id); + ps.setLong(2, jobId); + lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob); + ps.setString(4, NEW); + } + }); } @@ -90,22 +92,33 @@ public class StagingItemWriter extends JdbcDaoSupport implements StepExecutionLi public void flush() throws FlushFailedException { } - /* (non-Javadoc) - * @see org.springframework.batch.core.domain.StepListener#afterStep(StepExecution) + /* + * (non-Javadoc) + * + * @see + * org.springframework.batch.core.domain.StepListener#afterStep(StepExecution + * ) */ public ExitStatus afterStep(StepExecution stepExecution) { return null; } - /* (non-Javadoc) - * @see org.springframework.batch.core.domain.StepListener#beforeStep(org.springframework.batch.core.domain.StepExecution) + /* + * (non-Javadoc) + * + * @seeorg.springframework.batch.core.domain.StepListener#beforeStep(org. + * springframework.batch.core.domain.StepExecution) */ public void beforeStep(StepExecution stepExecution) { this.stepExecution = stepExecution; } - /* (non-Javadoc) - * @see org.springframework.batch.core.domain.StepListener#onErrorInStep(java.lang.Throwable) + /* + * (non-Javadoc) + * + * @see + * org.springframework.batch.core.domain.StepListener#onErrorInStep(java + * .lang.Throwable) */ public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return null;