Replace queryForStream with query to prevent connection leaks

Signed-off-by: Hyunsang Han <[email protected]>

Author: HyunSangHan

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

@@ -31,7 +31,7 @@
 import java.util.Map.Entry;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.stream.Stream;
+import java.util.List;
 
 import org.springframework.batch.core.job.JobExecution;
 
@@ -59,6 +59,7 @@
  * @author David Turanski
  * @author Mahmoud Ben Hassine
  * @author Yanming Zhou
+ * @author Hyunsang Han
  */
 public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implements ExecutionContextDao {
 
@@ -154,21 +155,19 @@ public ExecutionContext getExecutionContext(JobExecution jobExecution) {
 		Long executionId = jobExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_JOB_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId)) {
-			return stream.findFirst().orElseGet(ExecutionContext::new);
-		}
+		List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_JOB_EXECUTION_CONTEXT),
+				new ExecutionContextRowMapper(), executionId);
+		return !results.isEmpty() ? results.get(0) : new ExecutionContext();
 	}
 
 	@Override
 	public ExecutionContext getExecutionContext(StepExecution stepExecution) {
 		Long executionId = stepExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_STEP_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId)) {
-			return stream.findFirst().orElseGet(ExecutionContext::new);
-		}
+		List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT),
+				new ExecutionContextRowMapper(), executionId);
+		return !results.isEmpty() ? results.get(0) : new ExecutionContext();
 	}
 
 	@Override

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

@@ -21,7 +21,6 @@
 import java.sql.Types;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Stream;
 
 import org.springframework.batch.core.job.DefaultJobKeyGenerator;
 import org.springframework.batch.core.job.JobExecution;
@@ -58,6 +57,7 @@
  * @author Mahmoud Ben Hassine
  * @author Parikshit Dutta
  * @author Yanming Zhou
+ * @author Hyunsang Han
  */
 public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements JobInstanceDao, InitializingBean {
 
@@ -186,11 +186,14 @@ public JobInstance getJobInstance(final String jobName, final JobParameters jobP
 
 		RowMapper<JobInstance> rowMapper = new JobInstanceRowMapper();
 
-		try (Stream<JobInstance> stream = getJdbcTemplate().queryForStream(
+		List<JobInstance> instances = getJdbcTemplate().query(
 				getQuery(StringUtils.hasLength(jobKey) ? FIND_JOBS_WITH_KEY : FIND_JOBS_WITH_EMPTY_KEY), rowMapper,
-				jobName, jobKey)) {
-			return stream.findFirst().orElse(null);
+				jobName, jobKey);
+
+		if (!instances.isEmpty()) {
+			Assert.state(instances.size() == 1, "instance count must be 1 but was " + instances.size());
 		}
+		return instances.isEmpty() ? null : instances.get(0);
 
 	}
 

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

@@ -28,7 +28,6 @@
 import java.util.List;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.stream.Stream;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -68,6 +67,7 @@
  * @author Baris Cubukcuoglu
  * @author Minsoo Kim
  * @author Yanming Zhou
+ * @author Hyunsang Han
  * @see StepExecutionDao
  */
 public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implements StepExecutionDao, InitializingBean {
@@ -306,10 +306,12 @@ private long getJobExecutionId(long stepExecutionId) {
 	@Nullable
 	@Deprecated(since = "6.0", forRemoval = true)
 	public StepExecution getStepExecution(JobExecution jobExecution, long stepExecutionId) {
-		try (Stream<StepExecution> stream = getJdbcTemplate().queryForStream(getQuery(GET_STEP_EXECUTION),
-				new StepExecutionRowMapper(jobExecution), stepExecutionId)) {
-			return stream.findFirst().orElse(null);
-		}
+		List<StepExecution> executions = getJdbcTemplate().query(getQuery(GET_STEP_EXECUTION),
+				new StepExecutionRowMapper(jobExecution), stepExecutionId);
+
+		Assert.state(executions.size() <= 1,
+				"There can be at most one step execution with given name for single job execution");
+		return executions.isEmpty() ? null : executions.get(0);
 	}
 
 	@Override