BATCH 144 and BATCH 150: create Hibernate and iBatis input sources with sample jobs.

infrastructure/pom.xml

@@ -1,6 +1,4 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-batch-infrastructure</artifactId>
@@ -102,20 +100,44 @@
 			<optional>true</optional>
 		</dependency>
 		<dependency>
-			<groupId>org.springframework.ws</groupId>
+	        <groupId>org.springframework.ws</groupId>
 			<artifactId>spring-oxm</artifactId>
-			<version>1.0.0</version>
-			<optional>true</optional>
-			<exclusions>
-				<exclusion>
+	        <version>1.0.0</version>
+	        <optional>true</optional>
+	        <exclusions>
+	        	<exclusion>
 					<groupId>org.springframework</groupId>
 					<artifactId>spring-beans</artifactId>
-				</exclusion>
-				<exclusion>
+	        	</exclusion>
+	        	<exclusion>
 					<groupId>org.springframework</groupId>
 					<artifactId>spring-core</artifactId>
-				</exclusion>
-			</exclusions>
+	        	</exclusion>
+	        </exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.hibernate</groupId>
+			<artifactId>hibernate</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<!-- Needed by Hibernate if JTA is excluded -->
+			<groupId>org.apache.geronimo.specs</groupId>
+			<artifactId>geronimo-jta_1.1_spec</artifactId>
+			<version>1.1</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<!-- Force Hibernate to use a particular nodep version of cglib in case of clash with Spring AOP -->
+			<groupId>cglib</groupId>
+			<artifactId>cglib-nodep</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.ibatis</groupId>
+			<artifactId>ibatis-sqlmap</artifactId>
+			<version>2.3.0</version>
+			<optional>true</optional>
 		</dependency>
 	</dependencies>
 
@@ -125,9 +147,7 @@
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-clover-plugin</artifactId>
 				<configuration>
-					<licenseLocation>
-						${basedir}/src/test/resources/clover.license
-					</licenseLocation>
+					<licenseLocation>${basedir}/src/test/resources/clover.license</licenseLocation>
 				</configuration>
 				<executions>
 					<execution>

infrastructure/src/main/java/org/springframework/batch/io/orm/hibernate/HibernateInputSource.java

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.orm.hibernate;
+
+import java.util.Properties;
+
+import org.hibernate.ScrollableResults;
+import org.hibernate.SessionFactory;
+import org.hibernate.StatelessSession;
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.item.ResourceLifecycle;
+import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
+import org.springframework.batch.restart.GenericRestartData;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.util.Assert;
+
+/**
+ * {@link InputSource} for reading database records built on top of Hibernate.
+ * 
+ * @author Robert Kasanicky
+ */
+public class HibernateInputSource implements InputSource, Restartable, InitializingBean, DisposableBean,
+	ResourceLifecycle {
+	
+	private static final String RESTART_DATA_ROW_NUMBER_KEY = "HibernateInputSource.rowNumber";
+
+	private SessionFactory sessionFactory;
+	private StatelessSession session;
+	private ScrollableResults cursor;
+	private String queryString;
+	
+	private int lastCommitRowNumber = 0;
+	
+	private boolean initialized = false;
+	private TransactionSynchronization synchronization = new HibernateInputSourceTransactionSynchronization();
+	
+	
+	public Object read() {
+		if (!initialized) {
+			open();
+		}
+		if (cursor.next()) {
+			return cursor.get(0);
+		}
+		return null;
+	}
+
+	/**
+	 * Close the resultset cursor and hibernate session.
+	 */
+	public void close() {
+		initialized = false;
+		cursor.close();
+		session.close();
+	}
+
+	/**
+	 * Create cursor for the query
+	 */
+	public void open() {
+		session = sessionFactory.openStatelessSession();
+		cursor = session.createQuery(queryString).scroll();
+		
+		BatchTransactionSynchronizationManager.registerSynchronization(synchronization );
+		initialized = true;
+	}
+
+	public void setSessionFactory(SessionFactory sessionFactory) {
+		this.sessionFactory = sessionFactory;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(sessionFactory);
+		Assert.hasLength(queryString);
+	}
+
+	public void destroy() throws Exception {
+		close();
+	}
+
+	public void setQueryString(String queryString) {
+		this.queryString = queryString;
+	}
+
+	/**
+	 * @return the current row number wrapped as <code>RestartData</code>
+	 */
+	public RestartData getRestartData() {
+		Properties props = new Properties();
+		props.setProperty(RESTART_DATA_ROW_NUMBER_KEY, String.valueOf(cursor.getRowNumber()));
+		
+		return new GenericRestartData(props);
+	}
+
+	/**
+	 * Set the cursor to the received row number.
+	 */
+	public void restoreFrom(RestartData data) {
+		Assert.state(!initialized,
+				"Cannot restore when already intialized.  Call close() first before restore()");
+		
+		Properties props = data.getProperties();
+		if (props.getProperty(RESTART_DATA_ROW_NUMBER_KEY) == null) {
+			return;
+		}
+		int rowNumber = Integer.parseInt(props.getProperty(RESTART_DATA_ROW_NUMBER_KEY));
+		open();
+		cursor.setRowNumber(rowNumber);
+	}
+	
+	/**
+	 * Encapsulates transaction events handling.
+	 */
+	private class HibernateInputSourceTransactionSynchronization extends TransactionSynchronizationAdapter {
+		
+		public void afterCompletion(int status) {
+			if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
+				cursor.setRowNumber(lastCommitRowNumber);
+			} else if (status == TransactionSynchronization.STATUS_COMMITTED) {
+				lastCommitRowNumber = cursor.getRowNumber();
+			}
+		}
+	}
+
+}

infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisDrivingQueryInputSource.java

@@ -0,0 +1,98 @@
+package org.springframework.batch.io.orm.ibatis;
+
+import java.util.List;
+import java.util.Properties;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.sql.SingleKeySqlDrivingQueryInputSource;
+import org.springframework.batch.io.support.AbstractDrivingQueryInputSource;
+import org.springframework.batch.restart.GenericRestartData;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.orm.ibatis.SqlMapClientTemplate;
+import org.springframework.util.Assert;
+
+import com.ibatis.sqlmap.client.SqlMapClient;
+
+/**
+ * Driving query {@link InputSource} based on iBATIS ORM framework. It is functionally similar to
+ * {@link SingleKeySqlDrivingQueryInputSource} but does not make assumptions about the primary key
+ * structure.
+ *
+ * @see SingleKeySqlDrivingQueryInputSource
+ *
+ * @author Robert Kasanicky
+ * @author Lucas Ward
+ */
+public class IbatisDrivingQueryInputSource extends AbstractDrivingQueryInputSource implements Restartable {
+
+	public static final String RESTART_KEY = "IbatisDrivingQueryInputSource.keyIndex";
+
+	private SqlMapClientTemplate sqlMapClientTemplate;
+
+	private String drivingQuery;
+
+	private String restartQueryId;
+
+	protected List retrieveKeys() {
+		return sqlMapClientTemplate.queryForList(drivingQuery);
+	}
+
+
+	public RestartData getRestartData() {
+		Properties props = new Properties();
+		props.setProperty(RESTART_KEY, getCurrentKey().toString());
+
+		return new GenericRestartData(props);
+	}
+
+	/**
+	 * Restore the keys list given the provided restart data.
+	 * 
+	 * @see org.springframework.batch.io.support.AbstractDrivingQueryInputSource#restoreKeys(org.springframework.batch.restart.RestartData)
+	 */
+	public List restoreKeys(RestartData data) {
+
+		Properties props = data.getProperties();
+		Object key = props.getProperty(RESTART_KEY);
+		return sqlMapClientTemplate.queryForList(restartQueryId, key);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(sqlMapClientTemplate, "SqlMaperClientTemplate must not be null.");
+		Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty.");
+	}
+
+	/**
+	 * @param sqlMapClient configured iBATIS client
+	 */
+	public void setSqlMapClient(SqlMapClient sqlMapClient) {
+		this.sqlMapClientTemplate = new SqlMapClientTemplate();
+		this.sqlMapClientTemplate.setSqlMapClient(sqlMapClient);
+	}
+
+	/**
+	 * @param drivingQueryId id of the iBATIS select statement that will be used
+	 * to retrieve the list of primary keys
+	 */
+	public void setDrivingQueryId(String drivingQueryId) {
+		this.drivingQuery = drivingQueryId;
+	}
+
+	/**
+	 * Set the id of the restart query.
+	 *
+	 * @param restartQueryId id of the iBatis select statement that will be used
+	 * to retrieve the list of primary keys after a restart.
+	 */
+	public void setRestartQueryId(String restartQueryId) {
+		this.restartQueryId = restartQueryId;
+	}
+
+	public final SqlMapClientTemplate getSqlMapClientTemplate() {
+		  return sqlMapClientTemplate;
+		}
+}

infrastructure/src/main/java/org/springframework/batch/io/orm/ibatis/IbatisInputSource.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.orm.ibatis;
+
+/**
+ * @author Lucas Ward
+ *
+ */
+public class IbatisInputSource extends IbatisDrivingQueryInputSource {
+
+	String detailsQueryId;
+
+	public Object read() {
+		return getSqlMapClientTemplate().queryForObject(detailsQueryId, super.read());
+	}
+
+
+	/**
+	 * @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;
+	}
+}

infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeyRestartDataConverter.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.sql;
+
+import org.springframework.batch.restart.RestartData;
+
+/**
+ * Converts an object representing a composite key to RestartData and
+ * back again.
+ *
+ * @author Lucas Ward
+ *
+ */
+public interface CompositeKeyRestartDataConverter {
+
+	/**
+	 * Given the provided composite key, return a RestartData representation.
+	 *
+	 * @param compositeKey
+	 * @return ResartData representing the composite key.
+	 */
+	public RestartData createRestartData(Object compositeKey);
+
+	/**
+	 * Given the provided restart data, return an array of objects that can
+	 * be used as parameters to a driving query.
+	 *
+	 * @param restartData
+	 * @return an array of objects that can be used as arguments to a JdbcTemplate.
+	 */
+	public Object[] createArguments(RestartData restartData);
+}

infrastructure/src/main/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSource.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.sql;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.batch.io.support.AbstractDrivingQueryInputSource;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>Sql implementation of the DrivingQueryInputSource that works for composite keys.
+ * (i.e. keys represented by multiple columns)  A sql query to be used to return the keys and
+ * a RowMapper to map each row in the resultset to an Object must be set in order for the
+ * InputSource to work correctly.
+ * </p>
+ *
+ * @author Lucas Ward
+ * @see AbstractDrivingQueryInputSource
+ */
+public class CompositeKeySqlDrivingQueryInputSource extends
+		AbstractDrivingQueryInputSource {
+
+	public static final String RESTART_KEY = "CompositeKeySqlDrivingQueryInputSource.key";
+
+	JdbcTemplate jdbcTemplate;
+
+	RowMapper keyMapper;
+
+	String drivingQuery;
+
+	String restartQuery;
+
+	CompositeKeyRestartDataConverter restartDataConverter;
+
+	public CompositeKeySqlDrivingQueryInputSource() {
+		super();
+	}
+
+	/**
+	 * Construct a new InputSource.
+	 *
+	 * @param jdbcTemplate
+	 * @param drivingQuery - Sql statement that returns all keys to process.
+	 * @param keyMapper - RowMapper that maps each row of the ResultSet to an object.
+	 */
+	public CompositeKeySqlDrivingQueryInputSource(JdbcTemplate jdbcTemplate,
+			String drivingQuery, RowMapper keyMapper){
+		this();
+		Assert.notNull(jdbcTemplate, "The JdbcTemplate must not be null.");
+		Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty.");
+		Assert.notNull(keyMapper, "The key RowMapper must not be null.");
+		this.jdbcTemplate = jdbcTemplate;
+		this.drivingQuery = drivingQuery;
+		this.keyMapper = keyMapper;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryInputSource#retrieveKeys()
+	 */
+	protected List retrieveKeys() {
+		return jdbcTemplate.query(drivingQuery, keyMapper);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.io.sql.scratch.AbstractDrivingQueryInputSource#restoreKeys(org.springframework.batch.restart.RestartData)
+	 */
+	protected List restoreKeys(RestartData restartData) {
+
+		Assert.state(restartDataConverter != null, "RestartDataConverter must not be null.");
+		Assert.state(StringUtils.hasText(restartQuery), "The RestartQuery must not be null or empty" +
+		" in order to restart.");
+
+		if (restartData.getProperties() != null) {
+			return jdbcTemplate.query(restartQuery, restartDataConverter.createArguments(restartData), keyMapper);
+		}
+
+		return new ArrayList();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.restart.Restartable#getRestartData()
+	 */
+	public RestartData getRestartData() {
+		Assert.state(restartDataConverter != null, "RestartDataConverter must not be null.");
+		return restartDataConverter.createRestartData(getCurrentKey());
+	}
+
+	/**
+	 * Set the {@link RestartDataConverter} used to convert a composite key to
+	 * RestartData and back again.
+	 *
+	 * @param restartDataConverter
+	 */
+	public void setRestartDataConverter(
+			CompositeKeyRestartDataConverter restartDataConverter) {
+		this.restartDataConverter = restartDataConverter;
+	}
+
+	/**
+	 * Set the query to use to retrieve keys in order to restore the previous
+	 * state for restart.
+	 *
+	 * @param restartQuery
+	 */
+	public void setRestartQuery(String restartQuery) {
+		this.restartQuery = restartQuery;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(jdbcTemplate, "The JdbcTemplate must not be null.");
+		Assert.hasText(drivingQuery, "The DrivingQuery must not be null or empty.");
+		Assert.notNull(keyMapper, "The key RowMapper must not be null.");
+	}
+}

infrastructure/src/main/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSource.java

@@ -1,219 +0,0 @@
-/*
- * Copyright 2006-2007 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.batch.io.sql;
-
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Properties;
-
-import javax.sql.DataSource;
-
-import org.springframework.batch.io.InputSource;
-import org.springframework.batch.item.ResourceLifecycle;
-import org.springframework.batch.restart.GenericRestartData;
-import org.springframework.batch.restart.RestartData;
-import org.springframework.batch.restart.Restartable;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.util.Assert;
-
-/**
- * <p>
- * DrivingQuery based input source for input data that can be uniquely
- * identified by a single primary key. The current implementation is
- * forward-only, and requires a transactional buffer to ensure rollbacks are
- * handled correctly.
- * </p>
- * 
- * <p>
- * Users of this input source must provide a 'Driving Query' that returns only
- * one column. (If more than one is returned, the first column will be used) A
- * 'Details Query' must then be provided that requires only one parameter.
- * (question mark) Invalid queries will throw SqlExceptions from JdbcTemplate.
- * <p>
- * 
- * 
- * @author Lucas Ward
- * 
- */
-public class SingleKeySqlDrivingQueryInputSource implements ResourceLifecycle, InputSource, Restartable {
-
-	private static final String RESTART_KEY = "SingleKeySqlDrivingQueryInputSource.lastProcessedKey";
-
-	private DataSource dataSource;
-
-	private JdbcTemplate jdbcTemplate;
-
-	private String drivingQuery;
-
-	private String detailsQuery;
-
-	private String restartQuery;
-
-	private Object[] detailArgs = new Object[1];
-
-	private List keys;
-
-	private Iterator keysIterator;
-
-	private RowMapper mapper;
-
-	/**
-	 * Read one record by passing in the current key to the details query.
-	 * 
-	 */
-	public Object read() {
-		if (keys == null) {
-			retrieveKeys();
-		}
-
-		if (keysIterator.hasNext()) {
-			detailArgs[0] = keysIterator.next();
-
-			return jdbcTemplate.queryForObject(detailsQuery, detailArgs, mapper);
-		}
-
-		return null;
-	}
-
-	/*
-	 * Retrieve the keys by calling the DrivingQuery.
-	 */
-	private void retrieveKeys() {
-
-		jdbcTemplate = new JdbcTemplate(dataSource);
-
-		keys = jdbcTemplate.query(drivingQuery, new SingleColumnRowMapper());
-
-		keysIterator = keys.iterator();
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.batch.item.ResourceLifecycle#close()
-	 */
-	public void close() {
-		keys = null;
-		keysIterator = null;
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.batch.item.ResourceLifecycle#open()
-	 */
-	public void open() {
-
-	}
-
-	public void setDataSource(DataSource dataSource) {
-		this.dataSource = dataSource;
-	}
-
-	/**
-	 * Set the query to be used to obtain the list of keys at
-	 * initialization.  Each key returned will be fed into the
-	 * details query.
-	 * 
-	 * @param drivingQuery
-	 */
-	public void setDrivingQuery(String drivingQuery) {
-		this.drivingQuery = drivingQuery;
-	}
-
-	/**
-	 * Set the query to be used for each 'detail' record.  Meaning,
-	 * the query each key (row returned from the driving query) will
-	 * be fed into in order to return a row to be mapped. 
-	 * 
-	 * @param detailsQuery
-	 */
-	public void setDetailsQuery(String detailsQuery) {
-		this.detailsQuery = detailsQuery;
-	}
-
-	/**
-	 * Set the query to be used in the case of a restart.  The current
-	 * key at the time restart data is requested will be fed into this
-	 * query as a parameter upon restart, allowing for only the remaining
-	 * keys to be returned.
-	 * 
-	 * @param restartQuery
-	 */
-	public void setRestartQuery(String restartQuery) {
-		this.restartQuery = restartQuery;
-	}
-
-	/**
-	 * Set RowMapper to be used for each call to the provided details
-	 * query.
-	 * 
-	 * @param mapper
-	 */
-	public void setMapper(RowMapper mapper) {
-		this.mapper = mapper;
-	}
-
-	// Required because JdbcTemplate.queryForList returns a list
-	// of maps based on metadata.
-	private class SingleColumnRowMapper implements RowMapper {
-
-		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-			return rs.getObject(1);
-		}
-
-	}
-
-	public RestartData getRestartData() {
-		Properties props = new Properties();
-		props.setProperty(RESTART_KEY, detailArgs[0].toString());
-		return new GenericRestartData(props);
-	}
-
-	/**
-	 * Restore input source to previous state. If the input source has already
-	 * been initialized before calling restore (meaning, read has been called)
-	 * then an IllegalStateException will be thrown, since all input sources
-	 * should be restored before being read from, otherwise already processed
-	 * data could be returned. The RestartData attempting to be restored from
-	 * must have been obtained from the <strong>same input source as the one
-	 * being restored from</strong> otherwise it is invalid.
-	 * 
-	 * @param RestartData obtained by calling getRestartData during a previous
-	 * run.
-	 * @throws IllegalStateException if input source has already been read from.
-	 */
-	public void restoreFrom(RestartData data) {
-
-		Assert.notNull(data, "RestartData must not be null.");
-		
-		if (keys != null) {
-			throw new IllegalStateException("Cannot restore when already intialized.  Call"
-					+ " close() first before restore()");
-		}
-
-		Properties restartData = data.getProperties();
-		String lastProcessedKey = restartData.getProperty(RESTART_KEY);
-		if (lastProcessedKey != null) {
-			jdbcTemplate = new JdbcTemplate(dataSource);
-
-			keys = jdbcTemplate.query(restartQuery, new Object[] { lastProcessedKey }, new SingleColumnRowMapper());
-
-			keysIterator = keys.iterator();
-		}
-	}
-}

infrastructure/src/main/java/org/springframework/batch/io/support/AbstractDrivingQueryInputSource.java

@@ -0,0 +1,194 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.support;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.item.ResourceLifecycle;
+import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.util.Assert;
+
+/**
+ * <p>Abstract base class for driving query input sources.  Input Sources of
+ * this type use a 'driving query' to return back a list of keys.  Upon each
+ * call to read, a new key is returned.</p>
+ *
+ * <p>Mutability: Because this base class cannot guarantee that the keys returned
+ * by subclasses are immutable, care should be taken to not modify a key value.
+ * Doing so would cause issues if a rollback occurs.  For example, if a call
+ * to read() is made, and the returned key is modified, a rollback will cause
+ * the next call to read() to return the same object that was originally returned,
+ * since there is no way to create a defensive copy, and re-querying the database
+ * for all the keys would be too resource intensive.</p>
+ *
+ *
+ * @author Lucas Ward
+ *
+ */
+public abstract class AbstractDrivingQueryInputSource implements InputSource, ResourceLifecycle,
+	DisposableBean, Restartable, InitializingBean {
+
+	private boolean initialized = false;
+
+	private List keys;
+
+	private Iterator keysIterator;
+
+	private int currentIndex = 0;
+
+	private int lastCommitIndex = 0;
+
+	private TransactionSynchronization synchronization =
+		new DrivingQueryInputSourceTransactionSynchronization();
+
+	/**
+	 * Return the next key in the List.  If the InputSource has not been initialized yet,
+	 * then {@link AbstractDrivingQueryInputSource.open()} will be called.
+	 *
+	 * @return next key in the list if not index is not at the last element, null otherwise.
+	 */
+	public Object read() {
+		if (!initialized) {
+			open();
+		}
+
+		if (keysIterator.hasNext()) {
+			currentIndex++;
+			return keysIterator.next();
+		}
+
+		return null;
+	}
+
+	/**
+	 * Get the current key.  This method will return the same
+	 * object returned by the last read() method.  If the
+	 * InputSource hasn't been initialized yet, then null will
+	 * be returned.
+	 *
+	 * @return the current key.
+	 */
+	protected Object getCurrentKey(){
+		if(initialized){
+			return keys.get(currentIndex - 1);
+		}
+
+		return null;
+	}
+
+	/**
+	 * Close the resource by setting the list of keys to null, allowing them
+	 * to be garbage collected.
+	 */
+	public void close() {
+		initialized = false;
+		currentIndex = 0;
+		lastCommitIndex = 0;
+		keys = null;
+		keysIterator = null;
+	}
+
+	/**
+	 * Initialize the input source by delegating to the subclass in order to retrieve
+	 * the keys.  The input source will also be registered with the
+	 * {@link BatchTransactionSynchronizationManager} in order to ensure it is notified
+	 * about commits and rollbacks.
+	 *
+	 * @throws IllegalStateException if the keys list is null or initialized is true.
+	 */
+	public void open() {
+
+		Assert.state(keys == null || initialized, "Cannot open an already opened input source" +
+				", call close() first.");
+		keys = retrieveKeys();
+		keysIterator = keys.listIterator();
+		BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
+		initialized = true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
+	public void destroy() throws Exception {
+		close();
+	}
+
+	/**
+	 * Restore input source to previous state. If the input source has already
+	 * been initialized before calling restore (meaning, read has been called)
+	 * then an IllegalStateException will be thrown, since all input sources
+	 * should be restored before being read from, otherwise already processed
+	 * data could be returned. The RestartData attempting to be restored from
+	 * must have been obtained from the <strong>same input source as the one
+	 * being restored from</strong> otherwise it is invalid.
+	 *
+	 * @throws IllegalArgumentException if restart data or it's properties is null.
+	 * @throws IllegalStateException if the input source has already been intialized.
+	 */
+	public void restoreFrom(RestartData data) {
+
+		Assert.notNull(data, "RestartData must not be null.");
+		Assert.notNull(data.getProperties(), "RestartData properties must not be null.");
+		Assert.state(!initialized, "Cannot restore when already intialized.  Call"
+					+ " close() first before restore()");
+
+		if (data.getProperties().size() == 0) {
+			return;
+		}
+
+		keys = restoreKeys(data);
+
+		if(keys != null & keys.size() > 0){
+			keysIterator = keys.listIterator();
+			initialized = true;
+		}
+	}
+
+	//Abstract Methods
+	/**
+	 * @return list of keys returned by the driving query
+	 */
+	protected abstract List retrieveKeys();
+
+	/**
+	 * Restore the keys list based on provided restart data.
+	 *
+	 * @param restartData, the restart data to restore the keys list from.
+	 * @return a list of keys.
+	 */
+	protected abstract List restoreKeys(RestartData restartData);
+
+	/**
+	 * Encapsulates transaction events handling.
+	 */
+	private class DrivingQueryInputSourceTransactionSynchronization extends TransactionSynchronizationAdapter {
+		public void afterCompletion(int status) {
+			if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
+				keysIterator = keys.listIterator(lastCommitIndex);
+			} else if (status == TransactionSynchronization.STATUS_COMMITTED) {
+				lastCommitIndex = currentIndex;
+			}
+		}
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/orm/hibernate/HibernateInputSourceIntegrationTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.io.orm.hibernate;
+
+import org.hibernate.SessionFactory;
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.orm.hibernate.HibernateInputSource;
+import org.springframework.batch.io.support.AbstractDataSourceInputSourceIntegrationTests;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.orm.hibernate3.LocalSessionFactoryBean;
+
+/**
+ * Tests for {@link HibernateInputSource}
+ * 
+ * @author Robert Kasanicky
+ */
+public class HibernateInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests {
+
+	protected InputSource createInputSource() throws Exception {
+		LocalSessionFactoryBean factoryBean = new LocalSessionFactoryBean();
+		factoryBean.setDataSource(super.getJdbcTemplate().getDataSource());
+		factoryBean.setMappingLocations(new Resource[]{new ClassPathResource("Foo.hbm.xml", getClass())});
+		factoryBean.afterPropertiesSet();
+		
+		SessionFactory sessionFactory = (SessionFactory) factoryBean.getObject();
+		
+		String hsqlQuery = "from Foo";
+		
+		HibernateInputSource inputSource = new HibernateInputSource();
+		inputSource.setQueryString(hsqlQuery);
+		inputSource.setSessionFactory(sessionFactory);
+		inputSource.afterPropertiesSet();
+		
+		return inputSource;
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/orm/ibatis/IbatisInputSourceIntegrationTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.io.orm.ibatis;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.orm.ibatis.IbatisDrivingQueryInputSource;
+import org.springframework.batch.io.support.AbstractDataSourceInputSourceIntegrationTests;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.orm.ibatis.SqlMapClientFactoryBean;
+
+import com.ibatis.sqlmap.client.SqlMapClient;
+
+/**
+ * Tests for {@link IbatisDrivingQueryInputSource}
+ *
+ * @author Robert Kasanicky
+ */
+public class IbatisInputSourceIntegrationTests extends AbstractDataSourceInputSourceIntegrationTests {
+
+	protected InputSource createInputSource() throws Exception {
+
+		SqlMapClientFactoryBean factory = new SqlMapClientFactoryBean();
+		factory.setConfigLocation(new ClassPathResource("ibatis-config.xml", getClass()));
+		factory.setDataSource(super.getJdbcTemplate().getDataSource());
+		factory.afterPropertiesSet();
+		SqlMapClient sqlMapClient = (SqlMapClient) factory.getObject();
+
+		IbatisInputSource inputSource = new IbatisInputSource();
+		inputSource.setDrivingQueryId("getAllFooIds");
+		inputSource.setDetailsQueryId("getFooById");
+		inputSource.setRestartQueryId("getAllFooIdsRestart");
+		inputSource.setSqlMapClient(sqlMapClient);
+
+		return inputSource;
+	}
+
+
+}

infrastructure/src/test/java/org/springframework/batch/io/sample/domain/Foo.java

@@ -0,0 +1,54 @@
+package org.springframework.batch.io.sample.domain;
+
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+
+/**
+ * Simple domain object for testing purposes.
+ */
+public class Foo {
+	
+	private int id;
+	private String name;
+	private int value;
+	
+	public Foo(){}
+	
+	public Foo(int id, String name, int value) {
+		this.id = id;
+		this.name = name;
+		this.value = value;
+	}
+	
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
+	public int getValue() {
+		return value;
+	}
+	public void setValue(int value) {
+		this.value = value;
+	}
+	public int getId() {
+		return id;
+	}
+	public void setId(int id) {
+		this.id = id;
+	}
+	
+	public String toString() {
+		return "Foo[id=" +id +",name=" + name + ",value=" + value + "]";
+	}
+	
+	public boolean equals(Object obj) {
+		return EqualsBuilder.reflectionEquals(this, obj);
+	}
+	
+	public int hashCode() {
+		return HashCodeBuilder.reflectionHashCode(this);
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractSqlInputSourceIntegrationTests.java

@@ -0,0 +1,183 @@
+package org.springframework.batch.io.sql;
+
+import java.util.Properties;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
+import org.springframework.batch.restart.GenericRestartData;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationUtils;
+import org.springframework.util.Assert;
+
+/**
+ * Common scenarios for testing {@link InputSource} implementations which read data from database.
+ *
+ * @author Lucas Ward
+ * @author Robert Kasanicky
+ */
+public abstract class AbstractSqlInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests {
+
+	protected InputSource source;
+
+
+	/**
+	 * @return input source with all necessary dependencies set
+	 */
+	protected abstract InputSource createInputSource() throws Exception;
+
+	protected String[] getConfigLocations(){
+		return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"};
+	}
+
+	protected void onSetUp()throws Exception{
+		super.onSetUp();
+		BatchTransactionSynchronizationManager.clearSynchronizations();
+		source = createInputSource();
+		getAsInitializingBean(source).afterPropertiesSet();
+	}
+
+	protected void onTearDown()throws Exception {
+		getAsDisposableBean(source).destroy();
+		BatchTransactionSynchronizationManager.clearSynchronizations();
+		super.onTearDown();
+	}
+
+	/**
+	 * Regular scenario - read all rows and eventually return null.
+	 */
+	public void testNormalProcessing() throws Exception {
+		getAsInitializingBean(source).afterPropertiesSet();
+
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		Foo foo3 = (Foo) source.read();
+		assertEquals(3, foo3.getValue());
+
+		Foo foo4 = (Foo) source.read();
+		assertEquals(4, foo4.getValue());
+
+		Foo foo5 = (Foo) source.read();
+		assertEquals(5, foo5.getValue());
+
+		assertNull(source.read());
+	}
+
+	/**
+	 * Restart scenario.
+	 * @throws Exception
+	 */
+	public void testRestart() throws Exception {
+
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		RestartData restartData = getAsRestartable(source).getRestartData();
+
+		// create new input source
+		source = createInputSource();
+
+		getAsRestartable(source).restoreFrom(restartData);
+
+		Foo fooAfterRestart = (Foo) source.read();
+		assertEquals(3, fooAfterRestart.getValue());
+	}
+
+	/**
+	 * Reading from an input source and then trying to restore causes an error.
+	 */
+	public void testInvalidRestore() throws Exception {
+
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		RestartData restartData = getAsRestartable(source).getRestartData();
+
+		// create new input source
+		source = createInputSource();
+
+		Foo foo = (Foo) source.read();
+		assertEquals(1, foo.getValue());
+
+		try {
+			getAsRestartable(source).restoreFrom(restartData);
+			fail();
+		}
+		catch (IllegalStateException ex) {
+			// expected
+		}
+	}
+
+	/**
+	 * Empty restart data should be handled gracefully.
+	 */
+	public void testRestoreFromEmptyData() {
+		RestartData restartData = new GenericRestartData(new Properties());
+
+		getAsRestartable(source).restoreFrom(restartData);
+
+		Foo foo = (Foo) source.read();
+		assertEquals(1, foo.getValue());
+	}
+
+	/**
+	 * Rollback scenario.
+	 */
+	public void testRollback() {
+		Foo foo1 = (Foo) source.read();
+
+		commit();
+
+		Foo foo2 = (Foo) source.read();
+		Assert.state(!foo2.equals(foo1));
+
+		Foo foo3 = (Foo) source.read();
+		Assert.state(!foo2.equals(foo3));
+
+		rollback();
+
+		assertEquals(foo2, source.read());
+	}
+
+
+	private void commit() {
+		TransactionSynchronizationUtils.invokeAfterCompletion(
+				TransactionSynchronizationManager.getSynchronizations(),
+				TransactionSynchronization.STATUS_COMMITTED);
+	}
+
+	private void rollback() {
+		TransactionSynchronizationUtils.invokeAfterCompletion(
+				TransactionSynchronizationManager.getSynchronizations(),
+				TransactionSynchronization.STATUS_ROLLED_BACK);
+	}
+
+	private Restartable getAsRestartable(InputSource source) {
+		return (Restartable) source;
+	}
+
+	private InitializingBean getAsInitializingBean(InputSource source) {
+		return (InitializingBean) source;
+	}
+
+	private DisposableBean getAsDisposableBean(InputSource source) {
+		return (DisposableBean) source;
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeyFooDao.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.sql;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.support.JdbcDaoSupport;
+
+/**
+ * @author Lucas Ward
+ *
+ */
+public class CompositeKeyFooDao extends JdbcDaoSupport implements FooDao {
+
+	public CompositeKeyFooDao(JdbcTemplate jdbcTemplate) {
+		this.setJdbcTemplate(jdbcTemplate);
+	}
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.io.sql.scratch.FooDao#getFoo(java.lang.Object)
+	 */
+	public Foo getFoo(Object key) {
+
+		List keys = (List)key;
+		Object[] args = new Object[]{keys.get(0), keys.get(1)};
+
+		RowMapper fooMapper = new RowMapper(){
+			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+				Foo foo = new Foo();
+				foo.setId(rs.getInt(1));
+				foo.setName(rs.getString(2));
+				foo.setValue(rs.getInt(3));
+				return foo;
+			}
+		};
+
+		return (Foo)getJdbcTemplate().query("SELECT ID, NAME, VALUE from T_FOOS where ID = ? and VALUE = ?",
+				args, fooMapper).get(0);
+	}
+
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/CompositeKeySqlDrivingQueryInputSourceIntegrationTests.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.sql;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.sql.CompositeKeySqlDrivingQueryInputSource;
+import org.springframework.batch.restart.GenericRestartData;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.jdbc.core.RowMapper;
+
+/**
+ * @author Lucas Ward
+ *
+ */
+public class CompositeKeySqlDrivingQueryInputSourceIntegrationTests extends
+		AbstractSqlInputSourceIntegrationTests {
+
+	protected InputSource createInputSource() throws Exception {
+
+		CompositeKeySqlDrivingQueryInputSource inputSource =
+			new CompositeKeySqlDrivingQueryInputSource(getJdbcTemplate(),
+					"SELECT ID, VALUE from T_FOOS order by ID, VALUE",
+					new FooCompositeKeyMapper());
+
+		inputSource.setRestartQuery("SELECT ID from T_FOOS where ID > ? and VALUE > ? order by ID");
+		inputSource.setRestartDataConverter(new FooRestartDataConverter());
+		FooInputSource fooInputSource = new FooInputSource(inputSource, getJdbcTemplate());
+		fooInputSource.setFooDao(new CompositeKeyFooDao(getJdbcTemplate()));
+		return fooInputSource;
+	}
+
+	private class FooRestartDataConverter implements CompositeKeyRestartDataConverter{
+
+		private static final String ID_RESTART_KEY = "FooRestartDataConverter.id";
+		private static final String VALUE_RESTART_KEY = "FooRestartDataConverter.value";
+
+		public RestartData createRestartData(Object compositeKey) {
+
+			List values = (List)compositeKey;
+			Properties data = new Properties();
+			data.setProperty(ID_RESTART_KEY, values.get(0).toString());
+			data.setProperty(VALUE_RESTART_KEY, values.get(1).toString());
+			return new GenericRestartData(data);
+		}
+
+		public Object[] createArguments(RestartData restartData) {
+			Object[] args = new Object[2];
+			args[0] = restartData.getProperties().get(ID_RESTART_KEY);
+			args[1] = restartData.getProperties().getProperty(VALUE_RESTART_KEY);
+			return args;
+		}
+	}
+
+	private class FooCompositeKeyMapper implements RowMapper{
+		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+			List key = new ArrayList();
+			key.add(new Long(rs.getLong(1)));
+			key.add(new Long(rs.getLong(2)));
+			return key;
+		}
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/FooDao.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.io.sql;
+
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+/**
+ * @author Lucas Ward
+ *
+ */
+public interface FooDao {
+
+	Foo getFoo(Object key);
+
+	void setJdbcTemplate(JdbcTemplate jdbcTemplate);
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/FooInputSource.java

@@ -0,0 +1,48 @@
+package org.springframework.batch.io.sql;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.support.AbstractDrivingQueryInputSource;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+class FooInputSource implements InputSource, Restartable, DisposableBean, InitializingBean{
+
+	AbstractDrivingQueryInputSource inputSource;
+	FooDao fooDao = new SingleKeyFooDao();
+
+	public FooInputSource(AbstractDrivingQueryInputSource inputSource, JdbcTemplate jdbcTemplate) {
+		this.inputSource = inputSource;
+		fooDao.setJdbcTemplate(jdbcTemplate);
+	}
+
+	public Object read() {
+		Object key = inputSource.read();
+		if(key != null){
+			return fooDao.getFoo(key);
+		}else{
+			return null;
+		}
+	}
+
+	public RestartData getRestartData() {
+		return inputSource.getRestartData();
+	}
+
+	public void restoreFrom(RestartData data) {
+		inputSource.restoreFrom(data);
+	}
+
+	public void destroy() throws Exception {
+		inputSource.destroy();
+	}
+
+	public void setFooDao(FooDao fooDao) {
+		this.fooDao = fooDao;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+	};
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/FooRowMapper.java

@@ -0,0 +1,21 @@
+package org.springframework.batch.io.sql;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.jdbc.core.RowMapper;
+
+
+public class FooRowMapper implements RowMapper {
+
+		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+
+			Foo foo = new Foo();
+			foo.setId(rs.getInt(1));
+			foo.setName(rs.getString(2));
+			foo.setValue(rs.getInt(3));
+			
+			return foo;
+		}
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeyFooDao.java

@@ -0,0 +1,28 @@
+package org.springframework.batch.io.sql;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.support.JdbcDaoSupport;
+
+public class SingleKeyFooDao extends JdbcDaoSupport implements FooDao {
+
+	public Foo getFoo(Object key){
+
+		RowMapper fooMapper = new RowMapper(){
+			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+				Foo foo = new Foo();
+				foo.setId(rs.getInt(1));
+				foo.setName(rs.getString(2));
+				foo.setValue(rs.getInt(3));
+				return foo;
+			}
+		};
+
+		return (Foo)getJdbcTemplate().query("SELECT ID, NAME, VALUE from T_FOOS where ID = ?",
+				new Object[] {key}, fooMapper).get(0);
+
+	}
+}

infrastructure/src/test/java/org/springframework/batch/io/sql/SingleKeySqlDrivingQueryInputSourceIntegrationTests.java

@@ -1,152 +0,0 @@
-package org.springframework.batch.io.sql;
-
-import java.sql.ResultSet;
-import java.sql.SQLException;
-
-import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
-import org.springframework.batch.restart.RestartData;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
-
-public class SingleKeySqlDrivingQueryInputSourceIntegrationTests 
-	extends AbstractTransactionalDataSourceSpringContextTests {
-
-	SingleKeySqlDrivingQueryInputSource sqlInputSource;
-	
-	protected String[] getConfigLocations(){
-		return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"};
-	}
-	
-	protected void onSetUp()throws Exception{
-		super.onSetUp();
-		
-		sqlInputSource = createInputSource();
-	}
-	
-	protected SingleKeySqlDrivingQueryInputSource createInputSource(){
-		
-		SingleKeySqlDrivingQueryInputSource inputSource = new SingleKeySqlDrivingQueryInputSource();
-		inputSource.setDrivingQuery("SELECT ID from T_FOOS order by ID");
-		inputSource.setDetailsQuery("SELECT NAME, VALUE from T_FOOS where ID = ?");
-		inputSource.setRestartQuery("SELECT ID from T_FOOS where ID > ? order by ID");
-		inputSource.setMapper(new FooMapper());
-		inputSource.setDataSource(super.getJdbcTemplate().getDataSource());
-		return inputSource;
-	}
-	
-	protected void onTearDown()throws Exception{
-
-		BatchTransactionSynchronizationManager.clearSynchronizations();
-		sqlInputSource.close();
-		super.onTearDown();
-	}
-
-	public void testNormalProcessing(){
-		
-		Foo foo = (Foo)sqlInputSource.read();
-		assertEquals(1, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(2, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(3, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(4, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(5, foo.value);
-		
-		assertNull(sqlInputSource.read());
-	}
-	
-/*	public void testRollback(){
-		
-		SqlIdentityKey key = sqlInputSource.readKey();
-		assertEquals("1", key.getKeyValue("id"));
-		
-		key = sqlInputSource.readKey();
-		assertEquals("2", key.getKeyValue("id"));
-		
-		super.setComplete();
-		super.endTransaction();
-		super.startNewTransaction();
-		BatchTransactionSynchronizationManager.resynchronize();
-		
-		key = sqlInputSource.readKey();
-		assertEquals("3", key.getKeyValue("id"));
-		
-		key = sqlInputSource.readKey();
-		assertEquals("4", key.getKeyValue("id"));
-		
-		super.endTransaction();
-		super.startNewTransaction();
-		
-		key = sqlInputSource.readKey();
-		assertEquals("3", key.getKeyValue("id"));
-		
-	}*/
-	
-	public void testRestart(){
-		
-		Foo foo = (Foo)sqlInputSource.read();
-		assertEquals(1, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(2, foo.value);
-		
-		RestartData restartData = sqlInputSource.getRestartData();
-		
-		//create new input source
-		sqlInputSource = createInputSource();
-		
-		sqlInputSource.restoreFrom(restartData);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(3, foo.value);
-	}
-	
-	//test that reading from an input source and then trying to restore causes an error.
-	public void testInvalidRestore(){
-		
-		Foo foo = (Foo)sqlInputSource.read();
-		assertEquals(1, foo.value);
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(2, foo.value);
-		
-		RestartData restartData = sqlInputSource.getRestartData();
-		
-		//create new input source
-		sqlInputSource = createInputSource();
-		
-		foo = (Foo)sqlInputSource.read();
-		assertEquals(1, foo.value);
-		
-		try{
-			sqlInputSource.restoreFrom(restartData);
-			fail();
-		}
-		catch(IllegalStateException ex){
-			//expected
-		}
-	}
-	
-	private class Foo {
-		String name;
-		int value;
-	}
-	
-	private class FooMapper implements RowMapper{
-
-		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-			
-			Foo foo = new Foo();
-			foo.name = rs.getString(1);
-			foo.value = rs.getInt(2);
-			return foo;
-		}
-		
-	}
-}

infrastructure/src/test/java/org/springframework/batch/io/sql/SqlCursorInputSourceIntegrationTests.java

@@ -1,336 +1,27 @@
 package org.springframework.batch.io.sql;
 
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.Properties;
-
-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.restart.GenericRestartData;
-import org.springframework.batch.restart.RestartData;
-import org.springframework.dao.DataAccessException;
-import org.springframework.dao.InvalidDataAccessResourceUsageException;
-import org.springframework.jdbc.core.RowMapper;
-import org.springframework.jdbc.datasource.SingleConnectionDataSource;
-import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
+import org.springframework.batch.io.InputSource;
 
 /**
- *
- * @author Lucas Ward
- *
+ * Tests for {@link SqlCursorInputSource}
+ * 
+ * @author Robert Kasanicky
  */
-public class SqlCursorInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests {
-
-	protected SqlCursorInputSource sqlCursorInputSource;
-
-	private RowMapper mapper = new SqlRowMapper();
-
-	private static final String CURRENT_PROCESSED_ROW = "sqlCursorInput.lastProcessedRowNum";
-	private static final String SKIP_COUNT = "sqlCursorInput.skippedRrecordCount";
-
-	protected String[] getConfigLocations(){
-		return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"};
-	}
-
-	protected void onSetUp()throws Exception{
-		sqlCursorInputSource = getNewInputSource();
-		sqlCursorInputSource.setMapper(mapper);
-		sqlCursorInputSource.setFetchSize(10);
-		sqlCursorInputSource.setMaxRows(100);
-		sqlCursorInputSource.setQueryTimeout(1000);
-		RepeatSynchronizationManager.register(new RepeatContextSupport(null));
-		super.onSetUp();
-	}
-
-	protected void onTearDown()throws Exception{
-		//cursor must be closed between each test, and transaction synchronization
-		//list must be cleared.
-		BatchTransactionSynchronizationManager.clearSynchronizations();
-		RepeatSynchronizationManager.clear();
-		sqlCursorInputSource.destroy();
-		super.onTearDown();
-	}
-
-	protected SqlCursorInputSource getNewInputSource(){
+public class SqlCursorInputSourceIntegrationTests extends AbstractSqlInputSourceIntegrationTests{
 
+	protected InputSource createInputSource() throws Exception {
 		SqlCursorInputSource result = new SqlCursorInputSource();
 		result.setDataSource(super.getJdbcTemplate().getDataSource());
-		result.setSql("SELECT * from T_FOOS");
+		result.setSql("select ID, NAME, VALUE from T_FOOS");
 		result.setIgnoreWarnings(true);
 		result.setVerifyCursorPosition(true);
+		
+		result.setMapper(new FooRowMapper());
+		result.setFetchSize(10);
+		result.setMaxRows(100);
+		result.setQueryTimeout(1000);
 
 		return result;
 	}
 
-	public void testAfterPropertiesSet() throws Exception{
-		//all dependencies have been set, so there shouldn't be any exceptions
-		sqlCursorInputSource.afterPropertiesSet();
-
-		sqlCursorInputSource.setSql(null);
-
-		try{
-			sqlCursorInputSource.afterPropertiesSet();
-			fail();
-		}catch(IllegalArgumentException ex){
-			//expected
-		}
-	}
-
-	public void testNormalReading(){
-
-		int fooCount = 0;
-
-		for(;;){
-			Foo foo = (Foo)sqlCursorInputSource.read();
-
-			if( foo == null){
-				break;
-			}
-
-			fooCount++;
-			validateFoo(fooCount, "bar" + fooCount, fooCount, foo);
-			assertEquals(sqlCursorInputSource.getCurrentProcessedRow(), fooCount);
-
-		}
-
-		assertEquals(5, fooCount );
-	}
-
-	public void testModifyCursorPosition(){
-
-		sqlCursorInputSource.setMapper(new RowMapper(){
-
-			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-				rs.next();
-				return null;
-			}});
-
-		try{
-			sqlCursorInputSource.read();
-			fail();
-		}catch(InvalidDataAccessResourceUsageException ex){
-			//expected
-		}
-	}
-
-
-	public void testRestart(){
-
-		sqlCursorInputSource.read();
-		Foo foo = (Foo)sqlCursorInputSource.read();
-
-		validateFoo(2, "bar2", 2, foo);
-
-		RestartData restartData = sqlCursorInputSource.getRestartData();
-
-		sqlCursorInputSource = getNewInputSource();
-		sqlCursorInputSource.setMapper(mapper);
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(1, "bar1", 1, foo);
-
-		sqlCursorInputSource.restoreFrom(restartData);
-
-		foo = (Foo)sqlCursorInputSource.read();
-
-		validateFoo(3, "bar3", 3, foo);
-	}
-
-	public void testRestartWithEmptyRestartData(){
-
-		RestartData restartData = new GenericRestartData(new Properties());
-		sqlCursorInputSource.restoreFrom(restartData);
-	}
-
-	public void testReadWithNullMapper(){
-
-		//calling read without a mapper should throw an exception.
-		sqlCursorInputSource.setMapper(null);
-		try{
-			sqlCursorInputSource.read();
-			fail();
-		}
-		catch(IllegalStateException ex){
-			//expected
-		}
-	}
-
-	public void testStatistics(){
-
-		Properties statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("0", statistics.getProperty(CURRENT_PROCESSED_ROW));
-		sqlCursorInputSource.read();
-
-		statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("1", statistics.getProperty(CURRENT_PROCESSED_ROW));
-	}
-
-	public void testSkipCountStatistics(){
-
-		sqlCursorInputSource.read();
-		Foo foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(2, "bar2", 2, foo);
-
-		Properties statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("0", statistics.getProperty(SKIP_COUNT));
-
-		sqlCursorInputSource.skip();
-
-		statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("1", statistics.getProperty(SKIP_COUNT));
-
-		sqlCursorInputSource.read();
-
-		sqlCursorInputSource.read();
-		sqlCursorInputSource.skip();
-
-		statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("2", statistics.getProperty(SKIP_COUNT));
-
-		super.endTransaction();
-		super.startNewTransaction();
-
-		sqlCursorInputSource.read();
-
-		statistics = sqlCursorInputSource.getStatistics();
-		assertEquals("2", statistics.getProperty(SKIP_COUNT));
-	}
-
-	public void testRollback(){
-
-		sqlCursorInputSource.read();
-		Foo foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(2, "bar2", 2, foo);
-
-		super.setComplete();
-		super.endTransaction();
-		super.startNewTransaction();
-		BatchTransactionSynchronizationManager.resynchronize();
-
-		sqlCursorInputSource.read();
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(4, "bar4", 4, foo);
-
-		super.endTransaction();
-		super.startNewTransaction();
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(3, "bar3", 3, foo);
-	}
-
-	public void testSkip(){
-
-		sqlCursorInputSource.read();
-		Foo foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(2, "bar2", 2, foo);
-
-		sqlCursorInputSource.skip();
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(3, "bar3", 3, foo);
-
-		super.endTransaction();
-		super.startNewTransaction();
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(1, "bar1", 1, foo);
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(3, "bar3", 3, foo);
-	}
-
-	public void testSucessiveSkip(){
-
-		sqlCursorInputSource.read();
-		Foo foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(2, "bar2", 2, foo);
-		sqlCursorInputSource.skip();
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(3, "bar3", 3, foo);
-		sqlCursorInputSource.skip();
-
-		super.endTransaction();
-		super.startNewTransaction();
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(1, "bar1", 1, foo);
-
-		foo = (Foo)sqlCursorInputSource.read();
-		validateFoo(4, "bar4", 4, foo);
-	}
-
-	public void testMappingException(){
-
-		sqlCursorInputSource.setMapper(new RowMapper(){
-
-			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-				throw new SQLException();
-			}});
-
-		try{
-			sqlCursorInputSource.read();
-			fail();
-		}catch(DataAccessException ex){
-			//expected
-		}
-	}
-
-	public void testExecuteQueryException(){
-
-		sqlCursorInputSource.setDataSource(new ExceptionDataSource());
-
-		try{
-			sqlCursorInputSource.read();
-			fail();
-		}catch(DataAccessException ex){
-			//expected
-		}
-	}
-
-	public void testNoIgnoreWarnings(){
-
-		//there shouldn't be any exceptions if ignore warnings is false
-		//and there are no warnings.
-		sqlCursorInputSource.setIgnoreWarnings(false);
-		sqlCursorInputSource.read();
-	}
-
-
-	private void validateFoo(int id, String name, int value, Foo foo){
-
-		assertEquals(id, foo.id);
-		assertEquals(name, foo.name);
-		assertEquals(value, foo.value);
-	}
-
-	private class SqlRowMapper implements RowMapper {
-
-		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-
-			Foo foo = new Foo();
-			foo.id = rs.getInt(1);
-			foo.name = rs.getString(2);
-			foo.value = rs.getInt(3);
-
-			return foo;
-		}
-
-	}
-
-	private class Foo{
-
-		private int id;
-		private String name;
-		private int value;
-	}
-
-	private class ExceptionDataSource extends SingleConnectionDataSource{
-
-		public Connection getConnection() throws SQLException {
-			throw new SQLException();
-		}
-	}
 }

infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceInputSourceIntegrationTests.java

@@ -0,0 +1,183 @@
+package org.springframework.batch.io.support;
+
+import java.util.Properties;
+
+import org.springframework.batch.io.InputSource;
+import org.springframework.batch.io.sample.domain.Foo;
+import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
+import org.springframework.batch.restart.GenericRestartData;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationUtils;
+import org.springframework.util.Assert;
+
+/**
+ * Common scenarios for testing {@link InputSource} implementations which read data from database.
+ * 
+ * @author Lucas Ward
+ * @author Robert Kasanicky
+ */
+public abstract class AbstractDataSourceInputSourceIntegrationTests extends AbstractTransactionalDataSourceSpringContextTests {
+
+	protected InputSource source;
+	
+
+	/**
+	 * @return configured input source ready for use
+	 */
+	protected abstract InputSource createInputSource() throws Exception;
+	
+	protected String[] getConfigLocations(){
+		return new String[] { "org/springframework/batch/io/sql/data-source-context.xml"};
+	}
+	
+	protected void onSetUp()throws Exception{
+		super.onSetUp();
+		BatchTransactionSynchronizationManager.clearSynchronizations();
+		source = createInputSource();
+	}
+	
+	protected void onTearDown()throws Exception {
+		getAsDisposableBean(source).destroy();
+		BatchTransactionSynchronizationManager.clearSynchronizations();
+		super.onTearDown();
+	}
+
+	/**
+	 * Regular scenario - read all rows and eventually return null.
+	 */
+	public void testNormalProcessing() throws Exception {
+		getAsInitializingBean(source).afterPropertiesSet();
+		
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		Foo foo3 = (Foo) source.read();
+		assertEquals(3, foo3.getValue());
+
+		Foo foo4 = (Foo) source.read();
+		assertEquals(4, foo4.getValue());
+
+		Foo foo5 = (Foo) source.read();
+		assertEquals(5, foo5.getValue());
+
+		assertNull(source.read());
+	}
+
+	/**
+	 * Restart scenario - read records, save restart data, create new input source
+	 * and restore from restart data - the new input source should continue where
+	 * the old one finished.
+	 */
+	public void testRestart() throws Exception {
+
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		RestartData restartData = getAsRestartable(source).getRestartData();
+
+		// create new input source
+		source = createInputSource();
+
+		getAsRestartable(source).restoreFrom(restartData);
+
+		Foo fooAfterRestart = (Foo) source.read();
+		assertEquals(3, fooAfterRestart.getValue());
+	}
+
+	/**
+	 * Reading from an input source and then trying to restore causes an error.
+	 */
+	public void testInvalidRestore() throws Exception {
+
+		Foo foo1 = (Foo) source.read();
+		assertEquals(1, foo1.getValue());
+
+		Foo foo2 = (Foo) source.read();
+		assertEquals(2, foo2.getValue());
+
+		RestartData restartData = getAsRestartable(source).getRestartData();
+
+		// create new input source
+		source = createInputSource();
+
+		Foo foo = (Foo) source.read();
+		assertEquals(1, foo.getValue());
+
+		try {
+			getAsRestartable(source).restoreFrom(restartData);
+			fail();
+		}
+		catch (IllegalStateException ex) {
+			// expected
+		}
+	}
+	
+	/**
+	 * Empty restart data should be handled gracefully.
+	 */
+	public void testRestoreFromEmptyData() {
+		RestartData restartData = new GenericRestartData(new Properties());
+		
+		getAsRestartable(source).restoreFrom(restartData);
+		
+		Foo foo = (Foo) source.read();
+		assertEquals(1, foo.getValue());
+	}
+	
+	/**
+	 * Rollback scenario - input source rollbacks to last commit point.
+	 */
+	public void testRollback() {
+		Foo foo1 = (Foo) source.read();
+		
+		commit();
+		
+		Foo foo2 = (Foo) source.read();
+		Assert.state(!foo2.equals(foo1));
+		
+		Foo foo3 = (Foo) source.read();
+		Assert.state(!foo2.equals(foo3));
+		
+		rollback();
+		
+		assertEquals(foo2, source.read());
+	}
+	
+
+	private void commit() {
+		TransactionSynchronizationUtils.invokeAfterCompletion(
+				TransactionSynchronizationManager.getSynchronizations(), 
+				TransactionSynchronization.STATUS_COMMITTED);
+	}
+	
+	private void rollback() {
+		TransactionSynchronizationUtils.invokeAfterCompletion(
+				TransactionSynchronizationManager.getSynchronizations(), 
+				TransactionSynchronization.STATUS_ROLLED_BACK);
+	}
+	
+	private Restartable getAsRestartable(InputSource source) {
+		return (Restartable) source;
+	}
+	
+	private InitializingBean getAsInitializingBean(InputSource source) {
+		return (InitializingBean) source;
+	}
+	
+	private DisposableBean getAsDisposableBean(InputSource source) {
+		return (DisposableBean) source;
+	}
+
+}

infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo-write.hbm.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0"?>
+<!DOCTYPE hibernate-mapping PUBLIC
+        "-//Hibernate/Hibernate Mapping DTD 3.0//EN"
+        "http://hibernate.sourceforge.net/hibernate-mapping-3.0.dtd">
+
+<hibernate-mapping>
+	<class name="org.springframework.batch.io.sample.domain.Foo" table="T_WRITE_FOOS">
+        <id name="id" column="ID">
+            <generator class="increment"/>
+        </id>
+        <property name="name" />
+        <property name="value" />
+    </class>
+
+</hibernate-mapping>

infrastructure/src/test/resources/org/springframework/batch/io/orm/hibernate/Foo.hbm.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0"?>
+<!DOCTYPE hibernate-mapping PUBLIC
+        "-//Hibernate/Hibernate Mapping DTD 3.0//EN"
+        "http://hibernate.sourceforge.net/hibernate-mapping-3.0.dtd">
+
+<hibernate-mapping>
+	<class name="org.springframework.batch.io.sample.domain.Foo" table="T_FOOS">
+        <id name="id" column="ID">
+            <generator class="increment"/>
+        </id>
+        <property name="name" />
+        <property name="value" />
+    </class>
+
+</hibernate-mapping>

infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-config.xml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE sqlMapConfig PUBLIC "-//ibatis.apache.org//DTD SQL Map Config 2.0//EN"
+	"http://ibatis.apache.org/dtd/sql-map-config-2.dtd">
+
+<sqlMapConfig>
+	<!-- lazy loading is enabled by default, but it is emphasized here - if it was set
+		to false, input source would load all data into memory! -->
+	<settings lazyLoadingEnabled="true"/>
+
+	<sqlMap resource="org/springframework/batch/io/orm/ibatis/ibatis-foo.xml" />
+
+</sqlMapConfig>

infrastructure/src/test/resources/org/springframework/batch/io/orm/ibatis/ibatis-foo.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE sqlMap PUBLIC "-//ibatis.apache.org//DTD SQL Map 2.0//EN"
+	"http://ibatis.apache.org/dtd/sql-map-2.dtd">
+
+<sqlMap namespace="Foo">
+
+  <resultMap id="fooResult" class="org.springframework.batch.io.sample.domain.Foo">
+  	 <result property="id" column="ID" />
+     <result property="name" column="NAME" />
+     <result property="value" column="VALUE" />
+  </resultMap>
+
+  <select id="getAllFooIds" resultClass="int">
+  	select ID from T_FOOS
+  </select>
+
+  <select id="getFooById" parameterClass="int" resultMap="fooResult">
+  	select ID, NAME, VALUE from T_FOOS where ID = #value#
+  </select>
+
+  <select id="getAllFoos" resultMap="fooResult">
+  	select ID, NAME, VALUE from T_FOOS
+  </select>
+
+  <select id="getAllFooIdsRestart" resultClass="int">
+  	select ID from T_FOOS where ID > #id#
+  </select>
+
+  <insert id="insertFoo" parameterClass="org.springframework.batch.io.sample.domain.Foo">
+  	insert INTO T_WRITE_FOOS (ID, NAME, VALUE) VALUES (#id#, #name#, #value#)
+  </insert>
+
+  <update id="updateFoo" parameterClass="org.springframework.batch.io.sample.domain.Foo">
+  	update T_WRITE_FOOS set NAME = #name#, VALUE = #value# where ID = #id#
+  </update>
+
+  <delete id="deleteFoo" parameterClass="org.springframework.batch.io.sample.domain.Foo">
+  	delete from T_WRITE_FOOS where ID = #id#
+  </delete>
+
+</sqlMap>

infrastructure/src/test/resources/org/springframework/batch/io/sql/destroy-foo-schema-hsqldb.sql

@@ -1 +1,2 @@
-DROP TABLE T_FOOS;
+DROP TABLE T_FOOS;
+DROP TABLE T_WRITE_FOOS;

infrastructure/src/test/resources/org/springframework/batch/io/sql/init-foo-schema-hsqldb.sql

@@ -11,3 +11,11 @@ INSERT INTO t_foos (id, name, value) VALUES (2, 'bar2', 2);
 INSERT INTO t_foos (id, name, value) VALUES (3, 'bar3', 3);
 INSERT INTO t_foos (id, name, value) VALUES (4, 'bar4', 4);
 INSERT INTO t_foos (id, name, value) VALUES (5, 'bar5', 5);
+
+CREATE TABLE T_WRITE_FOOS (
+	ID BIGINT NOT NULL,
+	NAME VARCHAR(45),
+	VALUE INTEGER
+);
+
+ALTER TABLE T_WRITE_FOOS ADD PRIMARY KEY (ID);