Make JobOperator extend JobLauncher

Related to #4832

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java

@@ -52,8 +52,6 @@ class BatchRegistrar implements ImportBeanDefinitionRegistrar {
 
 	private static final String JOB_REPOSITORY = "jobRepository";
 
-	private static final String JOB_LAUNCHER = "jobLauncher";
-
 	private static final String JOB_REGISTRY = "jobRegistry";
 
 	private static final String JOB_LOADER = "jobLoader";
@@ -67,7 +65,6 @@ public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, B
 			.get(EnableBatchProcessing.class)
 			.synthesize();
 		registerJobRepository(registry, batchAnnotation);
-		registerJobLauncher(registry, batchAnnotation);
 		registerJobRegistry(registry);
 		registerJobRegistrySmartInitializingSingleton(registry);
 		registerJobOperator(registry, batchAnnotation);
@@ -147,25 +144,6 @@ private void registerJobRepository(BeanDefinitionRegistry registry, EnableBatchP
 		registry.registerBeanDefinition(JOB_REPOSITORY, beanDefinitionBuilder.getBeanDefinition());
 	}
 
-	private void registerJobLauncher(BeanDefinitionRegistry registry, EnableBatchProcessing batchAnnotation) {
-		if (registry.containsBeanDefinition(JOB_LAUNCHER)) {
-			LOGGER.info("Bean jobLauncher already defined in the application context, skipping"
-					+ " the registration of a jobLauncher");
-			return;
-		}
-		BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder
-			.genericBeanDefinition(TaskExecutorJobLauncher.class);
-		// set mandatory properties
-		beanDefinitionBuilder.addPropertyReference(JOB_REPOSITORY, JOB_REPOSITORY);
-
-		// set optional properties
-		String taskExecutorRef = batchAnnotation.taskExecutorRef();
-		if (registry.containsBeanDefinition(taskExecutorRef)) {
-			beanDefinitionBuilder.addPropertyReference("taskExecutor", taskExecutorRef);
-		}
-		registry.registerBeanDefinition(JOB_LAUNCHER, beanDefinitionBuilder.getBeanDefinition());
-	}
-
 	private void registerJobRegistry(BeanDefinitionRegistry registry) {
 		if (registry.containsBeanDefinition(JOB_REGISTRY)) {
 			LOGGER.info("Bean jobRegistry already defined in the application context, skipping"
@@ -205,7 +183,6 @@ private void registerJobOperator(BeanDefinitionRegistry registry, EnableBatchPro
 		beanDefinitionBuilder.addPropertyReference("transactionManager", transactionManagerRef);
 
 		beanDefinitionBuilder.addPropertyReference(JOB_REPOSITORY, JOB_REPOSITORY);
-		beanDefinitionBuilder.addPropertyReference(JOB_LAUNCHER, JOB_LAUNCHER);
 		beanDefinitionBuilder.addPropertyReference(JOB_REGISTRY, JOB_REGISTRY);
 
 		// set optional properties

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java

@@ -39,7 +39,6 @@
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.launch.support.JobOperatorFactoryBean;
-import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
 import org.springframework.batch.core.repository.ExecutionContextSerializer;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.dao.AbstractJdbcBatchMetadataDao;
@@ -145,27 +144,6 @@ public JobRepository jobRepository() throws BatchConfigurationException {
 		}
 	}
 
-	/**
-	 * Define a job launcher bean.
-	 * @param jobRepository the job repository
-	 * @return a job launcher
-	 * @throws BatchConfigurationException if unable to configure the default job launcher
-	 * @since 5.2
-	 */
-	@Bean
-	public JobLauncher jobLauncher(JobRepository jobRepository) throws BatchConfigurationException {
-		TaskExecutorJobLauncher taskExecutorJobLauncher = new TaskExecutorJobLauncher();
-		taskExecutorJobLauncher.setJobRepository(jobRepository);
-		taskExecutorJobLauncher.setTaskExecutor(getTaskExecutor());
-		try {
-			taskExecutorJobLauncher.afterPropertiesSet();
-			return taskExecutorJobLauncher;
-		}
-		catch (Exception e) {
-			throw new BatchConfigurationException("Unable to configure the default job launcher", e);
-		}
-	}
-
 	@Bean
 	public JobRegistry jobRegistry() throws BatchConfigurationException {
 		return new MapJobRegistry();
@@ -175,20 +153,19 @@ public JobRegistry jobRegistry() throws BatchConfigurationException {
 	 * Define a job operator bean.
 	 * @param jobRepository a job repository
 	 * @param jobRegistry a job registry
-	 * @param jobLauncher a job launcher
 	 * @return a job operator
 	 * @throws BatchConfigurationException if unable to configure the default job operator
 	 * @since 5.2
 	 */
 	@Bean
-	public JobOperator jobOperator(JobRepository jobRepository, JobRegistry jobRegistry, JobLauncher jobLauncher)
+	public JobOperator jobOperator(JobRepository jobRepository, JobRegistry jobRegistry)
 			throws BatchConfigurationException {
 		JobOperatorFactoryBean jobOperatorFactoryBean = new JobOperatorFactoryBean();
-		jobOperatorFactoryBean.setTransactionManager(getTransactionManager());
 		jobOperatorFactoryBean.setJobRepository(jobRepository);
 		jobOperatorFactoryBean.setJobRegistry(jobRegistry);
-		jobOperatorFactoryBean.setJobLauncher(jobLauncher);
+		jobOperatorFactoryBean.setTransactionManager(getTransactionManager());
 		jobOperatorFactoryBean.setJobParametersConverter(getJobParametersConverter());
+		jobOperatorFactoryBean.setTaskExecutor(getTaskExecutor());
 		try {
 			jobOperatorFactoryBean.afterPropertiesSet();
 			return jobOperatorFactoryBean.getObject();

spring-batch-core/src/main/java/org/springframework/batch/core/launch/JobOperator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2024 the original author or authors.
+ * Copyright 2006-2025 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.
@@ -42,7 +42,7 @@
  * @author Mahmoud Ben Hassine
  * @since 2.0
  */
-public interface JobOperator {
+public interface JobOperator extends JobLauncher {
 
 	/**
 	 * List the {@link JobExecution JobExecutions} associated with a particular

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/JobOperatorFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2024 the original author or authors.
+ * Copyright 2022-2025 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.
@@ -17,15 +17,20 @@
 
 import java.util.Properties;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.aop.framework.ProxyFactory;
 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.launch.JobLauncher;
 import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionManager;
 import org.springframework.transaction.annotation.Isolation;
@@ -46,6 +51,8 @@
  */
 public class JobOperatorFactoryBean implements FactoryBean<JobOperator>, InitializingBean {
 
+	protected static final Log logger = LogFactory.getLog(JobOperatorFactoryBean.class);
+
 	private static final String TRANSACTION_ISOLATION_LEVEL_PREFIX = "ISOLATION_";
 
 	private static final String TRANSACTION_PROPAGATION_PREFIX = "PROPAGATION_";
@@ -56,20 +63,25 @@ public class JobOperatorFactoryBean implements FactoryBean<JobOperator>, Initial
 
 	private JobRegistry jobRegistry;
 
-	private JobLauncher jobLauncher;
-
 	private JobRepository jobRepository;
 
 	private JobParametersConverter jobParametersConverter = new DefaultJobParametersConverter();
 
+	private TaskExecutor taskExecutor;
+
+	private MeterRegistry meterRegistry = Metrics.globalRegistry;
+
 	private final ProxyFactory proxyFactory = new ProxyFactory();
 
 	@Override
 	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(this.transactionManager, "TransactionManager must not be null");
-		Assert.notNull(this.jobLauncher, "JobLauncher must not be null");
-		Assert.notNull(this.jobRegistry, "JobRegistry must not be null");
 		Assert.notNull(this.jobRepository, "JobRepository must not be null");
+		Assert.notNull(this.jobRegistry, "JobRegistry must not be null");
+		Assert.notNull(this.transactionManager, "TransactionManager must not be null");
+		if (this.taskExecutor == null) {
+			logger.info("No TaskExecutor has been set, defaulting to synchronous executor.");
+			this.taskExecutor = new SyncTaskExecutor();
+		}
 		if (this.transactionAttributeSource == null) {
 			Properties transactionAttributes = new Properties();
 			String transactionProperties = String.join(",", TRANSACTION_PROPAGATION_PREFIX + Propagation.REQUIRED,
@@ -88,14 +100,6 @@ public void setJobRegistry(JobRegistry jobRegistry) {
 		this.jobRegistry = jobRegistry;
 	}
 
-	/**
-	 * Setter for the job launcher.
-	 * @param jobLauncher the job launcher to set
-	 */
-	public void setJobLauncher(JobLauncher jobLauncher) {
-		this.jobLauncher = jobLauncher;
-	}
-
 	/**
 	 * Setter for the job repository.
 	 * @param jobRepository the job repository to set
@@ -112,6 +116,25 @@ public void setJobParametersConverter(JobParametersConverter jobParametersConver
 		this.jobParametersConverter = jobParametersConverter;
 	}
 
+	/**
+	 * Set the TaskExecutor. (Optional)
+	 * @param taskExecutor instance of {@link TaskExecutor}.
+	 * @since 6.0
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * Set the meter registry to use for metrics. Defaults to
+	 * {@link Metrics#globalRegistry}.
+	 * @param meterRegistry the meter registry
+	 * @since 6.0
+	 */
+	public void setMeterRegistry(MeterRegistry meterRegistry) {
+		this.meterRegistry = meterRegistry;
+	}
+
 	/**
 	 * Setter for the transaction manager.
 	 * @param transactionManager the transaction manager to set
@@ -155,7 +178,8 @@ private SimpleJobOperator getTarget() throws Exception {
 		SimpleJobOperator simpleJobOperator = new SimpleJobOperator();
 		simpleJobOperator.setJobRegistry(this.jobRegistry);
 		simpleJobOperator.setJobRepository(this.jobRepository);
-		simpleJobOperator.setJobLauncher(this.jobLauncher);
+		simpleJobOperator.setTaskExecutor(this.taskExecutor);
+		simpleJobOperator.setMeterRegistry(this.meterRegistry);
 		simpleJobOperator.setJobParametersConverter(this.jobParametersConverter);
 		simpleJobOperator.afterPropertiesSet();
 		return simpleJobOperator;

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2024 the original author or authors.
+ * Copyright 2006-2025 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.
@@ -86,17 +86,13 @@
  * @author Mahmoud Ben Hassine
  * @since 2.0
  */
-public class SimpleJobOperator implements JobOperator, InitializingBean {
+public class SimpleJobOperator extends TaskExecutorJobLauncher implements JobOperator, InitializingBean {
 
 	private static final String ILLEGAL_STATE_MSG = "Illegal state (only happens on a race condition): "
 			+ "%s with name=%s and parameters=%s";
 
 	private ListableJobLocator jobRegistry;
 
-	private JobLauncher jobLauncher;
-
-	private JobRepository jobRepository;
-
 	private JobParametersConverter jobParametersConverter = new DefaultJobParametersConverter();
 
 	private final Log logger = LogFactory.getLog(getClass());
@@ -108,9 +104,8 @@ public class SimpleJobOperator implements JobOperator, InitializingBean {
 	 */
 	@Override
 	public void afterPropertiesSet() throws Exception {
-		Assert.state(jobLauncher != null, "JobLauncher must be provided");
+		super.afterPropertiesSet();
 		Assert.state(jobRegistry != null, "JobLocator must be provided");
-		Assert.state(jobRepository != null, "JobRepository must be provided");
 	}
 
 	/**
@@ -129,18 +124,6 @@ public void setJobRegistry(ListableJobLocator jobRegistry) {
 		this.jobRegistry = jobRegistry;
 	}
 
-	public void setJobRepository(JobRepository jobRepository) {
-		this.jobRepository = jobRepository;
-	}
-
-	/**
-	 * Public setter for the {@link JobLauncher}.
-	 * @param jobLauncher the {@link JobLauncher} to set
-	 */
-	public void setJobLauncher(JobLauncher jobLauncher) {
-		this.jobLauncher = jobLauncher;
-	}
-
 	@Override
 	public List<Long> getExecutions(long instanceId) throws NoSuchJobInstanceException {
 		JobInstance jobInstance = jobRepository.getJobInstance(instanceId);
@@ -233,7 +216,7 @@ public Long restart(long executionId) throws JobInstanceAlreadyCompleteException
 			logger.info(String.format("Attempting to resume job with name=%s and parameters=%s", jobName, parameters));
 		}
 		try {
-			return jobLauncher.run(job, parameters).getId();
+			return run(job, parameters).getId();
 		}
 		catch (JobExecutionAlreadyRunningException e) {
 			throw new UnexpectedJobExecutionException(
@@ -263,7 +246,7 @@ public Long start(String jobName, Properties parameters)
 				.info(String.format("Attempting to launch job with name=%s and parameters={%s}", jobName, parameters));
 		}
 		try {
-			return jobLauncher.run(job, jobParameters).getId();
+			return run(job, jobParameters).getId();
 		}
 		catch (JobExecutionAlreadyRunningException e) {
 			throw new UnexpectedJobExecutionException(
@@ -293,7 +276,7 @@ public Long startNextInstance(String jobName)
 			logger.info(String.format("Attempting to launch job with name=%s and parameters=%s", jobName, parameters));
 		}
 		try {
-			return jobLauncher.run(job, parameters).getId();
+			return run(job, parameters).getId();
 		}
 		catch (JobExecutionAlreadyRunningException e) {
 			throw new UnexpectedJobExecutionException(

spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/TaskExecutorJobLauncher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2024 the original author or authors.
+ * Copyright 2022-2025 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,13 +70,13 @@ public class TaskExecutorJobLauncher implements JobLauncher, InitializingBean {
 
 	protected static final Log logger = LogFactory.getLog(TaskExecutorJobLauncher.class);
 
-	private JobRepository jobRepository;
+	protected JobRepository jobRepository;
 
-	private TaskExecutor taskExecutor;
+	protected TaskExecutor taskExecutor;
 
-	private MeterRegistry meterRegistry = Metrics.globalRegistry;
+	protected MeterRegistry meterRegistry = Metrics.globalRegistry;
 
-	private Counter jobLaunchCount; // NoopCounter is still incubating
+	protected Counter jobLaunchCount; // NoopCounter is still incubating
 
 	/**
 	 * Run the provided job with the given {@link JobParameters}. The

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2024 the original author or authors.
+ * Copyright 2022-2025 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.
@@ -77,7 +77,6 @@ void testConfigurationWithUserDefinedBeans() {
 		var context = new AnnotationConfigApplicationContext(JobConfigurationWithUserDefinedInfrastructureBeans.class);
 
 		Assertions.assertTrue(Mockito.mockingDetails(context.getBean(JobRepository.class)).isMock());
-		Assertions.assertTrue(Mockito.mockingDetails(context.getBean(JobLauncher.class)).isMock());
 		Assertions.assertTrue(Mockito.mockingDetails(context.getBean(JobRegistry.class)).isMock());
 		Assertions.assertTrue(Mockito.mockingDetails(context.getBean(JobOperator.class)).isMock());
 		Assertions