RESOLVED - BATCH-1147:

*Fixed SimpleChunkProcessor so that the filter count is applied correctly
 *Updated SimpleChunkProcessorTests and CustomerFilterJobFunctionalTests to check the filter count.

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java

@@ -151,9 +151,11 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 			return;
 		}
 
+		int inputSize = inputs.size();
+		
 		Chunk<O> outputs = transform(contribution, inputs);
-
-		contribution.incrementFilterCount(inputs.size() - outputs.size());
+		
+		contribution.incrementFilterCount(inputSize - outputs.size());
 
 		/*
 		 * Need to remember the write skips across transactions, otherwise they

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java

@@ -3,6 +3,7 @@ package org.springframework.batch.core.step.item;
 import static org.junit.Assert.assertEquals;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import org.junit.Before;
@@ -12,12 +13,27 @@ import org.springframework.batch.core.JobInstance;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.item.support.PassThroughItemProcessor;
 
 public class SimpleChunkProcessorTests {
 
-	private SimpleChunkProcessor<String, String> processor;
+	private SimpleChunkProcessor<String, String> processor = new SimpleChunkProcessor<String, String>(
+			new ItemProcessor<String, String>() {
+				public String process(String item) throws Exception {
+					if (item.equals("err")) {
+						return null;
+					}
+					return item;
+				}
+			}, new ItemWriter<String>() {
+				public void write(List<? extends String> items) throws Exception {
+					if (items.contains("fail")) {
+						throw new RuntimeException("Planned failure!");
+					}
+					list.addAll(items);
+				}
+			});
 
 	private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(
 			new JobInstance(123L, new JobParameters(), "job"))));
@@ -26,24 +42,19 @@ public class SimpleChunkProcessorTests {
 
 	@Before
 	public void setUp() {
-		processor = new SimpleChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
-				new ItemWriter<String>() {
-					public void write(List<? extends String> items) throws Exception {
-						if (items.contains("fail")) {
-							throw new RuntimeException("Planned failure!");
-						}
-						list.addAll(items);
-					}
-				});
+		list.clear();
 	}
 
 	@Test
 	public void testProcess() throws Exception {
 		Chunk<String> chunk = new Chunk<String>();
 		chunk.add("foo");
+		chunk.add("err");
 		chunk.add("bar");
 		processor.process(contribution, chunk);
-		assertEquals(2, list.size());
+		assertEquals(Arrays.asList("foo", "bar"), list);
+		assertEquals(1, contribution.getFilterCount());
+		assertEquals(2, contribution.getWriteCount());
 	}
 
 }

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

@@ -22,20 +22,25 @@ import org.springframework.batch.item.file.mapping.FieldSetMapper;
 import org.springframework.batch.item.file.transform.FieldSet;
 
 /**
- * {@link FieldSetMapper} for mapping the 
+ * {@link FieldSetMapper} for mapping to a {@link CustomerUpdate}.
  * 
  * @author Lucas Ward
- *
+ * 
  */
 public class CustomerUpdateFieldSetMapper implements FieldSetMapper<CustomerUpdate> {
 
 	public CustomerUpdate mapFieldSet(FieldSet fs) {
-		
+
+		if (fs == null) {
+			return null;
+		}
+
 		CustomerOperation operation = CustomerOperation.fromCode(fs.readChar(0));
 		String name = fs.readString(1);
 		BigDecimal credit = fs.readBigDecimal(2);
-		
-		return  new CustomerUpdate(operation, name, credit);
+
+		return new CustomerUpdate(operation, name, credit);
+
 	}
-	
+
 }

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

@@ -17,7 +17,6 @@
 package org.springframework.batch.sample;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -32,23 +31,23 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.test.AbstractJobTests;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.core.io.Resource;
 import org.springframework.jdbc.core.RowCallbackHandler;
 import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
-
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration()
-public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLauncherTests {
+public class CustomerFilterJobFunctionalTests extends AbstractJobTests {
 
 	private static final String GET_CUSTOMERS = "select NAME, CREDIT from CUSTOMER order by NAME";
-	
+
 	private List<Customer> customers;
 	private int activeRow = 0;
-	
+
 	private SimpleJdbcTemplate simpleJdbcTemplate;
 	private Map<String, Double> credits = new HashMap<String, Double>();
 
@@ -56,7 +55,7 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 	public void setDataSource(DataSource dataSource) {
 		this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
 	}
-	
+
 	@Before
 	public void onSetUp() throws Exception {
 		simpleJdbcTemplate.update("delete from TRADE");
@@ -66,7 +65,7 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 			credits.put((String) map.get("NAME"), ((Number) map.get("CREDIT")).doubleValue());
 		}
 	}
-	
+
 	@After
 	public void tearDown() throws Exception {
 		simpleJdbcTemplate.update("delete from TRADE");
@@ -74,72 +73,71 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 	}
 
 	@Test
-	public void testLaunchJob() throws Exception{
-		super.testLaunchJob();
-	}
+	public void testFilterJob() throws Exception {
 
-	protected void validatePostConditions() {
-		
-		// assertTrue(((Resource)applicationContext.getBean("customerFileLocator")).exists());
-		
-		customers = Arrays.asList(new Customer("customer1", (credits.get("customer1") )),
-				new Customer("customer2", (credits.get("customer2"))),
-				new Customer("customer3", 100500),
-				new Customer("customer4", credits.get("customer4")),
-				new Customer("customer5", 32345),
-				new Customer("customer6", 123456));
+		JobExecution jobExecution = this.launchJob();
+
+		customers = Arrays.asList(new Customer("customer1", (credits.get("customer1"))), new Customer("customer2",
+				(credits.get("customer2"))), new Customer("customer3", 100500), new Customer("customer4", credits
+				.get("customer4")), new Customer("customer5", 32345), new Customer("customer6", 123456));
 
-		
 		// check content of the customer table
 		activeRow = 0;
 		simpleJdbcTemplate.getJdbcOperations().query(GET_CUSTOMERS, new RowCallbackHandler() {
 
 			public void processRow(ResultSet rs) throws SQLException {
 				Customer customer = customers.get(activeRow++);
-				assertEquals(customer.getName(),rs.getString(1));
+				assertEquals(customer.getName(), rs.getString(1));
 				assertEquals(customer.getCredit(), rs.getDouble(2), .01);
 			}
 		});
-		
-		assertEquals(customers.size(), activeRow);
-		
-		// check content of the output file
+
+		Map<String, Object> step1Execution = this.getStepExecution(jobExecution.getId(), "uploadCustomer");
+		assertEquals(new Long(4), step1Execution.get("READ_COUNT"));
+		assertEquals(new Long(1), step1Execution.get("FILTER_COUNT"));
+		assertEquals(new Long(3), step1Execution.get("WRITE_COUNT"));
+
 	}
 
-	protected void validatePreConditions() {
-		assertTrue(((Resource)applicationContext.getBean("customerFileResource")).exists());
+	private Map<String, Object> getStepExecution(long jobExecutionId, String stepName) {
+		return simpleJdbcTemplate.queryForMap(
+				"SELECT * from BATCH_STEP_EXECUTION where JOB_EXECUTION_ID = ? and STEP_NAME = ?", jobExecutionId,
+				stepName);
 	}
 
 	private static class Customer {
 		private String name;
 		private double credit;
-		
+
 		public Customer(String name, double credit) {
 			this.name = name;
 			this.credit = credit;
 		}
-		
-		public Customer(){
+
+		public Customer() {
 		}
-		
+
 		/**
 		 * @return the credit
 		 */
 		public double getCredit() {
 			return credit;
 		}
+
 		/**
 		 * @param credit the credit to set
 		 */
 		public void setCredit(double credit) {
 			this.credit = credit;
 		}
+
 		/**
 		 * @return the name
 		 */
 		public String getName() {
 			return name;
 		}
+
 		/**
 		 * @param name the name to set
 		 */
@@ -147,7 +145,9 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 			this.name = name;
 		}
 
-		/* (non-Javadoc)
+		/*
+		 * (non-Javadoc)
+		 * 
 		 * @see java.lang.Object#hashCode()
 		 */
 		public int hashCode() {
@@ -160,7 +160,9 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 			return result;
 		}
 
-		/* (non-Javadoc)
+		/*
+		 * (non-Javadoc)
+		 * 
 		 * @see java.lang.Object#equals(java.lang.Object)
 		 */
 		public boolean equals(Object obj) {
@@ -176,13 +178,12 @@ public class CustomerFilterJobFunctionalTests extends AbstractValidatingBatchLau
 			if (name == null) {
 				if (other.name != null)
 					return false;
-			} else if (!name.equals(other.name))
+			}
+			else if (!name.equals(other.name))
 				return false;
 			return true;
 		}
-		
-		
+
 	}
-	
-	
+
 }