Fix retrieval of null job parameters of type long/double

Before this commit, a null job parameter of type long
or double was retrieved as 0L or 0.0. This caused
JobOperator.restart to create a new job instance
instead of restarting the previous failed execution.

This commit fixes how null job parameters of type
long or double are retrieved from the database to
correctly restart the same job instance.

Issue #4087

Author: linhongcheng

spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/repository/MySQLJdbcJobRepositoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -110,6 +110,36 @@ public void testDateMillisecondPrecision() throws Exception {
 		Assert.assertEquals(2, jobExecutions.size());
 	}
 
+	/*
+	 * This test is for issue https://github.com/spring-projects/spring-batch/issues/4087:
+	 * A round trip from a `java.lang.Long` or `java.lang.Double` JobParameter that value is null to the database and back
+	 * again should preserve null. otherwise a different
+	 * job instance is created while the existing one should be used.
+	 *
+	 * This test ensures that round trip to the database with a `java.lang.Long` or `java.lang.Double`
+	 * parameter that value is null ends up with a single job instance (with two job executions)
+	 * being created and not two distinct job instances (with a job execution for
+	 * each one).
+	 *
+	 */
+	@Test
+	public void testLongOrDoubleNullable() throws Exception {
+		// given
+		JobParameters jobParameters = new JobParametersBuilder()
+				.addLong("attribute", null) // as same as Double type
+				.toJobParameters();
+
+		// when
+		JobExecution jobExecution = this.jobLauncher.run(this.job, jobParameters);
+		this.jobOperator.restart(jobExecution.getId()); // should load the null value for java.lang.Long or java.lang.Double
+
+		// then
+		List<Long> jobInstances = this.jobOperator.getJobInstances("job", 0, 100);
+		Assert.assertEquals(1, jobInstances.size());
+		List<Long> jobExecutions = this.jobOperator.getExecutions(jobInstances.get(0));
+		Assert.assertEquals(2, jobExecutions.size());
+	}
+
 	@Configuration
 	@EnableBatchProcessing
 	static class TestConfiguration {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2019 the original author or authors.
+ * Copyright 2006-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -374,9 +374,11 @@ public void processRow(ResultSet rs) throws SQLException {
 				if (type == ParameterType.STRING) {
 					value = new JobParameter(rs.getString(4), rs.getString(8).equalsIgnoreCase("Y"));
 				} else if (type == ParameterType.LONG) {
-					value = new JobParameter(rs.getLong(6), rs.getString(8).equalsIgnoreCase("Y"));
+					long longValue = rs.getLong(6);
+					value = new JobParameter(rs.wasNull() ? null : longValue, rs.getString(8).equalsIgnoreCase("Y"));
 				} else if (type == ParameterType.DOUBLE) {
-					value = new JobParameter(rs.getDouble(7), rs.getString(8).equalsIgnoreCase("Y"));
+					double doubleValue = rs.getDouble(7);
+					value = new JobParameter(rs.wasNull() ? null : doubleValue, rs.getString(8).equalsIgnoreCase("Y"));
 				} else if (type == ParameterType.DATE) {
 					value = new JobParameter(rs.getTimestamp(5), rs.getString(8).equalsIgnoreCase("Y"));
 				}