RESOLVED - BATCH-650: Use FORWARD_ONLY as scroll mode in HibernateCursorItemReader

added item buffer and made the cursor forward-only (+ had to fix samples using stateful items with rollback)

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

@@ -15,6 +15,11 @@
  */
 package org.springframework.batch.item.database;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ListIterator;
+
+import org.hibernate.ScrollMode;
 import org.hibernate.ScrollableResults;
 import org.hibernate.Session;
 import org.hibernate.SessionFactory;
@@ -25,7 +30,6 @@ import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 
 /**
  * {@link ItemReader} for reading database records built on top of Hibernate.
@@ -71,19 +75,46 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
 
 	private boolean saveState = false;
 
+	private boolean shouldReadBuffer = false;
+
+	private ListIterator itemBufferIterator = null;
+
+	private List itemBuffer = new ArrayList();
+	
+	private int lastMarkedBufferIndex = 0;
+
 	public HibernateCursorItemReader() {
-		setName(ClassUtils.getShortName(HibernateCursorItemReader.class));
+		setName(HibernateCursorItemReader.class.getSimpleName());
 	}
 
 	public Object read() {
 
-		if (cursor.next()) {
-			currentProcessedRow++;
-			Object[] data = cursor.get();
-			if (data.length > 1) {
-				return data;
+		currentProcessedRow++;
+
+		if (shouldReadBuffer) {
+			if (itemBufferIterator.hasNext()) {
+				return itemBufferIterator.next();
 			}
-			return data[0];
+			else {
+				// buffer is exhausted, continue reading from file
+				shouldReadBuffer = false;
+				itemBufferIterator = null;
+			}
+		}
+
+		if (cursor.next()) {
+			Object[] data = cursor.get();
+			Object item;
+
+			if (data.length > 1) {
+				item = data;
+			}
+			else {
+				item = data[0];
+			}
+
+			itemBuffer.add(item);
+			return item;
 		}
 		return null;
 	}
@@ -117,17 +148,19 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
 
 		if (useStatelessSession) {
 			statelessSession = sessionFactory.openStatelessSession();
-			cursor = statelessSession.createQuery(queryString).scroll();
+			cursor = statelessSession.createQuery(queryString).scroll(ScrollMode.FORWARD_ONLY);
 		}
 		else {
 			statefulSession = sessionFactory.openSession();
-			cursor = statefulSession.createQuery(queryString).scroll();
+			cursor = statefulSession.createQuery(queryString).scroll(ScrollMode.FORWARD_ONLY);
 		}
 		initialized = true;
 
 		if (executionContext.containsKey(getKey(RESTART_DATA_ROW_NUMBER_KEY))) {
 			currentProcessedRow = Integer.parseInt(executionContext.getString(getKey(RESTART_DATA_ROW_NUMBER_KEY)));
-			cursor.setRowNumber(currentProcessedRow - 1);
+			for (int i = 0; i < currentProcessedRow; i++) {
+				cursor.next();
+			}
 		}
 
 	}
@@ -181,10 +214,20 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
 	 * @see org.springframework.batch.item.ItemReader#mark()
 	 */
 	public void mark() {
-		lastCommitRowNumber = currentProcessedRow;
+
+		if (!shouldReadBuffer) {
+			itemBuffer.clear();
+			itemBufferIterator = null;
+			lastMarkedBufferIndex = 0;
+		}
+		else {
+			lastMarkedBufferIndex = itemBufferIterator.nextIndex();
+		}
+		
 		if (!useStatelessSession) {
 			statefulSession.clear();
 		}
+		lastCommitRowNumber = currentProcessedRow;
 	}
 
 	/*
@@ -194,14 +237,8 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple
 	 */
 	public void reset() {
 		currentProcessedRow = lastCommitRowNumber;
-		if (lastCommitRowNumber == 0) {
-			cursor.beforeFirst();
-		}
-		else {
-			// Set the cursor so that next time it is advanced it will
-			// come back to the committed row.
-			cursor.setRowNumber(lastCommitRowNumber - 1);
-		}
+		shouldReadBuffer = true;
+		itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
 	}
 
 	public void setSaveState(boolean saveState) {

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java

@@ -63,8 +63,15 @@ public abstract class CommonItemReaderTests extends TestCase {
 
 		assertEquals(foo2, tested.read());
 		
-		assertEquals(foo3, tested.read());
+//		TODO handle shortening the commit interval on the fly
+//		
+//		tested.mark();
+//		
+//		assertEquals(foo3, tested.read());
+//		
+//		tested.reset();
 		
+		assertEquals(foo3, tested.read());
 
 		Foo foo4 = (Foo) tested.read();
 		assertEquals(4, foo4.getValue());

spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/CustomerCredit.java

@@ -52,8 +52,12 @@ public class CustomerCredit {
         this.name = name;
     }
     
-    public void increaseCreditBy(BigDecimal sum) {
-    	this.credit = this.credit.add(sum);
+    public CustomerCredit increaseCreditBy(BigDecimal sum) {
+    	CustomerCredit newCredit = new CustomerCredit();
+    	newCredit.credit = this.credit.add(sum);
+    	newCredit.name = this.name;
+    	newCredit.id = this.id;
+    	return newCredit;
     }
     
     public boolean equals(Object o) {

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/BatchSqlCustomerCreditIncreaseWriter.java

@@ -45,8 +45,7 @@ public class BatchSqlCustomerCreditIncreaseWriter implements ItemWriter, Initial
 	 * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object)
 	 */
 	public void write(Object data) throws Exception {
-		CustomerCredit customerCredit = (CustomerCredit) data;
-		customerCredit.increaseCreditBy(FIXED_AMOUNT);
+		CustomerCredit customerCredit = ((CustomerCredit) data).increaseCreditBy(FIXED_AMOUNT);
 		delegate.write(customerCredit);
 	}
 

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseWriter.java

@@ -29,8 +29,7 @@ public class CustomerCreditIncreaseWriter extends AbstractItemWriter {
 	 * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object)
 	 */
 	public void write(Object data) throws Exception {
-		CustomerCredit customerCredit = (CustomerCredit) data;
-		customerCredit.increaseCreditBy(FIXED_AMOUNT);
+		CustomerCredit customerCredit = ((CustomerCredit) data).increaseCreditBy(FIXED_AMOUNT);
 		customerCreditDao.writeCredit(customerCredit);
 	}
 

spring-batch-samples/src/test/java/org/springframework/batch/sample/AbstractCustomerCreditIncreaseTests.java

@@ -108,10 +108,13 @@ public abstract class AbstractCustomerCreditIncreaseTests extends AbstractValida
 
 					public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
 						final BigDecimal creditBeforeUpdate = (BigDecimal) creditsBeforeUpdate.get(rowNum);
+						System.out.print("BEFORE:" + creditBeforeUpdate);
 						final BigDecimal expectedCredit = creditBeforeUpdate.add(CREDIT_INCREASE);
+						System.out.print(" EXPECTED:" + expectedCredit);
 						if (expectedCredit.equals(rs.getBigDecimal(CREDIT_COLUMN))) {
 							matches.add(rs.getBigDecimal(ID_COLUMN));
 						}
+						System.out.println(" ACTUAL: " + rs.getBigDecimal(CREDIT_COLUMN));
 						return null;
 					}
 

spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/CustomerCreditIncreaseProcessorTests.java

@@ -4,50 +4,41 @@ import java.math.BigDecimal;
 
 import junit.framework.TestCase;
 
-import org.easymock.MockControl;
 import org.springframework.batch.sample.dao.CustomerCreditDao;
 import org.springframework.batch.sample.domain.CustomerCredit;
-import org.springframework.batch.sample.item.writer.CustomerCreditIncreaseWriter;
 
 /**
  * Tests for {@link CustomerCreditIncreaseWriter}.
  * 
  * @author Robert Kasanicky
  */
-public class CustomerCreditIncreaseProcessorTests extends TestCase{
+public class CustomerCreditIncreaseProcessorTests extends TestCase {
 
 	private CustomerCreditIncreaseWriter writer = new CustomerCreditIncreaseWriter();
-	
-	private CustomerCreditDao outputSource;
-	private MockControl outputSourceControl = MockControl.createStrictControl(CustomerCreditDao.class);
-	
-	private CustomerCredit customerCredit = new CustomerCredit();
-	
-	
-	protected void setUp() throws Exception {
-		customerCredit.setId(1);
-		customerCredit.setName("testCustomer");
-		
-		outputSource = (CustomerCreditDao) outputSourceControl.getMock();
-		writer.setCustomerCreditDao(outputSource);
-	}
 
 	/**
 	 * Increases customer's credit by fixed value
 	 */
 	public void testProcess() throws Exception {
-		BigDecimal oldCredit = new BigDecimal(10.54);
+		
+		final BigDecimal oldCredit = new BigDecimal(10.54);
+		class CustomerDaoStub implements CustomerCreditDao {
+
+			public void writeCredit(CustomerCredit customerCredit) throws Exception {
+				BigDecimal expectedCredit = oldCredit.add(CustomerCreditIncreaseWriter.FIXED_AMOUNT);
+				assertTrue(customerCredit.getCredit().compareTo(expectedCredit) == 0);
+
+			}
+
+		}
+		CustomerCredit customerCredit = new CustomerCredit();
+		customerCredit.setId(1);
+		customerCredit.setName("testCustomer");
+		writer.setCustomerCreditDao(new CustomerDaoStub());
+
 		customerCredit.setCredit(oldCredit);
-		
-		outputSource.writeCredit(customerCredit);
-		outputSourceControl.setVoidCallable();
-		outputSourceControl.replay();
-		
+
 		writer.write(customerCredit);
-		
-		BigDecimal newCredit = customerCredit.getCredit();
-		BigDecimal expectedCredit = oldCredit.add(CustomerCreditIncreaseWriter.FIXED_AMOUNT);
-		assertTrue(newCredit.compareTo(expectedCredit) == 0);
-		outputSourceControl.verify();
+
 	}
 }