RESOLVED - issue BATCH-1422: HibernateCursorItemReader causes OutOfMemoryError when skipping large sets of data

RESOLVED - BATCH-1486: HiberbatePagingItemReader

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java

@@ -17,8 +17,6 @@ package org.springframework.batch.item.database;
 
 import java.util.Map;
 
-import org.hibernate.Query;
-import org.hibernate.ScrollMode;
 import org.hibernate.ScrollableResults;
 import org.hibernate.Session;
 import org.hibernate.SessionFactory;
@@ -27,12 +25,10 @@ import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
-import org.springframework.batch.item.database.support.AbstractHibernateQueryProvider;
 import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * {@link ItemReader} for reading database records built on top of Hibernate. It
@@ -59,124 +55,70 @@ import org.springframework.util.StringUtils;
 public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements ItemStream,
 		InitializingBean {
 
-	private SessionFactory sessionFactory;
-
-	private StatelessSession statelessSession;
-
-	private Session statefulSession;
-
-	private ScrollableResults cursor;
-
-	private String queryString = "";
-
-	private String queryName = "";
-
-	private Map<String, Object> parameterValues;
-
-	private HibernateQueryProvider queryProvider;
-
-	private boolean useStatelessSession = true;
-
-	private boolean initialized = false;
-
-	private int fetchSize = 0;
+	private HibernateItemReaderHelper<T> helper = new HibernateItemReaderHelper<T>();
 
 	public HibernateCursorItemReader() {
 		setName(ClassUtils.getShortName(HibernateCursorItemReader.class));
 	}
-	
+
+	private ScrollableResults cursor;
+
+	private boolean initialized = false;
+
+	public void afterPropertiesSet() throws Exception {
+		helper.afterPropertiesSet();
+	}
+
 	/**
 	 * The parameter values to apply to a query (map of name:value).
 	 * 
 	 * @param parameterValues the parameter values to set
 	 */
 	public void setParameterValues(Map<String, Object> parameterValues) {
-		this.parameterValues = parameterValues;
-	}
-
-	/**
-	 * Open appropriate type of hibernate session and create the query.
-	 */
-	private Query createQuery() {
-
-		if (useStatelessSession) {
-			statelessSession = sessionFactory.openStatelessSession();
-			if (queryProvider != null) {
-				queryProvider.setStatelessSession(statelessSession);
-			}
-			else {
-				if (StringUtils.hasText(queryName)) {
-					return statelessSession.getNamedQuery(queryName);
-				}
-				else {
-					return statelessSession.createQuery(queryString);
-				}
-			}
-		}
-		else {
-			statefulSession = sessionFactory.openSession();
-			if (queryProvider != null) {
-				queryProvider.setSession(statefulSession);
-			}
-			else {
-				if (StringUtils.hasText(queryName)) {
-					return statefulSession.getNamedQuery(queryName);
-				}
-				else {
-					return statefulSession.createQuery(queryString);
-				}
-			}
-		}
-
-		// if queryProvider is set use it to create a query
-		return queryProvider.createQuery();
-
-	}
-
-	/**
-	 * @param sessionFactory hibernate session factory
-	 */
-	public void setSessionFactory(SessionFactory sessionFactory) {
-		this.sessionFactory = sessionFactory;
-	}
-
-	public void afterPropertiesSet() throws Exception {
-
-		Assert.state(sessionFactory!=null, "A SessionFactory must be provided");
-		Assert.state(fetchSize >= 0, "fetchSize must not be negative");
-
-		if (queryProvider == null) {
-			Assert.notNull(sessionFactory, "session factory must be set");
-			Assert.state(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName),
-					"queryString or queryName must be set");
-		}
-		// making sure that the appropriate (Hibernate) query provider is set
-		else {
-			Assert.state(queryProvider instanceof AbstractHibernateQueryProvider,
-					"Hibernate query provider must be set");
-		}
-
+		helper.setParameterValues(parameterValues);
 	}
 
 	/**
+	 * A query name for an externalized query. Either this or the {
+	 * {@link #setQueryString(String) query string} or the {
+	 * {@link #setQueryProvider(HibernateQueryProvider) query provider} should
+	 * be set.
+	 * 
 	 * @param queryName name of a hibernate named query
 	 */
 	public void setQueryName(String queryName) {
-		this.queryName = queryName;
-	}
-
-	/**
-	 * @param queryString HQL query string
-	 */
-	public void setQueryString(String queryString) {
-		this.queryString = queryString;
+		helper.setQueryName(queryName);
 	}
 
 	/**
+	 * A query provider. Either this or the {{@link #setQueryString(String)
+	 * query string} or the {{@link #setQueryName(String) query name} should be
+	 * set.
+	 * 
 	 * @param queryProvider Hibernate query provider
 	 */
 	public void setQueryProvider(HibernateQueryProvider queryProvider) {
-		this.queryProvider = queryProvider;
+		helper.setQueryProvider(queryProvider);
+	}
+
+	/**
+	 * A query string in HQL. Either this or the {
+	 * {@link #setQueryProvider(HibernateQueryProvider) query provider} or the {
+	 * {@link #setQueryName(String) query name} should be set.
+	 * 
+	 * @param queryString HQL query string
+	 */
+	public void setQueryString(String queryString) {
+		helper.setQueryString(queryString);
+	}
+
+	/**
+	 * The Hibernate SessionFactory to use the create a session.
+	 * 
+	 * @param sessionFactory the {@link SessionFactory} to set
+	 */
+	public void setSessionFactory(SessionFactory sessionFactory) {
+		helper.setSessionFactory(sessionFactory);
 	}
 
 	/**
@@ -187,32 +129,7 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 	 * {@link Session}
 	 */
 	public void setUseStatelessSession(boolean useStatelessSession) {
-		Assert.state(!initialized);
-		this.useStatelessSession = useStatelessSession;
-	}
-
-	/**
-	 * Clears the session if not stateful and delegates to super class.
-	 */
-	@Override
-	public void update(ExecutionContext executionContext) throws ItemStreamException {
-		super.update(executionContext);
-		if (!useStatelessSession) {
-			statefulSession.clear();
-		}
-	}
-
-	/**
-	 * Gives the JDBC driver a hint as to the number of rows that should be
-	 * fetched from the database when more rows are needed for this
-	 * <code>ResultSet</code> object. If the fetch size specified is zero, the
-	 * JDBC driver ignores the value.
-	 * 
-	 * @param fetchSize the number of rows to fetch, 0 by default
-	 * @see Query#setFetchSize(int)
-	 */
-	public void setFetchSize(int fetchSize) {
-		this.fetchSize = fetchSize;
+		helper.setUseStatelessSession(useStatelessSession);
 	}
 
 	protected T doRead() throws Exception {
@@ -241,48 +158,53 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 	}
 
 	/**
-	 * Open hibernate session and create a forward-only cursor for the
-	 * query.
+	 * Open hibernate session and create a forward-only cursor for the query.
 	 */
 	protected void doOpen() throws Exception {
 		Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
-
-		Query query = createQuery();
-		if (parameterValues!=null) {
-			query.setProperties(parameterValues);
-		}
-		cursor = query.setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
-
+		cursor = helper.getForwardOnlyCursor();
 		initialized = true;
 	}
+	
+	/**
+	 * Update the context and clear the session if stateful.
+	 * 
+	 * @param executionContext the current {@link ExecutionContext}
+	 * @throws ItemStreamException if there is a problem
+	 */
+	@Override
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		super.update(executionContext);
+		helper.clear();
+	}
 
+	/**
+	 * Wind forward through the result set to the item requested. Also clears
+	 * the session every now and then (if stateful) to avoid memory problems.
+	 * The frequency of session clearing is the larger of the fetch size (if
+	 * set) and 100.
+	 * 
+	 * @param itemIndex the first item to read
+	 * @throws Exception if there is a problem
+	 * @see AbstractItemCountingItemStreamItemReader#jumpToItem(int)
+	 */
 	@Override
 	protected void jumpToItem(int itemIndex) throws Exception {
-		for (int i = 0; i < itemIndex; i++) {
-			cursor.next();
-		}
+		helper.jumpToItem(cursor, itemIndex);
 	}
 
 	/**
 	 * Close the cursor and hibernate session.
 	 */
 	protected void doClose() throws Exception {
+
 		initialized = false;
 
 		if (cursor != null) {
 			cursor.close();
 		}
 
-		if (useStatelessSession) {
-			if (statelessSession != null) {
-				statelessSession.close();
-			}
-		}
-		else {
-			if (statefulSession != null) {
-				statefulSession.close();
-			}
-		}
+		helper.close();
 
 	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java

@@ -0,0 +1,233 @@
+/*
+ * Copyright 2006-2010 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.item.database;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.hibernate.Query;
+import org.hibernate.ScrollMode;
+import org.hibernate.ScrollableResults;
+import org.hibernate.Session;
+import org.hibernate.SessionFactory;
+import org.hibernate.StatelessSession;
+import org.springframework.batch.item.database.support.AbstractHibernateQueryProvider;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Internal shared state helper for hibernate readers.
+ * 
+ * @author Dave Syer
+ * 
+ */
+class HibernateItemReaderHelper<T> {
+
+	private SessionFactory sessionFactory;
+
+	private String queryString = "";
+
+	private String queryName = "";
+
+	private Map<String, Object> parameterValues;
+
+	private HibernateQueryProvider queryProvider;
+
+	private boolean useStatelessSession = true;
+
+	private int fetchSize = 0;
+
+	private StatelessSession statelessSession;
+
+	private Session statefulSession;
+
+	/**
+	 * @param queryName name of a hibernate named query
+	 */
+	public void setQueryName(String queryName) {
+		this.queryName = queryName;
+	}
+
+	/**
+	 * @param queryString HQL query string
+	 */
+	public void setQueryString(String queryString) {
+		this.queryString = queryString;
+	}
+
+	/**
+	 * @param queryProvider Hibernate query provider
+	 */
+	public void setQueryProvider(HibernateQueryProvider queryProvider) {
+		this.queryProvider = queryProvider;
+	}
+
+	/**
+	 * Can be set only in uninitialized state.
+	 * 
+	 * @param useStatelessSession <code>true</code> to use
+	 * {@link StatelessSession} <code>false</code> to use standard hibernate
+	 * {@link Session}
+	 */
+	public void setUseStatelessSession(boolean useStatelessSession) {
+		Assert.state(statefulSession == null && statelessSession == null,
+				"The useStatelessSession flag can only be set before a session is initialized.");
+		this.useStatelessSession = useStatelessSession;
+	}
+
+	/**
+	 * The parameter values to apply to a query (map of name:value).
+	 * 
+	 * @param parameterValues the parameter values to set
+	 */
+	public void setParameterValues(Map<String, Object> parameterValues) {
+		this.parameterValues = parameterValues;
+	}
+
+	/**
+	 * @param sessionFactory hibernate session factory
+	 */
+	public void setSessionFactory(SessionFactory sessionFactory) {
+		this.sessionFactory = sessionFactory;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+
+		Assert.state(sessionFactory != null, "A SessionFactory must be provided");
+		Assert.state(fetchSize >= 0, "fetchSize must not be negative");
+
+		if (queryProvider == null) {
+			Assert.notNull(sessionFactory, "session factory must be set");
+			Assert.state(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName),
+					"queryString or queryName must be set");
+		}
+		// making sure that the appropriate (Hibernate) query provider is set
+		else {
+			Assert.state(queryProvider instanceof AbstractHibernateQueryProvider,
+					"Hibernate query provider must be set");
+		}
+
+	}
+
+	/**
+	 * Get a cursor over all of the results, with the forward-only flag set.
+	 * 
+	 * @return a forward-only {@link ScrollableResults}
+	 */
+	public ScrollableResults getForwardOnlyCursor() {
+		Query query = createQuery();
+		if (parameterValues != null) {
+			query.setProperties(parameterValues);
+		}
+		return query.setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
+	}
+
+	/**
+	 * Open appropriate type of hibernate session and create the query.
+	 */
+	private Query createQuery() {
+
+		if (useStatelessSession) {
+			statelessSession = sessionFactory.openStatelessSession();
+			if (queryProvider != null) {
+				queryProvider.setStatelessSession(statelessSession);
+			}
+			else {
+				if (StringUtils.hasText(queryName)) {
+					return statelessSession.getNamedQuery(queryName);
+				}
+				else {
+					return statelessSession.createQuery(queryString);
+				}
+			}
+		}
+		else {
+			statefulSession = sessionFactory.openSession();
+			if (queryProvider != null) {
+				queryProvider.setSession(statefulSession);
+			}
+			else {
+				if (StringUtils.hasText(queryName)) {
+					return statefulSession.getNamedQuery(queryName);
+				}
+				else {
+					return statefulSession.createQuery(queryString);
+				}
+			}
+		}
+
+		// if queryProvider is set use it to create a query
+		return queryProvider.createQuery();
+
+	}
+
+	/**
+	 * Scroll through the results up to the item specified.
+	 * 
+	 * @param cursor the results to scroll over
+	 */
+	public void jumpToItem(ScrollableResults cursor, int itemIndex) {
+		int flushSize = Math.max(fetchSize, 100);
+		for (int i = 0; i < itemIndex; i++) {
+			cursor.next();
+			if (i % flushSize == 0 && !useStatelessSession) {
+				statefulSession.clear(); // Clears in-memory cache
+			}
+		}
+	}
+
+	/**
+	 * Close the open session (stateful or otherwise).
+	 */
+	public void close() {
+		if (statelessSession != null) {
+			statelessSession.close();
+		}
+		if (statefulSession != null) {
+			statefulSession.close();
+		}
+	}
+
+	/**
+	 * @param page the page to read
+	 * @param pageSize the size of the page
+	 * @return a collection of items
+	 */
+	public Collection<? extends T> readPage(int page, int pageSize) {
+
+		clear();
+
+		Query query = createQuery();
+		if (parameterValues != null) {
+			query.setProperties(parameterValues);
+		}
+		@SuppressWarnings("unchecked")
+		List<T> result = query.setFetchSize(fetchSize).setFirstResult(page * pageSize).setMaxResults(pageSize).list();
+		return result;
+
+	}
+
+	/**
+	 * Clear the session if stateful.
+	 */
+	public void clear() {
+		if (statefulSession!=null) {
+			statefulSession.clear();
+		}
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java

@@ -0,0 +1,163 @@
+/*
+ * 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.item.database;
+
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.hibernate.Session;
+import org.hibernate.SessionFactory;
+import org.hibernate.StatelessSession;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.ClassUtils;
+
+/**
+ * {@link ItemReader} for reading database records built on top of Hibernate. It
+ * executes the HQL query when initialized iterates over the result set as
+ * {@link #read()} method is called, returning an object corresponding to
+ * current row. The query can be set directly using
+ * {@link #setQueryString(String)} or a named query can be used by
+ * {@link #setQueryName(String)}.
+ * 
+ * <p>
+ * The reader can be configured to use either {@link StatelessSession}
+ * sufficient for simple mappings without the need to cascade to associated
+ * objects or standard hibernate {@link Session} for more advanced mappings or
+ * when caching is desired. When stateful session is used it will be cleared in
+ * the {@link #update(ExecutionContext)} method without being flushed (no data
+ * modifications are expected).
+ * </p>
+ * 
+ * <p>
+ * The implementation is thread-safe in between calls to
+ * {@link #open(ExecutionContext)}, but remember to use
+ * <code>saveState=false</code> if used in a multi-threaded client (no restart
+ * available).
+ * </p>
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.1
+ */
+public class HibernatePagingItemReader<T> extends AbstractPagingItemReader<T> implements ItemStream, InitializingBean {
+
+	private HibernateItemReaderHelper<T> helper = new HibernateItemReaderHelper<T>();
+
+	public HibernatePagingItemReader() {
+		setName(ClassUtils.getShortName(HibernatePagingItemReader.class));
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		super.afterPropertiesSet();
+		helper.afterPropertiesSet();
+	}
+
+	/**
+	 * The parameter values to apply to a query (map of name:value).
+	 * 
+	 * @param parameterValues the parameter values to set
+	 */
+	public void setParameterValues(Map<String, Object> parameterValues) {
+		helper.setParameterValues(parameterValues);
+	}
+
+	/**
+	 * A query name for an externalized query. Either this or the {
+	 * {@link #setQueryString(String) query string} or the {
+	 * {@link #setQueryProvider(HibernateQueryProvider) query provider} should
+	 * be set.
+	 * 
+	 * @param queryName name of a hibernate named query
+	 */
+	public void setQueryName(String queryName) {
+		helper.setQueryName(queryName);
+	}
+
+	/**
+	 * A query provider. Either this or the {{@link #setQueryString(String)
+	 * query string} or the {{@link #setQueryName(String) query name} should be
+	 * set.
+	 * 
+	 * @param queryProvider Hibernate query provider
+	 */
+	public void setQueryProvider(HibernateQueryProvider queryProvider) {
+		helper.setQueryProvider(queryProvider);
+	}
+
+	/**
+	 * A query string in HQL. Either this or the {
+	 * {@link #setQueryProvider(HibernateQueryProvider) query provider} or the {
+	 * {@link #setQueryName(String) query name} should be set.
+	 * 
+	 * @param queryString HQL query string
+	 */
+	public void setQueryString(String queryString) {
+		helper.setQueryString(queryString);
+	}
+
+	/**
+	 * The Hibernate SessionFactory to use the create a session.
+	 * 
+	 * @param sessionFactory the {@link SessionFactory} to set
+	 */
+	public void setSessionFactory(SessionFactory sessionFactory) {
+		helper.setSessionFactory(sessionFactory);
+	}
+
+	/**
+	 * Can be set only in uninitialized state.
+	 * 
+	 * @param useStatelessSession <code>true</code> to use
+	 * {@link StatelessSession} <code>false</code> to use standard hibernate
+	 * {@link Session}
+	 */
+	public void setUseStatelessSession(boolean useStatelessSession) {
+		helper.setUseStatelessSession(useStatelessSession);
+	}
+
+	@Override
+	protected void doOpen() throws Exception {
+		super.doOpen();
+	}
+
+	@Override
+	protected void doReadPage() {
+
+		if (results == null) {
+			results = new CopyOnWriteArrayList<T>();
+		}
+		else {
+			results.clear();
+		}
+		
+		results.addAll(helper.readPage(getPage(), getPageSize()));
+
+	}
+
+	@Override
+	protected void doJumpToPage(int itemIndex) {
+	}
+
+	@Override
+	protected void doClose() throws Exception {
+		helper.close();
+		super.doClose();
+	}
+
+}