diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcExecutionContextDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcExecutionContextDao.java
index 918c4fbc8..e105d2cf8 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcExecutionContextDao.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcExecutionContextDao.java
@@ -27,11 +27,11 @@ import java.sql.SQLException;
 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.JobExecution;
 import org.springframework.batch.core.StepExecution;
@@ -58,6 +58,7 @@ import org.springframework.util.Assert;
  * @author Michael Minella
  * @author David Turanski
  * @author Mahmoud Ben Hassine
+ * @author Yanming Zhou
  */
 public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implements ExecutionContextDao {
 
@@ -153,13 +154,9 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
 		Long executionId = jobExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_JOB_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId);
-		if (!results.isEmpty()) {
-			return results.get(0);
-		}
-		else {
-			return new ExecutionContext();
+		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_JOB_EXECUTION_CONTEXT),
+				new ExecutionContextRowMapper(), executionId)) {
+			return stream.findFirst().orElseGet(ExecutionContext::new);
 		}
 	}
 
@@ -168,13 +165,9 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
 		Long executionId = stepExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		List<ExecutionContext> results = getJdbcTemplate().query(getQuery(FIND_STEP_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId);
-		if (results.size() > 0) {
-			return results.get(0);
-		}
-		else {
-			return new ExecutionContext();
+		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_STEP_EXECUTION_CONTEXT),
+				new ExecutionContextRowMapper(), executionId)) {
+			return stream.findFirst().orElseGet(ExecutionContext::new);
 		}
 	}
 
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobExecutionDao.java
index d4350b29a..f13977299 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobExecutionDao.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobExecutionDao.java
@@ -28,6 +28,7 @@ import java.util.Map;
 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;
@@ -336,16 +337,9 @@ public class JdbcJobExecutionDao extends AbstractJdbcBatchMetadataDao implements
 
 		Long id = jobInstance.getId();
 
-		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");
-
-		if (executions.isEmpty()) {
-			return null;
-		}
-		else {
-			return executions.get(0);
+		try (Stream<JobExecution> stream = getJdbcTemplate().queryForStream(getQuery(GET_LAST_EXECUTION),
+				new JobExecutionRowMapper(jobInstance), id, id)) {
+			return stream.findFirst().orElse(null);
 		}
 	}
 
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobInstanceDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobInstanceDao.java
index 4726fbaab..9c8f3ba31 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobInstanceDao.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcJobInstanceDao.java
@@ -21,6 +21,7 @@ import java.sql.SQLException;
 import java.sql.Types;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.springframework.batch.core.DefaultJobKeyGenerator;
 import org.springframework.batch.core.JobExecution;
@@ -173,21 +174,12 @@ public class JdbcJobInstanceDao extends AbstractJdbcBatchMetadataDao implements
 
 		RowMapper<JobInstance> rowMapper = new JobInstanceRowMapper();
 
-		List<JobInstance> instances;
-		if (StringUtils.hasLength(jobKey)) {
-			instances = getJdbcTemplate().query(getQuery(FIND_JOBS_WITH_KEY), rowMapper, jobName, jobKey);
-		}
-		else {
-			instances = getJdbcTemplate().query(getQuery(FIND_JOBS_WITH_EMPTY_KEY), rowMapper, jobName, jobKey);
+		try (Stream<JobInstance> stream = getJdbcTemplate().queryForStream(
+				getQuery(StringUtils.hasLength(jobKey) ? FIND_JOBS_WITH_KEY : FIND_JOBS_WITH_EMPTY_KEY), rowMapper,
+				jobName, jobKey)) {
+			return stream.findFirst().orElse(null);
 		}
 
-		if (instances.isEmpty()) {
-			return null;
-		}
-		else {
-			Assert.state(instances.size() == 1, "instance count must be 1 but was " + instances.size());
-			return instances.get(0);
-		}
 	}
 
 	@Override
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcStepExecutionDao.java
index 3bc69d1bd..d26f99ed8 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcStepExecutionDao.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcStepExecutionDao.java
@@ -29,6 +29,7 @@ import java.util.Iterator;
 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;
@@ -326,16 +327,9 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
 	@Override
 	@Nullable
 	public StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId) {
-		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");
-		if (executions.isEmpty()) {
-			return null;
-		}
-		else {
-			return executions.get(0);
+		try (Stream<StepExecution> stream = getJdbcTemplate().queryForStream(getQuery(GET_STEP_EXECUTION),
+				new StepExecutionRowMapper(jobExecution), jobExecution.getId(), stepExecutionId)) {
+			return stream.findFirst().orElse(null);
 		}
 	}