RESOLVED - issue BATCH-1420: Late Binding only happens first time when using inner bean definition with collection property

spring-batch-samples/src/main/java/org/springframework/batch/sample/common/ColumnRangePartitioner.java

@@ -0,0 +1,93 @@
+package org.springframework.batch.sample.common;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
+
+/**
+ * Simple minded partitioner for a range of values of a column in a database
+ * table. Works best if the values are uniformly distributed (e.g.
+ * auto-generated primary key values).
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class ColumnRangePartitioner implements Partitioner {
+
+	private SimpleJdbcTemplate jdbcTemplate;
+
+	private String table;
+
+	private String column;
+
+	/**
+	 * The name of the SQL table the data are in.
+	 * 
+	 * @param table the name of the table
+	 */
+	public void setTable(String table) {
+		this.table = table;
+	}
+
+	/**
+	 * The name of the column to partition.
+	 * 
+	 * @param column the column name.
+	 */
+	public void setColumn(String column) {
+		this.column = column;
+	}
+
+	/**
+	 * The data source for connecting to the database.
+	 * 
+	 * @param dataSource a {@link DataSource}
+	 */
+	public void setDataSource(DataSource dataSource) {
+		jdbcTemplate = new SimpleJdbcTemplate(dataSource);
+	}
+
+	/**
+	 * Partition a database table assuming that the data in the column specified
+	 * are uniformly distributed. The execution context values will have keys
+	 * <code>minValue</code> and <code>maxValue</code> specifying the range of
+	 * values to consider in each partition.
+	 * 
+	 * @see Partitioner#partition(int)
+	 */
+	public Map<String, ExecutionContext> partition(int gridSize) {
+
+		int min = jdbcTemplate.queryForInt("SELECT MIN(" + column + ") from " + table);
+		int max = jdbcTemplate.queryForInt("SELECT MAX(" + column + ") from " + table);
+		int targetSize = (max - min) / gridSize + 1;
+
+		Map<String, ExecutionContext> result = new HashMap<String, ExecutionContext>();
+		int number = 0;
+		int start = min;
+		int end = start + targetSize - 1;
+
+		while (start <= max) {
+
+			ExecutionContext value = new ExecutionContext();
+			result.put("partition" + number, value);
+
+			if (end >= max) {
+				end = max;
+			}
+			value.putInt("minValue", start);
+			value.putInt("maxValue", end);
+			start += targetSize;
+			end += targetSize;
+			number++;
+		}
+
+		return result;
+
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/common/OutputFileListener.java

@@ -30,6 +30,12 @@ public class OutputFileListener {
 
 	private String inputKeyName = "fileName";
 
+	private String path = "file:./target/output/";
+	
+	public void setPath(String path) {
+		this.path = path;
+	}
+
 	public void setOutputKeyName(String outputKeyName) {
 		this.outputKeyName = outputKeyName;
 	}
@@ -39,11 +45,15 @@ public class OutputFileListener {
 	}
 
 	@BeforeStep
-	public void CreateOutputNameFromInput(StepExecution stepExecution) {
+	public void createOutputNameFromInput(StepExecution stepExecution) {
 		ExecutionContext executionContext = stepExecution.getExecutionContext();
-		if (executionContext.containsKey(inputKeyName) && !executionContext.containsKey(outputKeyName)) {
-			String inputName = executionContext.getString(inputKeyName);
-			executionContext.putString(outputKeyName, "file:./target/output/" + FilenameUtils.getBaseName(inputName) + ".csv");
+		String inputName = stepExecution.getStepName().replace(":", "-");
+		if (executionContext.containsKey(inputKeyName)) {
+			inputName = executionContext.getString(inputKeyName);
+		}
+		if (!executionContext.containsKey(outputKeyName)) {
+			executionContext.putString(outputKeyName, path + FilenameUtils.getBaseName(inputName)
+					+ ".csv");
 		}
 	}