Persist step execution status before calling listeners

Before this commit, the step execution status was not persisted
in the job repository before calling step execution listeners.
Therefore, listeners see the step's status as STARTED and not
COMPLETED or FAILED. This was not coherent with the contract of
`afterStep` which states that the method is called after the
execution of the step's processing logic (whether successful or
failed), where the status (both in memory and in the DB) should
not be STARTED at that point, but rather a non-running status.

This commit adds a job repository update to persist the step's
status before calling step execution listeners.

Resolves #4362

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java

@@ -286,37 +286,46 @@ public final void execute(StepExecution stepExecution)
 				logger.info("Step: [" + stepExecution.getStepName() + "] executed in "
 						+ BatchMetrics.formatDuration(stepExecutionDuration));
 			}
-			try {
-				// Update the step execution to the latest known value so the
-				// listeners can act on it
-				exitStatus = exitStatus.and(stepExecution.getExitStatus());
-				stepExecution.setExitStatus(exitStatus);
-				exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
-			}
-			catch (Exception e) {
-				logger.error(String.format("Exception in afterStep callback in step %s in job %s", name,
-						stepExecution.getJobExecution().getJobInstance().getJobName()), e);
-			}
 
+			// Update the step execution to the latest known value so the
+			// listeners can act on it
+			exitStatus = exitStatus.and(stepExecution.getExitStatus());
+			stepExecution.setExitStatus(exitStatus);
+
+			// save status in job repository before calling listeners
+			// https://github.com/spring-projects/spring-batch/issues/4362
 			try {
+				getJobRepository().update(stepExecution);
 				getJobRepository().updateExecutionContext(stepExecution);
 			}
 			catch (Exception e) {
 				stepExecution.setStatus(BatchStatus.UNKNOWN);
-				exitStatus = exitStatus.and(ExitStatus.UNKNOWN);
+				stepExecution.setExitStatus(exitStatus.and(ExitStatus.UNKNOWN));
 				stepExecution.addFailureException(e);
 				logger.error(String.format(
 						"Encountered an error saving batch meta data for step %s in job %s. "
 								+ "This job is now in an unknown state and should not be restarted.",
 						name, stepExecution.getJobExecution().getJobInstance().getJobName()), e);
 			}
+
+			try {
+				exitStatus = exitStatus.and(getCompositeListener().afterStep(stepExecution));
+			}
+			catch (Exception e) {
+				logger.error(String.format("Exception in afterStep callback in step %s in job %s", name,
+						stepExecution.getJobExecution().getJobInstance().getJobName()), e);
+			}
+
 			stepExecutionEvent.exitStatus = stepExecution.getExitStatus().getExitCode();
 			stepExecutionEvent.commit();
 			stopObservation(stepExecution, observation);
 			stepExecution.setExitStatus(exitStatus);
 
+			// save status in job repository after calling listeners (since afterStep
+			// might have changed it)
 			try {
 				getJobRepository().update(stepExecution);
+				getJobRepository().updateExecutionContext(stepExecution);
 			}
 			catch (Exception e) {
 				stepExecution.setStatus(BatchStatus.UNKNOWN);

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java

@@ -128,7 +128,7 @@ public void open(ExecutionContext executionContext) throws ItemStreamException {
 		taskletStep.execute(stepExecution);
 		assertEquals(FAILED, stepExecution.getStatus());
 		assertTrue(stepExecution.getFailureExceptions().contains(exception));
-		assertEquals(2, jobRepository.getUpdateCount());
+		assertEquals(3, jobRepository.getUpdateCount());
 	}
 
 	@Test
@@ -144,7 +144,7 @@ public void beforeStep(StepExecution stepExecution) {
 		taskletStep.execute(stepExecution);
 		assertEquals(FAILED, stepExecution.getStatus());
 		assertTrue(stepExecution.getFailureExceptions().contains(exception));
-		assertEquals(2, jobRepository.getUpdateCount());
+		assertEquals(3, jobRepository.getUpdateCount());
 	}
 
 	@Test
@@ -170,7 +170,7 @@ public RepeatStatus execute(StepContribution contribution, ChunkContext attribut
 		taskletStep.execute(stepExecution);
 		assertEquals(COMPLETED, stepExecution.getStatus());
 		assertFalse(stepExecution.getFailureExceptions().contains(exception));
-		assertEquals(3, jobRepository.getUpdateCount());
+		assertEquals(4, jobRepository.getUpdateCount());
 	}
 
 	/*
@@ -191,7 +191,7 @@ public ExitStatus afterStep(StepExecution stepExecution) {
 		assertEquals(FAILED, stepExecution.getStatus());
 		assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
 		assertFalse(stepExecution.getFailureExceptions().contains(exception));
-		assertEquals(2, jobRepository.getUpdateCount());
+		assertEquals(3, jobRepository.getUpdateCount());
 	}
 
 	@Test
@@ -210,7 +210,7 @@ public void close() throws ItemStreamException {
 		assertEquals(FAILED, stepExecution.getStatus());
 		assertEquals(stepExecution.getFailureExceptions().get(0), taskletException);
 		assertEquals(stepExecution.getFailureExceptions().get(1).getSuppressed()[0], exception);
-		assertEquals(2, jobRepository.getUpdateCount());
+		assertEquals(3, jobRepository.getUpdateCount());
 	}
 
 	@Test