Update Task to BOOT 2.1.M1
Migrating to use ApplicationContextRunner or ImportAutoConfiguration with SpringApp.run, instead of SpringApplicationBuilder, because builder does not handle AutoConfiguration properly SimpleTaskAutoConfiguration now has an annotation AutoConfigureBefore the BatchTaskAutoConfig so that it is processed prior. THis is so that that BatchTaskAutoConfig can create the appropriate beans SimpleTaskAutoConfiguration has new annotations so that it is AutoConfigured after BindingServiceConfiguration and after SimpleTaskAutoConfiguration. This is so that it does not attempt to start emitting messages before stream is ready and it can create the appropriate beans after SimpleTaskAutoConfiguration has run. Renamed SimpleTaskConfiguration to SimpleTaskAutoConfiguration. Task version updated to 2.1.0 Added missing headers Updated documentation. Deprecated EnableTask Added ability to disable Task autoconfiguration. Removed @EnableTask from tests Resolves #439 Resolves #440 Resolves #448 Resolves #466
This commit is contained in:
Glenn Renfro
committed by
Michael Minella
Michael Minella
parent
d2c90c5256
commit
d2bc2530cc
@@ -37,12 +37,27 @@ public class TaskBatchProperties {
|
||||
private String jobNames = "";
|
||||
|
||||
/**
|
||||
* The order for the {@coce CommandLineRunner} used to run batch jobs when
|
||||
* The order for the {@code CommandLineRunner} used to run batch jobs when
|
||||
* {@code spring.cloud.task.batch.fail-on-job-failure=true}. Defaults to 0 (same as the
|
||||
* {@link org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner}).
|
||||
*/
|
||||
private int commandLineRunnerOrder = 0;
|
||||
|
||||
/**
|
||||
* Maximum wait time in milliseconds that Spring Cloud Task will wait for tasks to complete
|
||||
* when spring.cloud.task.batch.failOnJobFailure is set to true. Defaults
|
||||
* to 0. 0 indicates no wait time is enforced.
|
||||
*/
|
||||
private long failOnJobFailurewaitTime = 0;
|
||||
|
||||
/**
|
||||
* Fixed delay in milliseconds that Spring Cloud Task will wait when checking if
|
||||
* {@link org.springframework.batch.core.JobExecution}s have completed,
|
||||
* when spring.cloud.task.batch.failOnJobFailure is set to true. Defaults
|
||||
* to 5000.
|
||||
*/
|
||||
private long failOnJobFailurePollInterval = 5000l;
|
||||
|
||||
public String getJobNames() {
|
||||
return this.jobNames;
|
||||
}
|
||||
@@ -58,4 +73,20 @@ public class TaskBatchProperties {
|
||||
public void setCommandLineRunnerOrder(int commandLineRunnerOrder) {
|
||||
this.commandLineRunnerOrder = commandLineRunnerOrder;
|
||||
}
|
||||
|
||||
public long getFailOnJobFailurewaitTime() {
|
||||
return failOnJobFailurewaitTime;
|
||||
}
|
||||
|
||||
public void setFailOnJobFailurewaitTime(long failOnJobFailurewaitTime) {
|
||||
this.failOnJobFailurewaitTime = failOnJobFailurewaitTime;
|
||||
}
|
||||
|
||||
public long getFailOnJobFailurePollInterval() {
|
||||
return failOnJobFailurePollInterval;
|
||||
}
|
||||
|
||||
public void setFailOnJobFailurePollInterval(long failOnJobFailurePollInterval) {
|
||||
this.failOnJobFailurePollInterval = failOnJobFailurePollInterval;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,11 +18,19 @@ package org.springframework.cloud.task.batch.configuration;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.configuration.JobRegistry;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
|
||||
import org.springframework.batch.core.repository.support.SimpleJobRepository;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
|
||||
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
@@ -37,22 +45,31 @@ import org.springframework.context.annotation.Configuration;
|
||||
@Configuration
|
||||
@ConditionalOnProperty(name = "spring.cloud.task.batch.fail-on-job-failure", havingValue = "true", matchIfMissing = false)
|
||||
@EnableConfigurationProperties(TaskBatchProperties.class)
|
||||
@AutoConfigureBefore(BatchAutoConfiguration.class)
|
||||
public class TaskJobLauncherAutoConfiguration {
|
||||
|
||||
@Autowired
|
||||
private TaskBatchProperties properties;
|
||||
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(JobRepository.class)
|
||||
public JobRepository jobRepository(DataSource dataSource) throws Exception{
|
||||
JobRepositoryFactoryBean factoryBean = new JobRepositoryFactoryBean();
|
||||
factoryBean.setDataSource(dataSource);
|
||||
return factoryBean.getObject();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TaskJobLauncherCommandLineRunnerFactoryBean jobLauncherCommandLineRunner(JobLauncher jobLauncher,
|
||||
JobExplorer jobExplorer, List<Job> jobs, JobRegistry jobRegistry) {
|
||||
JobExplorer jobExplorer, List<Job> jobs, JobRegistry jobRegistry, JobRepository jobRepository) {
|
||||
TaskJobLauncherCommandLineRunnerFactoryBean taskJobLauncherCommandLineRunnerFactoryBean =
|
||||
new TaskJobLauncherCommandLineRunnerFactoryBean(jobLauncher,
|
||||
jobExplorer,
|
||||
jobs,
|
||||
this.properties.getJobNames(),
|
||||
jobRegistry);
|
||||
|
||||
taskJobLauncherCommandLineRunnerFactoryBean.setOrder(this.properties.getCommandLineRunnerOrder());
|
||||
this.properties,
|
||||
jobRegistry,
|
||||
jobRepository);
|
||||
|
||||
return taskJobLauncherCommandLineRunnerFactoryBean;
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.configuration.JobRegistry;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.beans.factory.FactoryBean;
|
||||
import org.springframework.cloud.task.batch.handler.TaskJobLauncherCommandLineRunner;
|
||||
import org.springframework.util.Assert;
|
||||
@@ -46,15 +47,23 @@ public class TaskJobLauncherCommandLineRunnerFactoryBean implements FactoryBean<
|
||||
|
||||
private Integer order = 0;
|
||||
|
||||
private TaskBatchProperties taskBatchProperties;
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
public TaskJobLauncherCommandLineRunnerFactoryBean(JobLauncher jobLauncher,
|
||||
JobExplorer jobExplorer, List<Job> jobs, String jobNames,
|
||||
JobRegistry jobRegistry) {
|
||||
JobExplorer jobExplorer, List<Job> jobs, TaskBatchProperties taskBatchProperties,
|
||||
JobRegistry jobRegistry, JobRepository jobRepository) {
|
||||
Assert.notNull(taskBatchProperties, "properties must not be null");
|
||||
this.jobLauncher = jobLauncher;
|
||||
this.jobExplorer = jobExplorer;
|
||||
Assert.notEmpty(jobs, "jobs must not be null nor empty");
|
||||
this.jobs = jobs;
|
||||
this.jobNames = jobNames;
|
||||
this.jobNames = taskBatchProperties.getJobNames();
|
||||
this.jobRegistry = jobRegistry;
|
||||
this.taskBatchProperties = taskBatchProperties;
|
||||
this.order = taskBatchProperties.getCommandLineRunnerOrder();
|
||||
this.jobRepository = jobRepository;
|
||||
}
|
||||
|
||||
public void setOrder(int order) {
|
||||
@@ -62,9 +71,9 @@ public class TaskJobLauncherCommandLineRunnerFactoryBean implements FactoryBean<
|
||||
}
|
||||
|
||||
@Override
|
||||
public TaskJobLauncherCommandLineRunner getObject() throws Exception {
|
||||
public TaskJobLauncherCommandLineRunner getObject() {
|
||||
TaskJobLauncherCommandLineRunner taskJobLauncherCommandLineRunner =
|
||||
new TaskJobLauncherCommandLineRunner(this.jobLauncher, this.jobExplorer);
|
||||
new TaskJobLauncherCommandLineRunner(this.jobLauncher, this.jobExplorer, this.jobRepository, this.taskBatchProperties);
|
||||
taskJobLauncherCommandLineRunner.setJobs(this.jobs);
|
||||
if(StringUtils.hasText(this.jobNames)) {
|
||||
taskJobLauncherCommandLineRunner.setJobNames(this.jobNames);
|
||||
@@ -74,7 +83,6 @@ public class TaskJobLauncherCommandLineRunnerFactoryBean implements FactoryBean<
|
||||
if(this.order != null) {
|
||||
taskJobLauncherCommandLineRunner.setOrder(this.order);
|
||||
}
|
||||
|
||||
return taskJobLauncherCommandLineRunner;
|
||||
}
|
||||
|
||||
@@ -82,4 +90,5 @@ public class TaskJobLauncherCommandLineRunnerFactoryBean implements FactoryBean<
|
||||
public Class<?> getObjectType() {
|
||||
return TaskJobLauncherCommandLineRunner.class;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -16,177 +16,218 @@
|
||||
|
||||
package org.springframework.cloud.task.batch.handler;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.Properties;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.batch.core.ExitStatus;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.JobExecutionException;
|
||||
import org.springframework.batch.core.JobParameter;
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
import org.springframework.batch.core.JobParametersBuilder;
|
||||
import org.springframework.batch.core.JobParametersIncrementer;
|
||||
import org.springframework.batch.core.JobParametersInvalidException;
|
||||
import org.springframework.batch.core.configuration.JobRegistry;
|
||||
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
|
||||
import org.springframework.batch.core.converter.JobParametersConverter;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.core.launch.JobParametersNotFoundException;
|
||||
import org.springframework.batch.core.launch.NoSuchJobException;
|
||||
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
|
||||
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.JobRestartException;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.batch.repeat.RepeatCallback;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
import org.springframework.batch.repeat.support.RepeatTemplate;
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.boot.autoconfigure.batch.JobExecutionEvent;
|
||||
import org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner;
|
||||
import org.springframework.cloud.task.batch.configuration.TaskBatchProperties;
|
||||
import org.springframework.cloud.task.listener.TaskException;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.context.ApplicationEventPublisherAware;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.util.PatternMatchUtils;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* {@link CommandLineRunner} to {@link JobLauncher launch} Spring Batch jobs. Runs all
|
||||
* jobs in the surrounding context by default and throw an exception upon the
|
||||
* first job that returns an {@link ExitStatus} of FAILED.
|
||||
* Can also be used to launch a specific job by providing a jobName. The
|
||||
* TaskJobLaunchercommandLineRunner takes the place of the
|
||||
* {@link org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner}
|
||||
* when it is in use.
|
||||
* jobs in the surrounding context by default and throws an exception upon the first job
|
||||
* that returns an {@link BatchStatus} of FAILED if a {@link TaskExecutor} in the
|
||||
* {@link JobLauncher} is not specified. If a {@link TaskExecutor} is specified
|
||||
* in the {@link JobLauncher} then all Jobs are launched and an
|
||||
* exception is thrown if one or more of the jobs has an {@link BatchStatus} of FAILED.
|
||||
* TaskJobLauncherCommandLineRunner can also be used to launch a specific job by
|
||||
* providing a jobName. The TaskJobLaunchercommandLineRunner takes the place of the
|
||||
* {@link org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner} when
|
||||
* it is in use.
|
||||
*
|
||||
* @author Glenn Renfro
|
||||
* @since 2.0.0
|
||||
*/
|
||||
public class TaskJobLauncherCommandLineRunner implements CommandLineRunner, Ordered, ApplicationEventPublisherAware{
|
||||
/**
|
||||
* The default order for the command line runner.
|
||||
*/
|
||||
public static final int DEFAULT_ORDER = 0;
|
||||
public class TaskJobLauncherCommandLineRunner extends JobLauncherCommandLineRunner {
|
||||
|
||||
private JobLauncher taskJobLauncher;
|
||||
|
||||
private JobExplorer taskJobExplorer;
|
||||
|
||||
private JobRepository taskJobRepository;
|
||||
|
||||
private static final Log logger = LogFactory
|
||||
.getLog(TaskJobLauncherCommandLineRunner.class);
|
||||
|
||||
private JobParametersConverter converter = new DefaultJobParametersConverter();
|
||||
private List<JobExecution> jobExecutionList = new ArrayList<>();
|
||||
|
||||
private JobLauncher jobLauncher;
|
||||
private ApplicationEventPublisher taskApplicationEventPublisher;
|
||||
|
||||
private JobRegistry jobRegistry;
|
||||
private TaskBatchProperties taskBatchProperties;
|
||||
|
||||
private JobExplorer jobExplorer;
|
||||
|
||||
private String jobNames;
|
||||
|
||||
private Collection<Job> jobs = Collections.emptySet();
|
||||
|
||||
private int order = DEFAULT_ORDER;
|
||||
|
||||
private ApplicationEventPublisher publisher;
|
||||
|
||||
public TaskJobLauncherCommandLineRunner(JobLauncher jobLauncher,
|
||||
JobExplorer jobExplorer) {
|
||||
this.jobLauncher = jobLauncher;
|
||||
this.jobExplorer = jobExplorer;
|
||||
/**
|
||||
* Create a new {@link TaskJobLauncherCommandLineRunner}.
|
||||
* @param jobLauncher to launch jobs
|
||||
* @param jobExplorer to check the job repository for previous executions
|
||||
* @param jobRepository to check if a job instance exists with the given parameters
|
||||
* when running a job
|
||||
* @param taskBatchProperties the properties used to configure the taskBatchProperties.
|
||||
*/
|
||||
public TaskJobLauncherCommandLineRunner(JobLauncher jobLauncher, JobExplorer jobExplorer,
|
||||
JobRepository jobRepository, TaskBatchProperties taskBatchProperties) {
|
||||
super(jobLauncher, jobExplorer, jobRepository);
|
||||
this.taskJobLauncher = jobLauncher;
|
||||
this.taskJobExplorer = jobExplorer;
|
||||
this.taskJobRepository = jobRepository;
|
||||
this.taskBatchProperties = taskBatchProperties;
|
||||
}
|
||||
|
||||
public void setOrder(int order) {
|
||||
this.order = order;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getOrder() {
|
||||
return this.order;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
|
||||
this.publisher = publisher;
|
||||
}
|
||||
|
||||
public void setJobRegistry(JobRegistry jobRegistry) {
|
||||
this.jobRegistry = jobRegistry;
|
||||
}
|
||||
|
||||
public void setJobNames(String jobNames) {
|
||||
this.jobNames = jobNames;
|
||||
}
|
||||
|
||||
public void setJobParametersConverter(JobParametersConverter converter) {
|
||||
this.converter = converter;
|
||||
}
|
||||
|
||||
public void setJobs(Collection<Job> jobs) {
|
||||
this.jobs = jobs;
|
||||
super.setApplicationEventPublisher(publisher);
|
||||
this.taskApplicationEventPublisher = publisher;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws JobExecutionException {
|
||||
logger.info("Running default command line with: " + Arrays.asList(args));
|
||||
launchJobFromProperties(StringUtils.splitArrayElementsIntoProperties(args, "="));
|
||||
}
|
||||
|
||||
protected void launchJobFromProperties(Properties properties)
|
||||
throws JobExecutionException {
|
||||
JobParameters jobParameters = this.converter.getJobParameters(properties);
|
||||
executeLocalJobs(jobParameters);
|
||||
executeRegisteredJobs(jobParameters);
|
||||
}
|
||||
|
||||
private void executeRegisteredJobs(JobParameters jobParameters)
|
||||
throws JobExecutionException {
|
||||
if (this.jobRegistry != null && StringUtils.hasText(this.jobNames)) {
|
||||
String[] jobsToRun = this.jobNames.split(",");
|
||||
for (String jobName : jobsToRun) {
|
||||
try {
|
||||
Job job = this.jobRegistry.getJob(jobName);
|
||||
if (this.jobs.contains(job)) {
|
||||
continue;
|
||||
}
|
||||
execute(job, jobParameters);
|
||||
}
|
||||
catch (NoSuchJobException ex) {
|
||||
logger.debug("No job found in registry for job name: " + jobName);
|
||||
}
|
||||
}
|
||||
}
|
||||
validateJobExecutions();
|
||||
}
|
||||
|
||||
protected void execute(Job job, JobParameters jobParameters)
|
||||
throws JobExecutionAlreadyRunningException, JobRestartException,
|
||||
JobInstanceAlreadyCompleteException, JobParametersInvalidException,
|
||||
JobParametersNotFoundException {
|
||||
JobParameters nextParameters = new JobParametersBuilder(jobParameters,
|
||||
this.jobExplorer).getNextJobParameters(job).toJobParameters();
|
||||
JobExecution execution = this.jobLauncher.run(job, nextParameters);
|
||||
if (this.publisher != null) {
|
||||
this.publisher.publishEvent(new JobExecutionEvent(execution));
|
||||
JobInstanceAlreadyCompleteException, JobParametersInvalidException {
|
||||
String jobName = job.getName();
|
||||
JobParameters parameters = jobParameters;
|
||||
boolean jobInstanceExists = this.taskJobRepository.isJobInstanceExists(jobName,
|
||||
parameters);
|
||||
if (jobInstanceExists) {
|
||||
JobExecution lastJobExecution = this.taskJobRepository
|
||||
.getLastJobExecution(jobName, jobParameters);
|
||||
if (lastJobExecution != null && isStoppedOrFailed(lastJobExecution)
|
||||
&& job.isRestartable()) {
|
||||
// Retry a failed or stopped execution with previous parameters
|
||||
JobParameters previousParameters = lastJobExecution.getJobParameters();
|
||||
/*
|
||||
* remove Non-identifying parameters from the previous execution's
|
||||
* parameters since there is no way to remove them programmatically. If
|
||||
* they are required (or need to be modified) on a restart, they need to
|
||||
* be (re)specified.
|
||||
*/
|
||||
JobParameters previousIdentifyingParameters = removeNonIdentifying(
|
||||
previousParameters);
|
||||
// merge additional parameters with previous ones (overriding those with
|
||||
// the same key)
|
||||
parameters = merge(previousIdentifyingParameters, jobParameters);
|
||||
}
|
||||
}
|
||||
if(execution.getExitStatus().getExitCode().equals(ExitStatus.FAILED.getExitCode())) {
|
||||
String message = String.format("Job %s failed during " +
|
||||
"execution for jobId %s with jobExecutionId of %s",
|
||||
execution.getJobInstance().getJobName(),
|
||||
execution.getJobId(), execution.getId());
|
||||
logger.error(message);
|
||||
throw new TaskException(message);
|
||||
else {
|
||||
JobParametersIncrementer incrementer = job.getJobParametersIncrementer();
|
||||
if (incrementer != null) {
|
||||
JobParameters nextParameters = new JobParametersBuilder(jobParameters,
|
||||
this.taskJobExplorer).getNextJobParameters(job).toJobParameters();
|
||||
parameters = merge(nextParameters, jobParameters);
|
||||
}
|
||||
}
|
||||
JobExecution execution = this.taskJobLauncher.run(job, parameters);
|
||||
if (this.taskApplicationEventPublisher != null) {
|
||||
this.taskApplicationEventPublisher.publishEvent(new JobExecutionEvent(execution));
|
||||
}
|
||||
this.jobExecutionList.add(execution);
|
||||
if (execution.getStatus().equals(BatchStatus.FAILED)) {
|
||||
throwJobFailedException(Collections.singletonList(execution));
|
||||
}
|
||||
}
|
||||
|
||||
private void executeLocalJobs(JobParameters jobParameters)
|
||||
throws JobExecutionException {
|
||||
for (Job job : this.jobs) {
|
||||
if (StringUtils.hasText(this.jobNames)) {
|
||||
String[] jobsToRun = this.jobNames.split(",");
|
||||
if (!PatternMatchUtils.simpleMatch(jobsToRun, job.getName())) {
|
||||
logger.debug("Skipped job: " + job.getName());
|
||||
continue;
|
||||
private void validateJobExecutions() {
|
||||
RepeatTemplate template = new RepeatTemplate();
|
||||
|
||||
Date startDate = new Date();
|
||||
|
||||
template.iterate(new RepeatCallback() {
|
||||
|
||||
public RepeatStatus doInIteration(RepeatContext context) throws InterruptedException {
|
||||
List<JobExecution> failedJobExecutions = new ArrayList<>();
|
||||
RepeatStatus repeatStatus = RepeatStatus.FINISHED;
|
||||
for (JobExecution jobExecution : jobExecutionList) {
|
||||
JobExecution currentJobExecution = taskJobExplorer.getJobExecution(jobExecution.getId());
|
||||
BatchStatus batchStatus = currentJobExecution.getStatus();
|
||||
if (batchStatus.isRunning()) {
|
||||
repeatStatus = RepeatStatus.CONTINUABLE;
|
||||
}
|
||||
if (batchStatus.equals(BatchStatus.FAILED)) {
|
||||
failedJobExecutions.add(jobExecution);
|
||||
}
|
||||
}
|
||||
Thread.sleep(taskBatchProperties.getFailOnJobFailurePollInterval());
|
||||
|
||||
if (repeatStatus.equals(RepeatStatus.FINISHED) && failedJobExecutions.size() > 0) {
|
||||
throwJobFailedException(failedJobExecutions);
|
||||
}
|
||||
if (repeatStatus.isContinuable() && taskBatchProperties.getFailOnJobFailurewaitTime() != 0
|
||||
&& (new Date()).getTime() - startDate.getTime() > taskBatchProperties.getFailOnJobFailurewaitTime()) {
|
||||
throw new IllegalStateException("Not all jobs were completed " +
|
||||
"within the time specified by spring.cloud.task.batch." +
|
||||
"failOnJobFailurewaitTime.");
|
||||
}
|
||||
return repeatStatus;
|
||||
}
|
||||
execute(job, jobParameters);
|
||||
|
||||
});
|
||||
}
|
||||
|
||||
public void throwJobFailedException(List<JobExecution> failedJobExecutions) {
|
||||
String message = "The following Jobs have failed: \n";
|
||||
for (JobExecution failedJobExecution : failedJobExecutions) {
|
||||
message += String.format("Job %s failed during " +
|
||||
"execution for jobId %s with jobExecutionId of %s \n",
|
||||
failedJobExecution.getJobInstance().getJobName(),
|
||||
failedJobExecution.getJobId(), failedJobExecution.getId());
|
||||
}
|
||||
logger.error(message);
|
||||
throw new TaskException(message);
|
||||
|
||||
}
|
||||
private JobParameters removeNonIdentifying(JobParameters parameters) {
|
||||
Map<String, JobParameter> parameterMap = parameters.getParameters();
|
||||
HashMap<String, JobParameter> copy = new HashMap<>(parameterMap);
|
||||
for (Map.Entry<String, JobParameter> parameter : copy.entrySet()) {
|
||||
if (!parameter.getValue().isIdentifying()) {
|
||||
parameterMap.remove(parameter.getKey());
|
||||
}
|
||||
}
|
||||
return new JobParameters(parameterMap);
|
||||
}
|
||||
private boolean isStoppedOrFailed(JobExecution execution) {
|
||||
BatchStatus status = execution.getStatus();
|
||||
return (status == BatchStatus.STOPPED || status == BatchStatus.FAILED);
|
||||
}
|
||||
private JobParameters merge(JobParameters parameters, JobParameters additionals) {
|
||||
Map<String, JobParameter> merged = new HashMap<>();
|
||||
merged.putAll(parameters.getParameters());
|
||||
merged.putAll(additionals.getParameters());
|
||||
return new JobParameters(merged);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user