diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java
index f8b3fef27..028cf53b3 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java
@@ -22,8 +22,10 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.batch.core.configuration.support.AutomaticJobRegistrar;
 import org.springframework.batch.core.configuration.support.DefaultJobLoader;
+import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
 import org.springframework.batch.core.configuration.support.MapJobRegistry;
 import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
+import org.springframework.batch.core.launch.support.JobOperatorFactoryBean;
 import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
 import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -60,6 +62,8 @@ class BatchRegistrar implements ImportBeanDefinitionRegistrar {
 		registerJobExplorer(registry, batchAnnotation, importingClassName);
 		registerJobLauncher(registry, batchAnnotation, importingClassName);
 		registerJobRegistry(registry);
+		registerJobRegistryBeanPostProcessor(registry);
+		registerJobOperator(registry, batchAnnotation);
 		registerAutomaticJobRegistrar(registry, batchAnnotation);
 		watch.stop();
 		LOGGER.info(LogMessage.format("Finished Spring Batch infrastrucutre beans configuration in %s ms.",
@@ -208,6 +212,39 @@ class BatchRegistrar implements ImportBeanDefinitionRegistrar {
 		registry.registerBeanDefinition("jobRegistry", beanDefinition);
 	}
 
+	private void registerJobRegistryBeanPostProcessor(BeanDefinitionRegistry registry) {
+		if (registry.containsBeanDefinition("jobRegistryBeanPostProcessor")) {
+			LOGGER.info("Bean jobRegistryBeanPostProcessor already defined in the application context, skipping"
+					+ " the registration of a jobRegistryBeanPostProcessor");
+			return;
+		}
+		BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(JobRegistryBeanPostProcessor.class);
+		beanDefinitionBuilder.addPropertyReference("jobRegistry", "jobRegistry");
+
+		registry.registerBeanDefinition("jobRegistryBeanPostProcessor", beanDefinitionBuilder.getBeanDefinition());
+	}
+
+	private void registerJobOperator(BeanDefinitionRegistry registry, EnableBatchProcessing batchAnnotation) {
+		if (registry.containsBeanDefinition("jobOperator")) {
+			LOGGER.info("Bean jobOperator already defined in the application context, skipping"
+					+ " the registration of a jobOperator");
+			return;
+		}
+		BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder
+				.genericBeanDefinition(JobOperatorFactoryBean.class);
+		// set mandatory properties
+		String transactionManagerRef = batchAnnotation.transactionManagerRef();
+		beanDefinitionBuilder.addPropertyReference("transactionManager", transactionManagerRef);
+
+		beanDefinitionBuilder.addPropertyReference("jobRepository", "jobRepository");
+		beanDefinitionBuilder.addPropertyReference("jobLauncher", "jobLauncher");
+		beanDefinitionBuilder.addPropertyReference("jobExplorer", "jobExplorer");
+		beanDefinitionBuilder.addPropertyReference("jobRegistry", "jobRegistry");
+
+		registry.registerBeanDefinition("jobOperator", beanDefinitionBuilder.getBeanDefinition());
+	}
+
 	private void registerAutomaticJobRegistrar(BeanDefinitionRegistry registry, EnableBatchProcessing batchAnnotation) {
 		if (!batchAnnotation.modular()) {
 			return;
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java
index ac28ad33b..e11e0a8f0 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java
@@ -88,6 +88,13 @@ import org.springframework.context.annotation.Import;
  * <li>a {@link org.springframework.batch.core.explore.JobExplorer} (bean name
  * "jobExplorer" of type
  * {@link org.springframework.batch.core.explore.support.SimpleJobExplorer})</li>
+ * <li>a {@link org.springframework.batch.core.launch.JobOperator} (bean name
+ * "jobOperator" of type
+ * {@link org.springframework.batch.core.launch.support.SimpleJobOperator})</li>
+ * <li>a
+ * {@link org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor}
+ * (bean name "jobRegistryBeanPostProcessor" of type
+ * {@link org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor})</li>
  * </ul>
  *
  * If the configuration is specified as <code>modular=true</code>, the context also
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java
index 0d85e6efa..ef74ed607 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java
@@ -29,6 +29,8 @@ import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
 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;
@@ -73,6 +75,8 @@ import org.springframework.transaction.annotation.Isolation;
  * <li>a {@link JobExplorer} named "jobExplorer"</li>
  * <li>a {@link JobLauncher} named "jobLauncher"</li>
  * <li>a {@link JobRegistry} named "jobRegistry"</li>
+ * <li>a {@link JobOperator} named "JobOperator"</li>
+ * <li>a {@link JobRegistryBeanPostProcessor} named "jobRegistryBeanPostProcessor"</li>
  * <li>a {@link org.springframework.batch.core.scope.StepScope} named "stepScope"</li>
  * <li>a {@link org.springframework.batch.core.scope.JobScope} named "jobScope"</li>
  * </ul>
@@ -177,10 +181,40 @@ public class DefaultBatchConfiguration implements ApplicationContextAware {
 	}
 
 	@Bean
-	public JobRegistry jobRegistry() throws Exception {
+	public JobRegistry jobRegistry() throws BatchConfigurationException {
 		return this.jobRegistry; // FIXME returning a new instance here does not work
 	}
 
+	@Bean
+	public JobOperator jobOperator() throws BatchConfigurationException {
+		JobOperatorFactoryBean jobOperatorFactoryBean = new JobOperatorFactoryBean();
+		jobOperatorFactoryBean.setTransactionManager(getTransactionManager());
+		jobOperatorFactoryBean.setJobRepository(jobRepository());
+		jobOperatorFactoryBean.setJobExplorer(jobExplorer());
+		jobOperatorFactoryBean.setJobRegistry(jobRegistry());
+		jobOperatorFactoryBean.setJobLauncher(jobLauncher());
+		try {
+			jobOperatorFactoryBean.afterPropertiesSet();
+			return jobOperatorFactoryBean.getObject();
+		}
+		catch (Exception e) {
+			throw new BatchConfigurationException("Unable to configure the default job operator", e);
+		}
+	}
+
+	@Bean
+	public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() throws BatchConfigurationException {
+		JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
+		jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry());
+		try {
+			jobRegistryBeanPostProcessor.afterPropertiesSet();
+			return jobRegistryBeanPostProcessor;
+		}
+		catch (Exception e) {
+			throw new BatchConfigurationException("Unable to configure the default job registry BeanPostProcessor", e);
+		}
+	}
+
 	/*
 	 * Getters to customize the configuration of infrastructure beans
 	 */
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java
index 1b42ef4f1..fd15239ca 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java
@@ -28,6 +28,7 @@ import org.springframework.aop.framework.Advised;
 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.launch.JobOperator;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.dao.JdbcExecutionContextDao;
 import org.springframework.batch.core.repository.dao.JdbcJobExecutionDao;
@@ -88,6 +89,8 @@ class BatchRegistrarTests {
 				context.getBean(JobLauncher.class));
 		Assertions.assertEquals(JobConfigurationWithUserDefinedInfrastrucutreBeans.jobRegistry,
 				context.getBean(JobRegistry.class));
+		Assertions.assertEquals(JobConfigurationWithUserDefinedInfrastrucutreBeans.jobOperator,
+				context.getBean(JobOperator.class));
 	}
 
 	@Test
@@ -159,6 +162,26 @@ class BatchRegistrarTests {
 		Assertions.assertEquals(context.getBean(JdbcTransactionManager.class), transactionManager);
 	}
 
+	@Test
+	void testDefaultInfrastructureBeansRegistration() {
+		// given
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(JobConfiguration.class);
+
+		// when
+		JobLauncher jobLauncher = context.getBean(JobLauncher.class);
+		JobRepository jobRepository = context.getBean(JobRepository.class);
+		JobExplorer jobExplorer = context.getBean(JobExplorer.class);
+		JobRegistry jobRegistry = context.getBean(JobRegistry.class);
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+
+		// then
+		Assertions.assertNotNull(jobLauncher);
+		Assertions.assertNotNull(jobRepository);
+		Assertions.assertNotNull(jobExplorer);
+		Assertions.assertNotNull(jobRegistry);
+		Assertions.assertNotNull(jobOperator);
+	}
+
 	@Configuration
 	@EnableBatchProcessing
 	public static class JobConfigurationWithoutDataSource {
@@ -188,6 +211,8 @@ class BatchRegistrarTests {
 
 		public static JobRegistry jobRegistry = Mockito.mock(JobRegistry.class);
 
+		public static JobOperator jobOperator = Mockito.mock(JobOperator.class);
+
 		@Bean
 		public JobRepository jobRepository() {
 			return jobRepository;
@@ -208,6 +233,11 @@ class BatchRegistrarTests {
 			return jobRegistry;
 		}
 
+		@Bean
+		public JobOperator jobOperator() {
+			return jobOperator;
+		}
+
 	}
 
 	@Configuration
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/DefaultBatchConfigurationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/DefaultBatchConfigurationTests.java
index 76f0b20ae..7815891e1 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/DefaultBatchConfigurationTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/support/DefaultBatchConfigurationTests.java
@@ -28,9 +28,12 @@ import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.Step;
+import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.configuration.xml.DummyJobRepository;
+import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.job.builder.JobBuilder;
 import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.core.step.tasklet.Tasklet;
@@ -93,6 +96,26 @@ class DefaultBatchConfigurationTests {
 		Assertions.assertInstanceOf(DummyJobRepository.class, jobRepository);
 	}
 
+	@Test
+	void testDefaultInfrastructureBeansRegistration() {
+		// given
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(MyJobConfiguration.class);
+
+		// when
+		JobLauncher jobLauncher = context.getBean(JobLauncher.class);
+		JobRepository jobRepository = context.getBean(JobRepository.class);
+		JobExplorer jobExplorer = context.getBean(JobExplorer.class);
+		JobRegistry jobRegistry = context.getBean(JobRegistry.class);
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+
+		// then
+		Assertions.assertNotNull(jobLauncher);
+		Assertions.assertNotNull(jobRepository);
+		Assertions.assertNotNull(jobExplorer);
+		Assertions.assertNotNull(jobRegistry);
+		Assertions.assertNotNull(jobOperator);
+	}
+
 	@Configuration
 	static class MyJobConfigurationWithoutDataSource extends DefaultBatchConfiguration {