Use batch inserts for job parameters

Use batch inserts for job parameters to reduce
the number of database round trips.

Issue #3867

Author: marschall

spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcJobExecutionDao.java

@@ -16,15 +16,16 @@
 
 package org.springframework.batch.core.repository.dao;
 
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.sql.Types;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Date;
 import java.util.Map;
-import java.util.Map.Entry;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
@@ -70,6 +71,7 @@
  * @author Michael Minella
  * @author Mahmoud Ben Hassine
  * @author Dimitrios Liapis
+ * @author Philippe Marschall
  */
 public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements JobExecutionDao, InitializingBean {
 
@@ -376,27 +378,34 @@ public void deleteJobExecutionParameters(JobExecution jobExecution) {
 	 */
 	private void insertJobParameters(Long executionId, JobParameters jobParameters) {
 
-		for (Entry<String, JobParameter<?>> entry : jobParameters.getParameters().entrySet()) {
-			JobParameter jobParameter = entry.getValue();
-			insertParameter(executionId, jobParameter.getType(), entry.getKey(), jobParameter.getValue(),
-					jobParameter.isIdentifying());
+		if (jobParameters.isEmpty()) {
+			return;
 		}
+
+		getJdbcTemplate().batchUpdate(getQuery(CREATE_JOB_PARAMETERS), jobParameters.getParameters().entrySet(), 100,
+				(ps, entry) -> {
+					JobParameter jobParameter = entry.getValue();
+					insertParameter(ps, executionId, jobParameter.getType(), entry.getKey(), jobParameter.getValue(),
+							jobParameter.isIdentifying());
+				});
 	}
 
 	/**
 	 * Convenience method that inserts an individual records into the JobParameters table.
+	 * @throws SQLException if the driver throws an exception
 	 */
-	private <T> void insertParameter(Long executionId, Class<T> type, String key, T value, boolean identifying) {
-
-		Object[] args = new Object[0];
-		int[] argTypes = new int[] { Types.BIGINT, Types.VARCHAR, Types.VARCHAR, Types.VARCHAR, Types.CHAR };
+	private <T> void insertParameter(PreparedStatement preparedStatement, Long executionId, Class<T> type, String key,
+			T value, boolean identifying) throws SQLException {
 
 		String identifyingFlag = identifying ? "Y" : "N";
 
 		String stringValue = this.conversionService.convert(value, String.class);
-		args = new Object[] { executionId, key, type.getName(), stringValue, identifyingFlag };
 
-		getJdbcTemplate().update(getQuery(CREATE_JOB_PARAMETERS), args, argTypes);
+		preparedStatement.setLong(1, executionId);
+		preparedStatement.setString(2, key);
+		preparedStatement.setString(3, type.getName());
+		preparedStatement.setString(4, stringValue);
+		preparedStatement.setString(5, identifyingFlag);
 	}
 
 	/**