From ae808ede21e55adb2a5c402a79a68da86c749a84 Mon Sep 17 00:00:00 2001
From: dhgarrette <dhgarrette@2e8e966d-6415-0410-94b9-dc4e65435d03>
Date: Mon, 9 Mar 2009 06:08:39 +0000
Subject: [PATCH] BATCH-1128: Updated FaultTolerantChunkProcessor's recovery
 procedure so that StepExecution.writeCount is not updated during throttling
 because writes will be rolled back.

---
 .../item/FaultTolerantChunkProcessor.java     |  3 +--
 .../sample/SkipSampleFunctionalTests.java     | 24 +++++++++++++++----
 2 files changed, 21 insertions(+), 6 deletions(-)

diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
index 6068fca74..aa25fb965 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
@@ -157,7 +157,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 					try {
 						writeItems(Collections.singletonList(item));
-						contribution.incrementWriteCount(1);
 					}
 					catch (Exception e) {
 						checkSkipPolicy(inputIterator, outputIterator, e, contribution);
@@ -174,7 +173,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 				}
 
 				doAfterWrite(outputs.getItems());
-				
+				contribution.incrementWriteCount(outputs.size());				
 				return null;
 
 			}
diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java
index cb6249462..5a1a48654 100644
--- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java
+++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java
@@ -134,7 +134,7 @@ public class SkipSampleFunctionalTests {
 		Map<String, Object> execution1 = this.getJobExecution(id1);
 		assertEquals("COMPLETED", execution1.get("STATUS"));
 
-		this.validateLaunchWithSkips();
+		this.validateLaunchWithSkips(id1);
 
 		//
 		// Clear the data
@@ -148,7 +148,7 @@ public class SkipSampleFunctionalTests {
 		Map<String, Object> execution2 = this.getJobExecution(id2);
 		assertEquals("COMPLETED", execution2.get("STATUS"));
 
-		this.validateLaunchWithoutSkips();
+		this.validateLaunchWithoutSkips(id2);
 
 		//
 		// Make sure that the launches were separate executions and separate
@@ -158,7 +158,7 @@ public class SkipSampleFunctionalTests {
 		assertTrue(!execution1.get("JOB_INSTANCE_ID").equals(execution2.get("JOB_INSTANCE_ID")));
 	}
 
-	private void validateLaunchWithSkips() {
+	private void validateLaunchWithSkips(long jobExecutionId) {
 		// Step1: 9 input records, 1 skipped in process, 1 skipped in write =>
 		// 7 written to output
 		assertEquals(7, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "TRADE"));
@@ -175,9 +175,14 @@ public class SkipSampleFunctionalTests {
 		}
 
 		assertEquals(new BigDecimal("340.45"), tradeWriter.getTotalPrice());
+
+		Map<String, Object> step1Execution = this.getStepExecution(jobExecutionId, "step1");
+		assertEquals(new Long(3), step1Execution.get("COMMIT_COUNT"));
+		assertEquals(new Long(8), step1Execution.get("READ_COUNT"));
+		assertEquals(new Long(7), step1Execution.get("WRITE_COUNT"));
 	}
 
-	private void validateLaunchWithoutSkips() {
+	private void validateLaunchWithoutSkips(long jobExecutionId) {
 		// Step1: 5 input records => 5 written to output
 		assertEquals(5, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "TRADE"));
 
@@ -195,6 +200,17 @@ public class SkipSampleFunctionalTests {
 				jobExecutionId);
 	}
 
+	private Map<String, Object> getStepExecution(long jobExecutionId, String stepName) {
+		for (Map<String, Object> rs : simpleJdbcTemplate.queryForList(
+				"SELECT * from BATCH_STEP_EXECUTION where JOB_EXECUTION_ID = ? and STEP_NAME = ?", jobExecutionId,
+				stepName)) {
+			System.err.println(rs);
+		}
+		return simpleJdbcTemplate.queryForMap(
+				"SELECT * from BATCH_STEP_EXECUTION where JOB_EXECUTION_ID = ? and STEP_NAME = ?", jobExecutionId,
+				stepName);
+	}
+
 	/**
 	 * Launch the entire job, including all steps, in order.
 	 *