From 8dde8529d36b646b33a1711219a1b1e8a046345a Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Thu, 22 May 2025 19:06:21 +0200 Subject: [PATCH] Make JobOperator use domain types in method signatures Resolves #4845 --- .../batch/core/launch/JobOperator.java | 86 +++++++++++++++++++ .../launch/support/SimpleJobOperator.java | 78 +++++++++++++++-- 2 files changed, 158 insertions(+), 6 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java index 5ed837e47..2050d361e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java @@ -28,6 +28,7 @@ import org.springframework.batch.core.JobParametersIncrementer; import org.springframework.batch.core.JobParametersInvalidException; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.configuration.JobRegistry; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRestartException; @@ -46,7 +47,10 @@ public interface JobOperator extends JobLauncher { * List the available job names that can be launched with * {@link #start(String, Properties)}. * @return a set of job names + * @deprecated since 6.0 in favor of {@link JobRegistry#getJobNames()}. Scheduled for + * removal in 6.2 or later. */ + @Deprecated(since = "6.0", forRemoval = true) Set getJobNames(); /** @@ -104,10 +108,32 @@ public interface JobOperator extends JobLauncher { * @throws JobRestartException if there is a non-specific error with the restart (e.g. * corrupt or inconsistent restart data) * @throws JobParametersInvalidException if the parameters are not valid for this job + * @deprecated since 6.0 in favor of {@link #restart(JobExecution)}. Scheduled for + * removal in 6.2 or later. */ + @Deprecated(since = "6.0", forRemoval = true) Long restart(long executionId) throws JobInstanceAlreadyCompleteException, NoSuchJobExecutionException, NoSuchJobException, JobRestartException, JobParametersInvalidException; + /** + * Restart a failed or stopped {@link JobExecution}. Fails with an exception if the + * execution provided does not exist or corresponds to a {@link JobInstance} that in + * normal circumstances already completed successfully. + * @param jobExecution the failed or stopped {@link JobExecution} to restart + * @return the {@link JobExecution} that was started + * @throws JobInstanceAlreadyCompleteException if the job was already successfully + * completed + * @throws NoSuchJobExecutionException if the id was not associated with any + * {@link JobExecution} + * @throws NoSuchJobException if the {@link JobExecution} was found, but its + * corresponding {@link Job} is no longer available for launching + * @throws JobRestartException if there is a non-specific error with the restart (e.g. + * corrupt or inconsistent restart data) + * @throws JobParametersInvalidException if the parameters are not valid for this job + */ + JobExecution restart(JobExecution jobExecution) throws JobInstanceAlreadyCompleteException, + NoSuchJobExecutionException, NoSuchJobException, JobRestartException, JobParametersInvalidException; + /** * Launch the next in a sequence of {@link JobInstance} determined by the * {@link JobParametersIncrementer} attached to the specified job. If the previous @@ -132,11 +158,42 @@ public interface JobOperator extends JobLauncher { * that is already executing. * @throws JobInstanceAlreadyCompleteException thrown if attempting to restart a * completed job. + * @deprecated since 6.0 in favor of {@link #startNextInstance(Job)}. Scheduled for + * removal in 6.2 or later. */ + @Deprecated(since = "6.0", forRemoval = true) Long startNextInstance(String jobName) throws NoSuchJobException, JobParametersNotFoundException, JobRestartException, JobExecutionAlreadyRunningException, JobInstanceAlreadyCompleteException, UnexpectedJobExecutionException, JobParametersInvalidException; + /** + * Launch the next in a sequence of {@link JobInstance} determined by the + * {@link JobParametersIncrementer} attached to the specified job. If the previous + * instance is still in a failed state, this method should still create a new instance + * and run it with different parameters (as long as the + * {@link JobParametersIncrementer} is working).
+ *
+ * + * The last three exception described below should be extremely unlikely, but cannot + * be ruled out entirely. It points to some other thread or process trying to use this + * method (or a similar one) at the same time. + * @param job the job to launch + * @return the {@link JobExecution} created when the job is launched + * @throws NoSuchJobException if there is no such job definition available + * @throws JobParametersNotFoundException if the parameters cannot be found + * @throws JobParametersInvalidException thrown if some of the job parameters are + * invalid. + * @throws UnexpectedJobExecutionException if an unexpected condition arises + * @throws JobRestartException thrown if a job is restarted illegally. + * @throws JobExecutionAlreadyRunningException thrown if attempting to restart a job + * that is already executing. + * @throws JobInstanceAlreadyCompleteException thrown if attempting to restart a + * completed job. + */ + JobExecution startNextInstance(Job job) throws NoSuchJobException, JobParametersNotFoundException, + JobRestartException, JobExecutionAlreadyRunningException, JobInstanceAlreadyCompleteException, + UnexpectedJobExecutionException, JobParametersInvalidException; + /** * Send a stop signal to the {@link JobExecution} with the supplied id. The signal is * successfully sent if this method returns true, but that doesn't mean that the job @@ -148,9 +205,24 @@ public interface JobOperator extends JobLauncher { * supplied * @throws JobExecutionNotRunningException if the {@link JobExecution} is not running * (so cannot be stopped) + * @deprecated since 6.0 in favor of {@link #stop(JobExecution)}. Scheduled for + * removal in 6.2 or later. */ + @Deprecated(since = "6.0", forRemoval = true) boolean stop(long executionId) throws NoSuchJobExecutionException, JobExecutionNotRunningException; + /** + * Send a stop signal to the supplied {@link JobExecution}. The signal is successfully + * sent if this method returns true, but that doesn't mean that the job has stopped. + * The only way to be sure of that is to poll the job execution status. + * @param jobExecution the running {@link JobExecution} + * @return true if the message was successfully sent (does not guarantee that the job + * has stopped) + * @throws JobExecutionNotRunningException if the supplied {@link JobExecution} is not + * running (so cannot be stopped) + */ + boolean stop(JobExecution jobExecution) throws JobExecutionNotRunningException; + /** * Mark the {@link JobExecution} as ABANDONED. If a stop signal is ignored because the * process died this is the best way to mark a job as finished with (as opposed to @@ -161,9 +233,23 @@ public interface JobOperator extends JobLauncher { * jobExecutionId. * @throws JobExecutionAlreadyRunningException if the job is running (it should be * stopped first) + * @deprecated since 6.0 in favor of {@link #abandon(JobExecution)}. Scheduled for + * removal in 6.2 or later. */ + @Deprecated(since = "6.0", forRemoval = true) JobExecution abandon(long jobExecutionId) throws NoSuchJobExecutionException, JobExecutionAlreadyRunningException; + /** + * Mark the {@link JobExecution} as ABANDONED. If a stop signal is ignored because the + * process died this is the best way to mark a job as finished with (as opposed to + * STOPPED). An abandoned job execution cannot be restarted by the framework. + * @param jobExecution the job execution to abort + * @return the {@link JobExecution} that was aborted + * @throws JobExecutionAlreadyRunningException if the job execution is running (it + * should be stopped first) + */ + JobExecution abandon(JobExecution jobExecution) throws JobExecutionAlreadyRunningException; + /** * List the {@link JobExecution JobExecutions} associated with a particular * {@link JobInstance}, in reverse order of creation (and therefore usually of diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java index ddaac05fb..37f89be51 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java @@ -125,11 +125,7 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe } @Override - public Set getJobNames() { - return new TreeSet<>(jobRegistry.getJobNames()); - } - - @Override + @Deprecated(since = "6.0", forRemoval = true) public Long start(String jobName, Properties parameters) throws NoSuchJobException, JobInstanceAlreadyExistsException, JobParametersInvalidException { if (logger.isInfoEnabled()) { @@ -168,6 +164,7 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe } @Override + @Deprecated(since = "6.0", forRemoval = true) public Long restart(long executionId) throws JobInstanceAlreadyCompleteException, NoSuchJobExecutionException, NoSuchJobException, JobRestartException, JobParametersInvalidException { @@ -194,6 +191,28 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe } @Override + public JobExecution restart(JobExecution jobExecution) throws JobInstanceAlreadyCompleteException, + NoSuchJobExecutionException, NoSuchJobException, JobRestartException, JobParametersInvalidException { + + String jobName = jobExecution.getJobInstance().getJobName(); + Job job = jobRegistry.getJob(jobName); + JobParameters parameters = jobExecution.getJobParameters(); + + if (logger.isInfoEnabled()) { + logger.info(String.format("Attempting to resume job with name=%s and parameters=%s", jobName, parameters)); + } + try { + return run(job, parameters); + } + catch (JobExecutionAlreadyRunningException e) { + throw new UnexpectedJobExecutionException( + String.format(ILLEGAL_STATE_MSG, "job execution already running", jobName, parameters), e); + } + + } + + @Override + @Deprecated(since = "6.0", forRemoval = true) public Long startNextInstance(String jobName) throws NoSuchJobException, UnexpectedJobExecutionException, JobParametersInvalidException { if (logger.isInfoEnabled()) { @@ -224,9 +243,43 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe } @Override + public JobExecution startNextInstance(Job job) + throws NoSuchJobException, UnexpectedJobExecutionException, JobParametersInvalidException { + + JobParameters parameters = new JobParametersBuilder(jobRepository).getNextJobParameters(job).toJobParameters(); + if (logger.isInfoEnabled()) { + logger.info(String.format("Attempting to launch job with name=%s and parameters=%s", job.getName(), + parameters)); + } + try { + return run(job, parameters); + } + catch (JobExecutionAlreadyRunningException e) { + throw new UnexpectedJobExecutionException( + String.format(ILLEGAL_STATE_MSG, "job already running", job.getName(), parameters), e); + } + catch (JobRestartException e) { + throw new UnexpectedJobExecutionException( + String.format(ILLEGAL_STATE_MSG, "job not restartable", job.getName(), parameters), e); + } + catch (JobInstanceAlreadyCompleteException e) { + throw new UnexpectedJobExecutionException( + String.format(ILLEGAL_STATE_MSG, "job instance already complete", job.getName(), parameters), e); + } + + } + + @Override + @Deprecated(since = "6.0", forRemoval = true) public boolean stop(long executionId) throws NoSuchJobExecutionException, JobExecutionNotRunningException { JobExecution jobExecution = findExecutionById(executionId); + return stop(jobExecution); + } + + @Override + public boolean stop(JobExecution jobExecution) throws JobExecutionNotRunningException { + // Indicate the execution should be stopped by setting it's status to // 'STOPPING'. It is assumed that // the step implementation will check this status at chunk boundaries. @@ -241,7 +294,7 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe try { Job job = jobRegistry.getJob(jobExecution.getJobInstance().getJobName()); if (job instanceof StepLocator) {// can only process as StepLocator is the - // only way to get the step object + // only way to get the step object // get the current stepExecution for (StepExecution stepExecution : jobExecution.getStepExecutions()) { if (stepExecution.getStatus().isRunning()) { @@ -272,10 +325,17 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe } @Override + @Deprecated(since = "6.0", forRemoval = true) public JobExecution abandon(long jobExecutionId) throws NoSuchJobExecutionException, JobExecutionAlreadyRunningException { JobExecution jobExecution = findExecutionById(jobExecutionId); + return abandon(jobExecution); + } + + @Override + public JobExecution abandon(JobExecution jobExecution) throws JobExecutionAlreadyRunningException { + if (jobExecution.getStatus().isLessThan(BatchStatus.STOPPING)) { throw new JobExecutionAlreadyRunningException( "JobExecution is running or complete and therefore cannot be aborted"); @@ -290,6 +350,12 @@ public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOpe return jobExecution; } + @Override + @Deprecated(since = "6.0", forRemoval = true) + public Set getJobNames() { + return new TreeSet<>(jobRegistry.getJobNames()); + } + @Override @Deprecated(since = "6.0", forRemoval = true) public List getExecutions(long instanceId) throws NoSuchJobInstanceException {