Fix OptimisticLockingFailureTests#testAsyncStopOfStartingJob

This test was failing intermittently due to an incorrect
way of waiting for a job execution to finish, which is:

```
while(jobExecution.isRunning()) {
   // wait for async launched job to complete execution
}
```

`JobExecution#isRunning()` is based on the status
of the job execution in memory which might not be
persisted yet. Here is an excerpt from the Javadoc:

```
Test if this JobExecution indicates that it is
running. It should be noted that this does not
necessarily mean that it has been persisted as
such yet.
```

That's why in the case where `isRunning` returns
false and the JobExecution is not persisted yet
(which is still in a running status in db),
the second attempt of re-running the job fails
with a `JobExecutionAlreadyRunningException`.

This commit fixes the loop by continuously
checking the status of the Job execution
in the job repository until it reaches one
of the end statuses.

Issue #1121

(cherry picked from commit cf26422)

Author: fmbenhassine

spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/OptimisticLockingFailureTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2020 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.
@@ -18,18 +18,22 @@
 
 import static org.junit.Assert.assertTrue;
 
+import java.util.EnumSet;
 import java.util.List;
+import java.util.Set;
 
 import org.junit.Test;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.JobParametersBuilder;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.launch.JobOperator;
+import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.item.ItemWriter;
@@ -39,24 +43,31 @@
 import org.springframework.lang.Nullable;
 
 public class OptimisticLockingFailureTests {
+
+	private static final Set<BatchStatus> END_STATUSES =
+			EnumSet.of(BatchStatus.COMPLETED, BatchStatus.FAILED, BatchStatus.STOPPED);
+
 	@Test
 	public void testAsyncStopOfStartingJob() throws Exception {
 		ApplicationContext applicationContext =
 				new ClassPathXmlApplicationContext("org/springframework/batch/core/repository/dao/OptimisticLockingFailureTests-context.xml");
 		Job job = applicationContext.getBean(Job.class);
 		JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);
 		JobOperator jobOperator = applicationContext.getBean(JobOperator.class);
+		JobRepository jobRepository = applicationContext.getBean(JobRepository.class);
 
-		JobExecution jobExecution = jobLauncher.run(job, new JobParametersBuilder()
+		JobParameters jobParameters = new JobParametersBuilder()
 				.addLong("test", 1L)
-				.toJobParameters());
+				.toJobParameters();
+		JobExecution jobExecution = jobLauncher.run(job, jobParameters);
 
 		Thread.sleep(1000);
 
 		jobOperator.stop(jobExecution.getId());
 
-		while(jobExecution.isRunning()) {
-			// wait for async launched job to complete execution
+		JobExecution lastJobExecution = jobRepository.getLastJobExecution("locking", jobParameters);
+		while (lastJobExecution != null && !END_STATUSES.contains(lastJobExecution.getStatus())) {
+			lastJobExecution = jobRepository.getLastJobExecution("locking", jobParameters);
 		}
 
 		int numStepExecutions = jobExecution.getStepExecutions().size();
@@ -70,14 +81,13 @@ public void testAsyncStopOfStartingJob() throws Exception {
 		assertTrue("Step execution status should be STOPPED but got: " + stepExecutionStatus, stepExecutionStatus.equals(BatchStatus.STOPPED));
 		assertTrue("Job execution status should be STOPPED but got:" + jobExecutionStatus, jobExecutionStatus.equals(BatchStatus.STOPPED));
 
-		JobExecution restartJobExecution = jobLauncher.run(job, new JobParametersBuilder()
-				.addLong("test", 1L)
-				.toJobParameters());
+		JobExecution restartJobExecution = jobLauncher.run(job, jobParameters);
 
 		Thread.sleep(1000);
 
-		while(restartJobExecution.isRunning()) {
-			// wait for async launched job to complete execution
+		lastJobExecution = jobRepository.getLastJobExecution("locking", jobParameters);
+		while (lastJobExecution != null && !END_STATUSES.contains(lastJobExecution.getStatus())) {
+			lastJobExecution = jobRepository.getLastJobExecution("locking", jobParameters);
 		}
 
 		int restartNumStepExecutions = restartJobExecution.getStepExecutions().size();