Optimize findRunningJobExecutions to use batch query

Replace N+1 MongoDB queries with a single batch query using
the $in operator in MongoJobExecutionDao.findRunningJobExecutions().

Resolves #5356

Signed-off-by: Wordbe <seonghojin3@gmail.com>

Author: Wordbe

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024-2025 the original author or authors.
+ * Copyright 2024-2026 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.
@@ -23,7 +23,10 @@
 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.job.JobInstance;
@@ -116,18 +119,19 @@ public JobExecution getLastJobExecution(JobInstance jobInstance) {
 	@Override
 	public Set<JobExecution> findRunningJobExecutions(String jobName) {
 		List<JobInstance> jobInstances = this.jobInstanceDao.findJobInstancesByName(jobName);
-		Set<JobExecution> runningJobExecutions = new HashSet<>();
-		for (JobInstance jobInstance : jobInstances) {
-			Query query = query(
-					where("jobInstanceId").is(jobInstance.getId()).and("status").in("STARTING", "STARTED", "STOPPING"));
-			this.mongoOperations
-				.find(query, org.springframework.batch.core.repository.persistence.JobExecution.class,
-						JOB_EXECUTIONS_COLLECTION_NAME)
-				.stream()
-				.map(jobExecution -> convert(jobExecution, jobInstance))
-				.forEach(runningJobExecutions::add);
+		if (jobInstances.isEmpty()) {
+			return Collections.emptySet();
 		}
-		return runningJobExecutions;
+		Map<Long, JobInstance> jobInstanceMap = jobInstances.stream()
+			.collect(Collectors.toMap(JobInstance::getId, Function.identity()));
+		Query query = query(
+				where("jobInstanceId").in(jobInstanceMap.keySet()).and("status").in("STARTING", "STARTED", "STOPPING"));
+		return this.mongoOperations
+			.find(query, org.springframework.batch.core.repository.persistence.JobExecution.class,
+					JOB_EXECUTIONS_COLLECTION_NAME)
+			.stream()
+			.map(jobExecution -> convert(jobExecution, jobInstanceMap.get(jobExecution.getJobInstanceId())))
+			.collect(Collectors.toSet());
 	}
 
 	@Override

spring-batch-core/src/test/java/org/springframework/batch/core/repository/support/MongoJobExecutionDaoIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2025 the original author or authors.
+ * Copyright 2008-2026 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.
@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -241,6 +242,39 @@ void testFindRunningExecutions(@Autowired StepExecutionDao stepExecutionDao) {
 
 	}
 
+	@Test
+	void testFindRunningExecutionsAcrossMultipleInstances(@Autowired JobInstanceDao jobInstanceDao) {
+		String jobName = "multiInstanceJob";
+		JobInstance instance1 = jobInstanceDao.createJobInstance(jobName,
+				new JobParametersBuilder().addLong("id", 1L).toJobParameters());
+		JobInstance instance2 = jobInstanceDao.createJobInstance(jobName,
+				new JobParametersBuilder().addLong("id", 2L).toJobParameters());
+		LocalDateTime now = LocalDateTime.now();
+
+		JobExecution running1 = dao.createJobExecution(instance1, jobParameters);
+		running1.setStatus(BatchStatus.STARTED);
+		running1.setStartTime(now);
+		dao.updateJobExecution(running1);
+
+		JobExecution completed = dao.createJobExecution(instance2, jobParameters);
+		completed.setStatus(BatchStatus.COMPLETED);
+		completed.setStartTime(now);
+		completed.setEndTime(now);
+		dao.updateJobExecution(completed);
+
+		JobExecution running2 = dao.createJobExecution(instance2, jobParameters);
+		running2.setStatus(BatchStatus.STARTED);
+		running2.setStartTime(now);
+		dao.updateJobExecution(running2);
+
+		Set<JobExecution> result = dao.findRunningJobExecutions(jobName);
+
+		assertEquals(2, result.size());
+		Set<Long> ids = result.stream().map(JobExecution::getId).collect(Collectors.toSet());
+		assertTrue(ids.contains(running1.getId()));
+		assertTrue(ids.contains(running2.getId()));
+	}
+
 	/**
 	 * Check the execution is returned
 	 */