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

@@ -27,11 +27,11 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.stream.Stream;
 
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.step.StepExecution;
@@ -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/JdbcJobExecutionDao.java

@@ -28,7 +28,6 @@
 import java.util.Set;
 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;
@@ -79,6 +78,7 @@
  * @author Philippe Marschall
  * @author Jinwoo Bae
  * @author Yanming Zhou
+ * @author Hyunsang Han
  */
 public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements JobExecutionDao, InitializingBean {
 
@@ -337,10 +337,12 @@ public JobExecution getLastJobExecution(JobInstance jobInstance) {
 
 		Long id = jobInstance.getId();
 
-		try (Stream<JobExecution> stream = getJdbcTemplate().queryForStream(getQuery(GET_LAST_EXECUTION),
-				new JobExecutionRowMapper(jobInstance), id, id)) {
-			return stream.findFirst().orElse(null);
-		}
+		List<JobExecution> executions = getJdbcTemplate().query(getQuery(GET_LAST_EXECUTION),
+				new JobExecutionRowMapper(jobInstance), id, id);
+
+		Assert.state(executions.size() <= 1, "There must be at most one latest job execution");
+
+		return executions.isEmpty() ? null : executions.get(0);
 	}
 
 	@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 {
 
@@ -176,10 +176,16 @@ 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()) {
+			return null;
+		}
+		else {
+			Assert.state(instances.size() == 1, "instance count must be 1 but was " + instances.size());
+			return instances.get(0);
 		}
 
 	}

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

@@ -29,7 +29,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;
@@ -70,6 +69,7 @@
  * @author Baris Cubukcuoglu
  * @author Minsoo Kim
  * @author Yanming Zhou
+ * @author Hyunsang Han
  * @see StepExecutionDao
  */
 public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implements StepExecutionDao, InitializingBean {
@@ -327,10 +327,12 @@ private String truncateExitDescription(String description) {
 	@Override
 	@Nullable
 	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