Simplify monitoring of job executions

Author: hpoettker

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunner.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-2023 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.
@@ -70,17 +70,17 @@ public class TaskJobLauncherApplicationRunner extends JobLauncherApplicationRunn
 
 	private static final Log logger = LogFactory.getLog(TaskJobLauncherApplicationRunner.class);
 
-	private JobLauncher taskJobLauncher;
+	private final JobLauncher taskJobLauncher;
 
-	private JobExplorer taskJobExplorer;
+	private final JobExplorer taskJobExplorer;
 
-	private JobRepository taskJobRepository;
+	private final JobRepository taskJobRepository;
 
-	private List<JobExecution> jobExecutionList = new ArrayList<>();
+	private final List<JobExecution> jobExecutionList = new ArrayList<>();
 
 	private ApplicationEventPublisher taskApplicationEventPublisher;
 
-	private TaskBatchProperties taskBatchProperties;
+	private final TaskBatchProperties taskBatchProperties;
 
 	/**
 	 * Create a new {@link TaskJobLauncherApplicationRunner}.
@@ -100,6 +100,7 @@ public TaskJobLauncherApplicationRunner(JobLauncher jobLauncher, JobExplorer job
 		this.taskBatchProperties = taskBatchProperties;
 	}
 
+	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
 		super.setApplicationEventPublisher(publisher);
 		this.taskApplicationEventPublisher = publisher;
@@ -112,6 +113,7 @@ public void run(String... args) throws JobExecutionException {
 		monitorJobExecutions();
 	}
 
+	@Override
 	protected void execute(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException,
 			JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException {
 		String jobName = job.getName();
@@ -158,26 +160,31 @@ private void monitorJobExecutions() {
 		template.iterate(context -> {
 
 			List<JobExecution> failedJobExecutions = new ArrayList<>();
-			RepeatStatus repeatStatus = RepeatStatus.FINISHED;
 			for (JobExecution jobExecution : this.jobExecutionList) {
-				JobExecution currentJobExecution = this.taskJobExplorer.getJobExecution(jobExecution.getId());
-				BatchStatus batchStatus = currentJobExecution.getStatus();
+				BatchStatus batchStatus = getCurrentBatchStatus(jobExecution);
 				if (batchStatus.isRunning()) {
-					repeatStatus = RepeatStatus.CONTINUABLE;
+					Thread.sleep(this.taskBatchProperties.getFailOnJobFailurePollInterval());
+					return RepeatStatus.CONTINUABLE;
 				}
 				if (batchStatus.equals(BatchStatus.FAILED)) {
 					failedJobExecutions.add(jobExecution);
 				}
 			}
-			Thread.sleep(this.taskBatchProperties.getFailOnJobFailurePollInterval());
 
-			if (repeatStatus.equals(RepeatStatus.FINISHED) && failedJobExecutions.size() > 0) {
+			if (failedJobExecutions.size() > 0) {
 				throwJobFailedException(failedJobExecutions);
 			}
-			return repeatStatus;
+			return RepeatStatus.FINISHED;
 		});
 	}
 
+	private BatchStatus getCurrentBatchStatus(JobExecution jobExecution) {
+		if (jobExecution.getStatus().isRunning()) {
+			return this.taskJobExplorer.getJobExecution(jobExecution.getId()).getStatus();
+		}
+		return jobExecution.getStatus();
+	}
+
 	private void throwJobFailedException(List<JobExecution> failedJobExecutions) {
 		StringBuilder message = new StringBuilder("The following Jobs have failed: \n");
 		for (JobExecution failedJobExecution : failedJobExecutions) {

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunnerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -21,20 +21,16 @@
 
 import javax.sql.DataSource;
 
-import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.function.Executable;
 
 import org.springframework.batch.core.Job;
-import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.job.builder.JobBuilder;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.builder.StepBuilder;
-import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
@@ -165,17 +161,12 @@ private void validateContext() {
 		assertThat(taskExplorer.getTaskExecution(jobExecutionIds.iterator().next()).getExecutionId()).isEqualTo(1);
 	}
 
-	private void validateForFail(String errorMessage, Class clazz, String[] enabledArgs) {
+	private void validateForFail(String errorMessage, Class<?> clazz, String[] enabledArgs) {
 		Executable executable = () -> this.applicationContext = SpringApplication
 				.run(new Class[] { clazz, PropertyPlaceholderAutoConfiguration.class }, enabledArgs);
 
-		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(executable::execute)
-				.has(new Condition<Throwable>() {
-					@Override
-					public boolean matches(Throwable value) {
-						return errorMessage.equals(value.getCause().getMessage());
-					}
-				});
+		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(executable::execute).havingCause()
+				.withMessage(errorMessage);
 	}
 
 	@TaskBatchTest
@@ -186,12 +177,9 @@ public static class JobConfiguration {
 		@Bean
 		public Job job(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
 			return new JobBuilder("job", jobRepository)
-					.start(new StepBuilder("step1", jobRepository).tasklet(new Tasklet() {
-						@Override
-						public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
-							System.out.println("Executed");
-							return RepeatStatus.FINISHED;
-						}
+					.start(new StepBuilder("step1", jobRepository).tasklet((contribution, chunkContext) -> {
+						System.out.println("Executed");
+						return RepeatStatus.FINISHED;
 					}, transactionManager).build()).build();
 		}
 
@@ -240,27 +228,20 @@ public static class JobWithFailureConfiguration {
 
 		@Bean
 		public Job jobFail() {
-			return new JobBuilder("jobA").repository(this.jobRepository)
-					.start(new StepBuilder("step1").repository(this.jobRepository).tasklet(new Tasklet() {
-						@Override
-						public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext)
-								throws Exception {
-							System.out.println("Executed");
-							throw new IllegalStateException("WHOOPS");
-						}
-					}).transactionManager(transactionManager).build()).build();
+			return new JobBuilder("jobA", this.jobRepository)
+					.start(new StepBuilder("step1", this.jobRepository).tasklet((contribution, chunkContext) -> {
+						System.out.println("Executed");
+						throw new IllegalStateException("WHOOPS");
+					}, transactionManager).build()).build();
 		}
 
 		@Bean
 		public Job jobFun() {
-			return new JobBuilder("jobSucceed").repository(this.jobRepository)
-					.start(new StepBuilder("step1Succeed").repository(this.jobRepository).tasklet(new Tasklet() {
-						@Override
-						public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
-							System.out.println("Executed");
-							return RepeatStatus.FINISHED;
-						}
-					}).transactionManager(transactionManager).build()).build();
+			return new JobBuilder("jobSucceed", this.jobRepository)
+					.start(new StepBuilder("step1Succeed", this.jobRepository).tasklet((contribution, chunkContext) -> {
+						System.out.println("Executed");
+						return RepeatStatus.FINISHED;
+					}, transactionManager).build()).build();
 		}
 
 	}