Add builders to simplify remote partitioning setup

Resolves BATCH-2687

* use hard coded constants in docs
* change visibility of method `beanFactory` to public in builders

spring-batch-docs/asciidoc/spring-batch-integration.adoc

@@ -945,7 +945,7 @@ two beans that can be autowired in the application context:
 * `RemoteChunkingMasterStepBuilderFactory`: used to configure the master step
 * `RemoteChunkingWorkerBuilder`: used to configure the remote worker integration flow
 
-These APIs will take care of configuring a number of components as described in the following diagram:
+These APIs take care of configuring a number of components as described in the following diagram:
 
 .Remote Chunking Configuration
 image::{batch-asciidoc}images/remote-chunking-config.png[Remote Chunking Configuration, scaledwidth="80%"]
@@ -1261,3 +1261,92 @@ You must also ensure that the partition `handler` attribute maps to the `partiti
 
 You can find a complete example of a remote partitioning job
 link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-partitioning-sample$$[here].
+
+The `@EnableBatchIntegration` annotation that can be used to simplify a remote
+ partitioning setup. This annotation provides two beans useful for remote partitioning:
+
+* `RemotePartitioningMasterStepBuilderFactory`: used to configure the master step
+* `RemotePartitioningWorkerStepBuilderFactory`: used to configure the worker step
+
+These APIs take care of configuring a number of components as described in the following diagram:
+
+.Remote Partitioning Configuration (with job repository polling)
+image::{batch-asciidoc}images/remote-partitioning-polling-config.png[Remote Partitioning Configuration (with job repository polling), scaledwidth="80%"]
+
+.Remote Partitioning Configuration (with replies aggregation)
+image::{batch-asciidoc}images/remote-partitioning-aggregation-config.png[Remote Partitioning Configuration (with replies aggregation), scaledwidth="80%"]
+
+On the master side, the `RemotePartitioningMasterStepBuilderFactory` allows you to
+configure a master step by declaring:
+
+* the `Partitioner` used to partition data
+* the output channel ("Outgoing requests") to send requests to workers
+* the input channel ("Incoming replies") to receive replies from workers (when configuring replies aggregation)
+* the poll interval and timeout parameters (when configuring job repository polling)
+
+The `MessageChannelPartitionHandler` and the `MessagingTemplate` are not needed to be explicitly configured
+(Those can still be explicitly configured if required).
+
+On the worker side, the `RemotePartitioningWorkerStepBuilderFactory` allows you to configure a worker to:
+
+* listen to requests sent by the master on the input channel ("Incoming requests")
+* call the `handle` method of `StepExecutionRequestHandler` for each request
+* send replies on the output channel ("Outgoing replies") to the master
+
+There is no need to explicitly configure the `StepExecutionRequestHandler` (which can be explicitly configured if required).
+
+The following example shows how to use these APIs:
+
+[source, java]
+----
+@Configuration
+@EnableBatchProcessing
+@EnableBatchIntegration
+public class RemotePartitioningJobConfiguration {
+
+    @Configuration
+    public static class MasterConfiguration {
+
+        @Autowired
+        private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+
+        @Bean
+        public Step masterStep() {
+                 return this.masterStepBuilderFactory
+                    .get("masterStep")
+                    .partitioner("workerStep", partitioner())
+                    .gridSize(10)
+                    .outputChannel(outgoingRequestsToWorkers())
+                    .inputChannel(incomingRepliesFromWorkers())
+                    .build();
+        }
+
+        // Middleware beans setup omitted
+
+    }
+
+    @Configuration
+    public static class WorkerConfiguration {
+
+        @Autowired
+        private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
+
+        @Bean
+        public Step workerStep() {
+                 return this.workerStepBuilderFactory
+                    .get("workerStep")
+                    .inputChannel(incomingRequestsFromMaster())
+                    .outputChannel(outgoingRepliesToMaster())
+                    .chunk(100)
+                    .reader(itemReader())
+                    .processor(itemProcessor())
+                    .writer(itemWriter())
+                    .build();
+        }
+
+        // Middleware beans setup omitted
+
+    }
+
+}
+----

spring-batch-docs/asciidoc/whatsnew.adoc

@@ -9,7 +9,7 @@
 The Spring Batch 4.1 release adds the following features:
 
 * A new `@SpringBatchTest` annotation to simplify testing batch components
-* A new `@EnableBatchIntegration` annotation to simplify remote chunking configuration
+* A new `@EnableBatchIntegration` annotation to simplify remote chunking and partitioning configuration
 * A new `JsonItemReader` and `JsonFileItemWriter` to support the JSON format
 * Add support for validating items with the Bean Validation API
 
@@ -125,6 +125,55 @@ that uses these new APIs in the
 link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-chunking-sample$$[samples module]
 as well as more details in the <<spring-batch-integration.adoc#remote-chunking,Spring Batch Integration>> chapter.
 
+Just like the remote chunking configuration simplification, this version also
+introduces new APIs to simplify a remote partitioning setup:
+`RemotePartitioningMasterStepBuilder` and `RemotePartitioningWorkerStepBuilder`.
+Those can be autowired in your configuration class if the
+`@EnableBatchIntegration` is present as shown in the following example:
+
+[source, java]
+----
+@Configuration
+@EnableBatchProcessing
+@EnableBatchIntegration
+public class RemotePartitioningAppConfig {
+
+   @Autowired
+   private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+
+   @Autowired
+   private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
+
+   @Bean
+   public Step masterStep() {
+            return this.masterStepBuilderFactory
+               .get("masterStep")
+               .partitioner("workerStep", partitioner())
+               .gridSize(10)
+               .outputChannel(outgoingRequestsToWorkers())
+               .inputChannel(incomingRepliesFromWorkers())
+               .build();
+   }
+
+   @Bean
+   public Step workerStep() {
+            return this.workerStepBuilderFactory
+               .get("workerStep")
+               .inputChannel(incomingRequestsFromMaster())
+               .outputChannel(outgoingRepliesToMaster())
+               .chunk(100)
+               .reader(itemReader())
+               .processor(itemProcessor())
+               .writer(itemWriter())
+               .build();
+   }
+
+   // Middleware beans setup omitted
+}
+----
+
+You can find more details about these new APIs in the <<spring-batch-integration.adoc#remote-partitioning,Spring Batch Integration>> chapter.
+
 [[whatsNewJson]]
 === JSON support
 

spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java

@@ -15,9 +15,12 @@
  */
 package org.springframework.batch.integration.config.annotation;
 
+import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
 import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
+import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -32,6 +35,8 @@ import org.springframework.transaction.PlatformTransactionManager;
 @Configuration
 public class BatchIntegrationConfiguration {
 
+	private JobExplorer jobExplorer;
+
 	private JobRepository jobRepository;
 
 	private PlatformTransactionManager transactionManager;
@@ -39,9 +44,11 @@ public class BatchIntegrationConfiguration {
 	@Autowired
 	public BatchIntegrationConfiguration(
 			JobRepository jobRepository,
+			JobExplorer jobExplorer,
 			PlatformTransactionManager transactionManager) {
 
 		this.jobRepository = jobRepository;
+		this.jobExplorer = jobExplorer;
 		this.transactionManager = transactionManager;
 	}
 
@@ -56,4 +63,16 @@ public class BatchIntegrationConfiguration {
 		return new RemoteChunkingWorkerBuilder<>();
 	}
 
+	@Bean
+	public RemotePartitioningMasterStepBuilderFactory remotePartitioningMasterStepBuilderFactory() {
+		return new RemotePartitioningMasterStepBuilderFactory(this.jobRepository,
+				this.jobExplorer, this.transactionManager);
+	}
+
+	@Bean
+	public RemotePartitioningWorkerStepBuilderFactory remotePartitioningWorkerStepBuilderFactory() {
+		return new RemotePartitioningWorkerStepBuilderFactory(this.jobRepository,
+				this.jobExplorer, this.transactionManager);
+	}
+
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java

@@ -23,24 +23,33 @@ import java.lang.annotation.Target;
 
 import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
 import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.context.annotation.Import;
+import org.springframework.integration.config.EnableIntegration;
 
 /**
  * Enable Spring Batch Integration features and provide a base configuration for
- * setting up remote chunking infrastructure beans.
+ * setting up remote chunking or partitioning infrastructure beans.
  *
  * By adding this annotation on a {@link org.springframework.context.annotation.Configuration}
  * class, it will be possible to autowire the following beans:
  *
  * <ul>
  *     <li>{@link RemoteChunkingMasterStepBuilderFactory}:
- *     used to create a master step by automatically setting up the job repository
- *     and transaction manager.</li>
+ *     used to create a master step of a remote chunking setup by automatically
+ *     setting the job repository and transaction manager.</li>
  *     <li>{@link RemoteChunkingWorkerBuilder}: used to create the integration
- *     flow on the worker side.</li>
+ *     flow on the worker side of a remote chunking setup.</li>
+ *     <li>{@link RemotePartitioningMasterStepBuilderFactory}: used to create
+ *     a master step of a remote partitioning setup by automatically setting
+ *     the job repository, job explorer, bean factory and transaction manager.</li>
+ *     <li>{@link RemotePartitioningWorkerStepBuilderFactory}: used to create
+ *     a worker step of a remote partitioning setup by automatically setting
+ *     the job repository, job explorer, bean factory and transaction manager.</li>
  * </ul>
  *
- * For example:
+ * For remote chunking, an example of a configuration class would be:
  *
  * <pre class="code">
  * &#064;Configuration
@@ -75,15 +84,60 @@ import org.springframework.context.annotation.Import;
  *       		.build();
  * 	}
  *
+ * 	// Middleware beans omitted
+ *
  * }
  * </pre>
  *
+ * For remote partitioning, an example of a configuration class would be:
+ *
+ * <pre class="code">
+ * &#064;Configuration
+ * &#064;EnableBatchIntegration
+ * &#064;EnableBatchProcessing
+ * public class RemotePartitioningAppConfig {
+ *
+ * 	&#064;Autowired
+ * 	private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+ *
+ * 	&#064;Autowired
+ * 	private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
+ *
+ * 	&#064;Bean
+ * 	public Step masterStep() {
+ *       	 return this.masterStepBuilderFactory
+ *       		.get("masterStep")
+ *       		.partitioner("workerStep", partitioner())
+ *       		.gridSize(10)
+ *       		.outputChannel(outgoingRequestsToWorkers())
+ *       		.inputChannel(incomingRepliesFromWorkers())
+ *       		.build();
+ * 	}
+ *
+ * 	&#064;Bean
+ * 	public Step workerStep() {
+ *       	 return this.workerStepBuilderFactory
+ *       		.get("workerStep")
+ *       		.inputChannel(incomingRequestsFromMaster())
+ *       		.outputChannel(outgoingRepliesToMaster())
+ *       		.chunk(100)
+ *       		.reader(itemReader())
+ *       		.processor(itemProcessor())
+ *       		.writer(itemWriter())
+ *       		.build();
+ * 	}
+ *
+ * 	// Middleware beans omitted
+ *
+ * }
+ * </pre>
  * @since 4.1
  * @author Mahmoud Ben Hassine
  */
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
+@EnableIntegration
 @Import(BatchIntegrationConfiguration.class)
 public @interface EnableBatchIntegration {
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java

@@ -0,0 +1,297 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.partition.PartitionHandler;
+import org.springframework.batch.core.partition.StepExecutionSplitter;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.core.partition.support.StepExecutionAggregator;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.PartitionStepBuilder;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.StandardIntegrationFlow;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+/**
+ * Builder for a master step in a remote partitioning setup. This builder creates and
+ * sets a {@link MessageChannelPartitionHandler} on the master step.
+ *
+ * <p>If no {@code messagingTemplate} is provided through
+ * {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)},
+ * this builder will create one. The {@code outputChannel} set with
+ * {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)} will be
+ * used as a default channel of the messaging template and will <strong>override any default
+ * channel already set on the messaging template</strong>.
+ *
+ * @since 4.1
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
+
+	private static final long DEFAULT_POLL_INTERVAL = 10000L;
+	private static final long DEFAULT_TIMEOUT = -1L;
+
+	private MessagingTemplate messagingTemplate;
+	private MessageChannel inputChannel;
+	private MessageChannel outputChannel;
+	private JobExplorer jobExplorer;
+	private BeanFactory beanFactory;
+	private long pollInterval = DEFAULT_POLL_INTERVAL;
+	private long timeout = DEFAULT_TIMEOUT;
+
+	/**
+	 * Create a new {@link RemotePartitioningMasterStepBuilder}.
+	 * @param stepName name of the master step
+	 */
+	public RemotePartitioningMasterStepBuilder(String stepName) {
+		super(new StepBuilder(stepName));
+	}
+
+	/**
+	 * Set the input channel on which replies from workers will be received.
+	 * @param inputChannel the input channel
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder inputChannel(MessageChannel inputChannel) {
+		Assert.notNull(inputChannel, "inputChannel must not be null");
+		this.inputChannel = inputChannel;
+		return this;
+	}
+
+	/**
+	 * Set the output channel on which requests to workers will be sent.
+	 * The output channel will be set as a default channel on the provided
+	 * {@link MessagingTemplate} trough
+	 * {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)}
+	 * (or the one created by this builder if no messaging template is provided).
+	 * @param outputChannel the output channel.
+	 * @return this builder instance for fluent chaining
+	 *
+	 * @see RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)
+	 */
+	public RemotePartitioningMasterStepBuilder outputChannel(MessageChannel outputChannel) {
+		Assert.notNull(outputChannel, "outputChannel must not be null");
+		this.outputChannel = outputChannel;
+		return this;
+	}
+
+	/**
+	 * Set the {@link MessagingTemplate} to use to send data to workers.
+	 * <p><strong>The default destination of the messaging template will be
+	 * overridden by the output channel provided through
+	 * {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}</strong>.</p>
+	 *
+	 * @param messagingTemplate the messaging template to use
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder messagingTemplate(MessagingTemplate messagingTemplate) {
+		Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
+		this.messagingTemplate = messagingTemplate;
+		return this;
+	}
+
+	/**
+	 * Set the job explorer.
+	 * @param jobExplorer the job explorer to use.
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder jobExplorer(JobExplorer jobExplorer) {
+		Assert.notNull(jobExplorer, "jobExplorer must not be null");
+		this.jobExplorer = jobExplorer;
+		return this;
+	}
+
+	/**
+	 * How often to poll the job repository for the status of the workers. Defaults to 10 seconds.
+	 * @param pollInterval the poll interval value in milliseconds
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder pollInterval(long pollInterval) {
+		Assert.isTrue(pollInterval > 0, "The poll interval must be greater than zero");
+		this.pollInterval = pollInterval;
+		return this;
+	}
+
+	/**
+	 * When using job repository polling, the time limit to wait. Defaults to -1 (no timeout).
+	 * @param timeout the timeout value in milliseconds
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder timeout(long timeout) {
+		this.timeout = timeout;
+		return this;
+	}
+
+	/**
+	 * Set the bean factory.
+	 * @param beanFactory the bean factory to use
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningMasterStepBuilder beanFactory(BeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
+		return this;
+	}
+
+	public Step build() {
+		Assert.notNull(this.outputChannel, "An outputChannel must be provided");
+
+		// configure messaging template
+		if (this.messagingTemplate == null) {
+			this.messagingTemplate = new MessagingTemplate();
+		}
+		this.messagingTemplate.setDefaultChannel(this.outputChannel);
+
+		// Configure the partition handler
+		final MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
+		partitionHandler.setStepName(getStepName());
+		partitionHandler.setGridSize(getGridSize());
+		partitionHandler.setMessagingOperations(this.messagingTemplate);
+
+		if (isPolling()) {
+			partitionHandler.setJobExplorer(this.jobExplorer);
+			partitionHandler.setPollInterval(this.pollInterval);
+			partitionHandler.setTimeout(this.timeout);
+		}
+		else {
+			PollableChannel replies = new QueueChannel();
+			partitionHandler.setReplyChannel(replies);
+			StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows
+					.from(this.inputChannel)
+					.aggregate(aggregatorSpec -> aggregatorSpec.processor(partitionHandler))
+					.channel(replies)
+					.get();
+			IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
+			integrationFlowContext.registration(standardIntegrationFlow)
+					.autoStartup(false)
+					.register();
+		}
+
+		try {
+			partitionHandler.afterPropertiesSet();
+			super.partitionHandler(partitionHandler);
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Unable to create a master step for remote partitioning", e);
+		}
+
+		return super.build();
+	}
+
+	private boolean isPolling() {
+		return this.inputChannel == null;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder repository(JobRepository jobRepository) {
+		super.repository(jobRepository);
+		return this;
+	}
+
+	@Override
+	public  RemotePartitioningMasterStepBuilder transactionManager(PlatformTransactionManager transactionManager) {
+		super.transactionManager(transactionManager);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder partitioner(String slaveStepName, Partitioner partitioner) {
+		super.partitioner(slaveStepName, partitioner);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder gridSize(int gridSize) {
+		super.gridSize(gridSize);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder step(Step step) {
+		super.step(step);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder splitter(StepExecutionSplitter splitter) {
+		super.splitter(splitter);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder aggregator(StepExecutionAggregator aggregator) {
+		super.aggregator(aggregator);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder startLimit(int startLimit) {
+		super.startLimit(startLimit);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder listener(Object listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder listener(StepExecutionListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningMasterStepBuilder allowStartIfComplete(boolean allowStartIfComplete) {
+		super.allowStartIfComplete(allowStartIfComplete);
+		return this;
+	}
+
+	/**
+	 * This method will throw a {@link UnsupportedOperationException} since
+	 * the partition handler of the master step will be automatically set to an
+	 * instance of {@link MessageChannelPartitionHandler}.
+	 *
+	 * When building a master step for remote partitioning using this builder,
+	 * no partition handler must be provided.
+	 *
+	 * @param partitionHandler a partition handler
+	 * @return this builder instance for fluent chaining
+	 * @throws UnsupportedOperationException if a partition handler is provided
+	 */
+	@Override
+	public RemotePartitioningMasterStepBuilder partitionHandler(PartitionHandler partitionHandler) throws UnsupportedOperationException {
+		throw new UnsupportedOperationException("When configuring a master step " +
+				"for remote partitioning using the RemotePartitioningMasterStepBuilder, " +
+				"the partition handler will be automatically set to an instance " +
+				"of MessageChannelPartitionHandler. The partition handler must " +
+				"not be provided in this case.");
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.transaction.PlatformTransactionManager;
+
+/**
+ * Convenient factory for a {@link RemotePartitioningMasterStepBuilder} which sets
+ * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
+ * {@link PlatformTransactionManager} automatically.
+ *
+ * @since 4.1
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningMasterStepBuilderFactory implements BeanFactoryAware {
+
+	private BeanFactory beanFactory;
+	final private JobExplorer jobExplorer;
+	final private JobRepository jobRepository;
+	final private PlatformTransactionManager transactionManager;
+
+
+	/**
+	 * Create a new {@link RemotePartitioningMasterStepBuilderFactory}.
+	 * @param jobRepository the job repository to use
+	 * @param jobExplorer the job explorer to use
+	 * @param transactionManager the transaction manager to use
+	 */
+	public RemotePartitioningMasterStepBuilderFactory(JobRepository jobRepository,
+		JobExplorer jobExplorer, PlatformTransactionManager transactionManager) {
+
+		this.jobRepository = jobRepository;
+		this.jobExplorer = jobExplorer;
+		this.transactionManager = transactionManager;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	/**
+	 * Creates a {@link RemotePartitioningMasterStepBuilder} and initializes its job
+	 * repository, job explorer, bean factory and transaction manager.
+	 * @param name the name of the step
+	 * @return a {@link RemotePartitioningMasterStepBuilder}
+	 */
+	public RemotePartitioningMasterStepBuilder get(String name) {
+		return new RemotePartitioningMasterStepBuilder(name)
+				.repository(this.jobRepository)
+				.jobExplorer(this.jobExplorer)
+				.beanFactory(this.beanFactory)
+				.transactionManager(this.transactionManager);
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java

@@ -0,0 +1,258 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.job.flow.Flow;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.StepLocator;
+import org.springframework.batch.core.step.builder.FlowStepBuilder;
+import org.springframework.batch.core.step.builder.JobStepBuilder;
+import org.springframework.batch.core.step.builder.PartitionStepBuilder;
+import org.springframework.batch.core.step.builder.SimpleStepBuilder;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.core.step.builder.TaskletStepBuilder;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.repeat.CompletionPolicy;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.StandardIntegrationFlow;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+/**
+ * Builder for a worker step in a remote partitioning setup. This builder
+ * creates an {@link IntegrationFlow} that:
+ *
+ * <ul>
+ *     <li>listens to {@link StepExecutionRequest}s coming from the master
+ *     on the input channel</li>
+ *     <li>invokes the {@link StepExecutionRequestHandler} to execute the worker
+ *     step for each incoming request. The worker step is located using the provided
+ *     {@link StepLocator}. If no {@link StepLocator} is provided, a {@link BeanFactoryStepLocator}
+ *     configured with the current {@link BeanFactory} will be used
+ *     <li>replies to the master on the output channel (when the master step is
+ *     configured to aggregate replies from workers). If no output channel
+ *     is provided, a {@link NullChannel} will be used (assuming the master side
+ *     is configured to poll the job repository for workers status)</li>
+ * </ul>
+ *
+ * @since 4.1
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
+
+	private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handle";
+	private static final Log logger = LogFactory.getLog(RemotePartitioningWorkerStepBuilder.class);
+
+	private MessageChannel inputChannel;
+	private MessageChannel outputChannel;
+	private JobExplorer jobExplorer;
+	private StepLocator stepLocator;
+	private BeanFactory beanFactory;
+
+	/**
+	 * Initialize a step builder for a step with the given name.
+	 * @param name the name of the step
+	 */
+	public RemotePartitioningWorkerStepBuilder(String name) {
+		super(name);
+	}
+
+	/**
+	 * Set the input channel on which step execution requests sent by the master
+	 * are received.
+	 * @param inputChannel the input channel
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningWorkerStepBuilder inputChannel(MessageChannel inputChannel) {
+		Assert.notNull(inputChannel, "inputChannel must not be null");
+		this.inputChannel = inputChannel;
+		return this;
+	}
+
+	/**
+	 * Set the output channel on which replies will be sent to the master step.
+	 * @param outputChannel the input channel
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningWorkerStepBuilder outputChannel(MessageChannel outputChannel) {
+		Assert.notNull(outputChannel, "outputChannel must not be null");
+		this.outputChannel = outputChannel;
+		return this;
+	}
+
+	/**
+	 * Set the job explorer.
+	 * @param jobExplorer the job explorer to use
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningWorkerStepBuilder jobExplorer(JobExplorer jobExplorer) {
+		Assert.notNull(jobExplorer, "jobExplorer must not be null");
+		this.jobExplorer = jobExplorer;
+		return this;
+	}
+
+	/**
+	 * Set the step locator used to locate the worker step to execute.
+	 * @param stepLocator the step locator to use
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningWorkerStepBuilder stepLocator(StepLocator stepLocator) {
+		Assert.notNull(stepLocator, "stepLocator must not be null");
+		this.stepLocator = stepLocator;
+		return this;
+	}
+
+	/**
+	 * Set the bean factory.
+	 * @param beanFactory the bean factory
+	 * @return this builder instance for fluent chaining
+	 */
+	public RemotePartitioningWorkerStepBuilder beanFactory(BeanFactory beanFactory) {
+		Assert.notNull(beanFactory, "beanFactory must not be null");
+		this.beanFactory = beanFactory;
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder repository(JobRepository jobRepository) {
+		super.repository(jobRepository);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder transactionManager(PlatformTransactionManager transactionManager) {
+		super.transactionManager(transactionManager);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder startLimit(int startLimit) {
+		super.startLimit(startLimit);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder listener(Object listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder listener(StepExecutionListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningWorkerStepBuilder allowStartIfComplete(boolean allowStartIfComplete) {
+		super.allowStartIfComplete(allowStartIfComplete);
+		return this;
+	}
+
+	@Override
+	public TaskletStepBuilder tasklet(Tasklet tasklet) {
+		configureWorkerIntegrationFlow();
+		return super.tasklet(tasklet);
+	}
+
+	@Override
+	public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
+		configureWorkerIntegrationFlow();
+		return super.chunk(chunkSize);
+	}
+
+	@Override
+	public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
+		configureWorkerIntegrationFlow();
+		return super.chunk(completionPolicy);
+	}
+
+	@Override
+	public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) {
+		configureWorkerIntegrationFlow();
+		return super.partitioner(stepName, partitioner);
+	}
+
+	@Override
+	public PartitionStepBuilder partitioner(Step step) {
+		configureWorkerIntegrationFlow();
+		return super.partitioner(step);
+	}
+
+	@Override
+	public JobStepBuilder job(Job job) {
+		configureWorkerIntegrationFlow();
+		return super.job(job);
+	}
+
+	@Override
+	public FlowStepBuilder flow(Flow flow) {
+		configureWorkerIntegrationFlow();
+		return super.flow(flow);
+	}
+
+	/**
+	 * Create an {@link IntegrationFlow} with a {@link StepExecutionRequestHandler}
+	 * configured as a service activator listening to the input channel and replying
+	 * on the output channel.
+	 */
+	private void configureWorkerIntegrationFlow() {
+		Assert.notNull(this.inputChannel, "An InputChannel must be provided");
+		Assert.notNull(this.jobExplorer, "A JobExplorer must be provided");
+
+		if (this.stepLocator == null) {
+			BeanFactoryStepLocator beanFactoryStepLocator = new BeanFactoryStepLocator();
+			beanFactoryStepLocator.setBeanFactory(this.beanFactory);
+			this.stepLocator = beanFactoryStepLocator;
+		}
+		if (this.outputChannel == null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("The output channel is set to a NullChannel. " +
+						"The master step must poll the job repository for workers status.");
+			}
+			this.outputChannel = new NullChannel();
+		}
+
+		StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
+		stepExecutionRequestHandler.setJobExplorer(this.jobExplorer);
+		stepExecutionRequestHandler.setStepLocator(this.stepLocator);
+
+		StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows
+				.from(this.inputChannel)
+				.handle(stepExecutionRequestHandler, SERVICE_ACTIVATOR_METHOD_NAME)
+				.channel(this.outputChannel)
+				.get();
+		IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
+		integrationFlowContext.registration(standardIntegrationFlow)
+				.autoStartup(false)
+				.register();
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.transaction.PlatformTransactionManager;
+
+/**
+ * Convenient factory for a {@link RemotePartitioningWorkerStepBuilder} which sets
+ * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
+ * {@link PlatformTransactionManager} automatically.
+ *
+ * @since 4.1
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAware {
+
+	private BeanFactory beanFactory;
+	final private JobExplorer jobExplorer;
+	final private JobRepository jobRepository;
+	final private PlatformTransactionManager transactionManager;
+
+
+	/**
+	 * Create a new {@link RemotePartitioningWorkerStepBuilderFactory}.
+	 * @param jobRepository the job repository to use
+	 * @param jobExplorer the job explorer to use
+	 * @param transactionManager the transaction manager to use
+	 */
+	public RemotePartitioningWorkerStepBuilderFactory(JobRepository jobRepository,
+		JobExplorer jobExplorer,
+		PlatformTransactionManager transactionManager) {
+
+		this.jobExplorer = jobExplorer;
+		this.jobRepository = jobRepository;
+		this.transactionManager = transactionManager;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	/**
+	 * Creates a {@link RemotePartitioningWorkerStepBuilder} and initializes its job
+	 * repository, job explorer, bean factory and transaction manager.
+	 * @param name the name of the step
+	 * @return a {@link RemotePartitioningWorkerStepBuilder}
+	 */
+	public RemotePartitioningWorkerStepBuilder get(String name) {
+		return new RemotePartitioningWorkerStepBuilder(name)
+				.repository(this.jobRepository)
+				.jobExplorer(this.jobExplorer)
+				.beanFactory(this.beanFactory)
+				.transactionManager(this.transactionManager);
+	}
+
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
+import org.springframework.batch.core.partition.PartitionHandler;
+import org.springframework.batch.core.partition.support.Partitioner;
+import org.springframework.batch.core.partition.support.StepExecutionAggregator;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.springframework.test.util.ReflectionTestUtils.getField;
+
+/**
+ * @author Mahmoud Ben Hassine
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration(classes = {RemotePartitioningMasterStepBuilderTests.BatchConfiguration.class})
+public class RemotePartitioningMasterStepBuilderTests {
+
+	@Rule
+	public ExpectedException expectedException = ExpectedException.none();
+
+	@Autowired
+	private JobRepository jobRepository;
+
+	@Test
+	public void inputChannelMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("inputChannel must not be null");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").inputChannel(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void outputChannelMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("outputChannel must not be null");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").outputChannel(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void messagingTemplateMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("messagingTemplate must not be null");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").messagingTemplate(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void jobExplorerMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("jobExplorer must not be null");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").jobExplorer(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void pollIntervalMustBeGreaterThanZero() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("The poll interval must be greater than zero");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").pollInterval(-1);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void testMandatoryOutputChannel() {
+		// given
+		RemotePartitioningMasterStepBuilder builder = new RemotePartitioningMasterStepBuilder("step");
+
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("An outputChannel must be provided");
+
+		// when
+		Step step = builder.build();
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void testUnsupportedOperationExceptionWhenSpecifyingPartitionHandler() {
+		// given
+		PartitionHandler partitionHandler = Mockito.mock(PartitionHandler.class);
+		this.expectedException.expect(UnsupportedOperationException.class);
+		this.expectedException.expectMessage("When configuring a master step " +
+				"for remote partitioning using the RemotePartitioningMasterStepBuilder, " +
+				"the partition handler will be automatically set to an instance " +
+				"of MessageChannelPartitionHandler. The partition handler must " +
+				"not be provided in this case.");
+
+		// when
+		new RemotePartitioningMasterStepBuilder("step").partitionHandler(partitionHandler);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void testMasterStepCreationWhenPollingRepository() {
+		// given
+		int gridSize = 5;
+		int startLimit = 3;
+		long timeout = 1000L;
+		long pollInterval = 5000L;
+		DirectChannel outputChannel = new DirectChannel();
+		Partitioner partitioner = Mockito.mock(Partitioner.class);
+		StepExecutionAggregator stepExecutionAggregator = (result, executions) -> { };
+
+		// when
+		Step step = new RemotePartitioningMasterStepBuilder("masterStep")
+				.repository(jobRepository)
+				.outputChannel(outputChannel)
+				.partitioner("workerStep", partitioner)
+				.gridSize(gridSize)
+				.pollInterval(pollInterval)
+				.timeout(timeout)
+				.startLimit(startLimit)
+				.aggregator(stepExecutionAggregator)
+				.allowStartIfComplete(true)
+				.build();
+
+		// then
+		Assert.assertNotNull(step);
+		Assert.assertEquals(getField(step, "startLimit"), startLimit);
+		Assert.assertEquals(getField(step, "jobRepository"), this.jobRepository);
+		Assert.assertEquals(getField(step, "stepExecutionAggregator"), stepExecutionAggregator);
+		Assert.assertTrue((Boolean) getField(step, "allowStartIfComplete"));
+
+		Object partitionHandler = getField(step, "partitionHandler");
+		Assert.assertNotNull(partitionHandler);
+		Assert.assertTrue(partitionHandler instanceof MessageChannelPartitionHandler);
+		MessageChannelPartitionHandler messageChannelPartitionHandler = (MessageChannelPartitionHandler) partitionHandler;
+		Assert.assertEquals(getField(messageChannelPartitionHandler, "gridSize"), gridSize);
+		Assert.assertEquals(getField(messageChannelPartitionHandler, "pollInterval"), pollInterval);
+		Assert.assertEquals(getField(messageChannelPartitionHandler, "timeout"), timeout);
+
+		Object messagingGateway = getField(messageChannelPartitionHandler, "messagingGateway");
+		Assert.assertNotNull(messagingGateway);
+		MessagingTemplate messagingTemplate = (MessagingTemplate) messagingGateway;
+		Assert.assertEquals(getField(messagingTemplate, "defaultDestination"), outputChannel);
+	}
+
+	@Test
+	public void testMasterStepCreationWhenAggregatingReplies() {
+		// given
+		int gridSize = 5;
+		int startLimit = 3;
+		DirectChannel outputChannel = new DirectChannel();
+		Partitioner partitioner = Mockito.mock(Partitioner.class);
+		StepExecutionAggregator stepExecutionAggregator = (result, executions) -> { };
+
+		// when
+		Step step = new RemotePartitioningMasterStepBuilder("masterStep")
+				.repository(jobRepository)
+				.outputChannel(outputChannel)
+				.partitioner("workerStep", partitioner)
+				.gridSize(gridSize)
+				.startLimit(startLimit)
+				.aggregator(stepExecutionAggregator)
+				.allowStartIfComplete(true)
+				.build();
+
+		// then
+		Assert.assertNotNull(step);
+		Assert.assertEquals(getField(step, "startLimit"), startLimit);
+		Assert.assertEquals(getField(step, "jobRepository"), this.jobRepository);
+		Assert.assertEquals(getField(step, "stepExecutionAggregator"), stepExecutionAggregator);
+		Assert.assertTrue((Boolean) getField(step, "allowStartIfComplete"));
+
+		Object partitionHandler = getField(step, "partitionHandler");
+		Assert.assertNotNull(partitionHandler);
+		Assert.assertTrue(partitionHandler instanceof MessageChannelPartitionHandler);
+		MessageChannelPartitionHandler messageChannelPartitionHandler = (MessageChannelPartitionHandler) partitionHandler;
+		Assert.assertEquals(getField(messageChannelPartitionHandler, "gridSize"), gridSize);
+
+		Object replyChannel = getField(messageChannelPartitionHandler, "replyChannel");
+		Assert.assertNotNull(replyChannel);
+		Assert.assertTrue(replyChannel instanceof QueueChannel);
+
+		Object messagingGateway = getField(messageChannelPartitionHandler, "messagingGateway");
+		Assert.assertNotNull(messagingGateway);
+		MessagingTemplate messagingTemplate = (MessagingTemplate) messagingGateway;
+		Assert.assertEquals(getField(messagingTemplate, "defaultDestination"), outputChannel);
+	}
+
+	@Configuration
+	@EnableBatchProcessing
+	public static class BatchConfiguration {
+
+	}
+}

spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderTests.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.integration.partition;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.Mock;
+
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.integration.channel.DirectChannel;
+
+/**
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningWorkerStepBuilderTests {
+
+	@Rule
+	public ExpectedException expectedException = ExpectedException.none();
+
+	@Mock
+	private Tasklet tasklet;
+
+	@Test
+	public void inputChannelMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("inputChannel must not be null");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").inputChannel(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void outputChannelMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("outputChannel must not be null");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").outputChannel(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void jobExplorerMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("jobExplorer must not be null");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").jobExplorer(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void stepLocatorMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("stepLocator must not be null");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").stepLocator(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void beanFactoryMustNotBeNull() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("beanFactory must not be null");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").beanFactory(null);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void testMandatoryInputChannel() {
+		// given
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("An InputChannel must be provided");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step").tasklet(this.tasklet);
+
+		// then
+		// expected exception
+	}
+
+	@Test
+	public void testMandatoryJobExplorer() {
+		// given
+		DirectChannel inputChannel = new DirectChannel();
+		this.expectedException.expect(IllegalArgumentException.class);
+		this.expectedException.expectMessage("A JobExplorer must be provided");
+
+		// when
+		new RemotePartitioningWorkerStepBuilder("step")
+				.inputChannel(inputChannel)
+				.tasklet(this.tasklet);
+
+		// then
+		// expected exception
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java

@@ -21,22 +21,15 @@ import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
-import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
-import org.springframework.batch.core.partition.PartitionHandler;
-import org.springframework.batch.core.partition.support.Partitioner;
-import org.springframework.batch.integration.partition.MessageChannelPartitionHandler;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
 import org.springframework.batch.sample.remotepartitioning.BasicPartitioner;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.config.AggregatorFactoryBean;
-import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.jms.dsl.Jms;
@@ -49,24 +42,22 @@ import org.springframework.integration.jms.dsl.Jms;
  */
 @Configuration
 @EnableBatchProcessing
-@EnableIntegration
+@EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
 public class MasterConfiguration {
 
 	private static final int GRID_SIZE = 3;
 
-	private static final long RECEIVE_TIMEOUT = 600000L;
-
 	private final JobBuilderFactory jobBuilderFactory;
 
-	private final StepBuilderFactory stepBuilderFactory;
+	private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
 
 
 	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
-								StepBuilderFactory stepBuilderFactory) {
+							   RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) {
 
 		this.jobBuilderFactory = jobBuilderFactory;
-		this.stepBuilderFactory = stepBuilderFactory;
+		this.masterStepBuilderFactory = masterStepBuilderFactory;
 	}
 
 	/*
@@ -89,64 +80,31 @@ public class MasterConfiguration {
 	 * Configure inbound flow (replies coming from workers)
 	 */
 	@Bean
-	public QueueChannel replies() {
-		return new QueueChannel();
-	}
-
-	@Bean
-	public DirectChannel inboundStaging() {
+	public DirectChannel replies() {
 		return new DirectChannel();
 	}
 
 	@Bean
-	public IntegrationFlow inboundStagingFlow(ActiveMQConnectionFactory connectionFactory) {
+	public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
 		return IntegrationFlows
 				.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies"))
-				.channel(inboundStaging())
+				.channel(replies())
 				.get();
 	}
 
 	/*
-	 * Configure master step components
+	 * Configure the master step
 	 */
 	@Bean
 	public Step masterStep() {
-		return this.stepBuilderFactory.get("masterStep")
-				.partitioner("slaveStep", partitioner())
-				.partitionHandler(partitionHandler())
+		return this.masterStepBuilderFactory.get("masterStep")
+				.partitioner("workerStep", new BasicPartitioner())
 				.gridSize(GRID_SIZE)
+				.outputChannel(requests())
+				.inputChannel(replies())
 				.build();
 	}
 
-	@Bean
-	public Partitioner partitioner() {
-		return new BasicPartitioner();
-	}
-
-	@Bean
-	public PartitionHandler partitionHandler() {
-		MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
-		partitionHandler.setStepName("slaveStep");
-		partitionHandler.setGridSize(GRID_SIZE);
-		partitionHandler.setReplyChannel(replies());
-
-		MessagingTemplate template = new MessagingTemplate();
-		template.setDefaultChannel(requests());
-		template.setReceiveTimeout(RECEIVE_TIMEOUT);
-		partitionHandler.setMessagingOperations(template);
-
-		return partitionHandler;
-	}
-
-	@Bean
-	@ServiceActivator(inputChannel = "inboundStaging")
-	public AggregatorFactoryBean partitioningMessageHandler() {
-		AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
-		aggregatorFactoryBean.setProcessorBean(partitionHandler());
-		aggregatorFactoryBean.setOutputChannel(replies());
-		return aggregatorFactoryBean;
-	}
-
 	@Bean
 	public Job remotePartitioningJob() {
 		return this.jobBuilderFactory.get("remotePartitioningJob")

spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/WorkerConfiguration.java

@@ -19,23 +19,18 @@ import org.apache.activemq.ActiveMQConnectionFactory;
 
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
-import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
 import org.springframework.batch.core.configuration.annotation.StepScope;
-import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.step.tasklet.Tasklet;
-import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
-import org.springframework.batch.integration.partition.StepExecutionRequestHandler;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.jms.dsl.Jms;
@@ -48,24 +43,15 @@ import org.springframework.integration.jms.dsl.Jms;
  */
 @Configuration
 @EnableBatchProcessing
-@EnableIntegration
+@EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
 public class WorkerConfiguration {
 
-	private final StepBuilderFactory stepBuilderFactory;
-
-	private final ApplicationContext applicationContext;
-
-	private final JobExplorer jobExplorer;
+	private final RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
 
 
-	public WorkerConfiguration(StepBuilderFactory stepBuilderFactory,
-								JobExplorer jobExplorer,
-								ApplicationContext applicationContext) {
-
-		this.stepBuilderFactory = stepBuilderFactory;
-		this.applicationContext = applicationContext;
-		this.jobExplorer = jobExplorer;
+	public WorkerConfiguration(RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory) {
+		this.workerStepBuilderFactory = workerStepBuilderFactory;
 	}
 
 	/*
@@ -101,29 +87,20 @@ public class WorkerConfiguration {
 	}
 
 	/*
-	 * Configure worker components
+	 * Configure the worker step
 	 */
 	@Bean
-	@ServiceActivator(inputChannel = "requests", outputChannel = "replies")
-	public StepExecutionRequestHandler stepExecutionRequestHandler() {
-		StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
-		stepExecutionRequestHandler.setJobExplorer(this.jobExplorer);
-		BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator();
-		stepLocator.setBeanFactory(this.applicationContext);
-		stepExecutionRequestHandler.setStepLocator(stepLocator);
-		return stepExecutionRequestHandler;
-	}
-
-	@Bean
-	public Step slaveStep() {
-		return this.stepBuilderFactory.get("slaveStep")
-				.tasklet(getTasklet(null))
+	public Step workerStep() {
+		return this.workerStepBuilderFactory.get("workerStep")
+				.inputChannel(requests())
+				.outputChannel(replies())
+				.tasklet(tasklet(null))
 				.build();
 	}
 
 	@Bean
 	@StepScope
-	public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
+	public Tasklet tasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
 		return (contribution, chunkContext) -> {
 			System.out.println("processing " + partition);
 			return RepeatStatus.FINISHED;

spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java

@@ -15,18 +15,14 @@
  */
 package org.springframework.batch.sample.remotepartitioning.polling;
 
-import javax.sql.DataSource;
-
 import org.apache.activemq.ActiveMQConnectionFactory;
 
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
-import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
-import org.springframework.batch.core.partition.PartitionHandler;
-import org.springframework.batch.core.partition.support.Partitioner;
-import org.springframework.batch.integration.partition.MessageChannelPartitionHandler;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
 import org.springframework.batch.sample.remotepartitioning.BasicPartitioner;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
@@ -34,8 +30,6 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.jms.dsl.Jms;
@@ -48,26 +42,22 @@ import org.springframework.integration.jms.dsl.Jms;
  */
 @Configuration
 @EnableBatchProcessing
-@EnableIntegration
+@EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
 public class MasterConfiguration {
 
 	private static final int GRID_SIZE = 3;
 
-	private static final long RECEIVE_TIMEOUT = 600000L;
-
-	private static final long POLL_INTERVAL = 2000L;
-
 	private final JobBuilderFactory jobBuilderFactory;
 
-	private final StepBuilderFactory stepBuilderFactory;
+	private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
 
 
 	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
-								StepBuilderFactory stepBuilderFactory) {
+							   RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) {
 
 		this.jobBuilderFactory = jobBuilderFactory;
-		this.stepBuilderFactory = stepBuilderFactory;
+		this.masterStepBuilderFactory = masterStepBuilderFactory;
 	}
 
 	/*
@@ -87,38 +77,17 @@ public class MasterConfiguration {
 	}
 
 	/*
-	 * Configure master step components
+	 * Configure the master step
 	 */
 	@Bean
 	public Step masterStep() {
-		return this.stepBuilderFactory.get("masterStep")
-				.partitioner("slaveStep", partitioner())
-				.partitionHandler(partitionHandler(null))
+		return this.masterStepBuilderFactory.get("masterStep")
+				.partitioner("workerStep", new BasicPartitioner())
 				.gridSize(GRID_SIZE)
+				.outputChannel(requests())
 				.build();
 	}
 
-	@Bean
-	public Partitioner partitioner() {
-		return new BasicPartitioner();
-	}
-
-	@Bean
-	public PartitionHandler partitionHandler(DataSource dataSource) {
-		MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
-		partitionHandler.setStepName("slaveStep");
-		partitionHandler.setGridSize(GRID_SIZE);
-		partitionHandler.setDataSource(dataSource);
-		partitionHandler.setPollInterval(POLL_INTERVAL);
-
-		MessagingTemplate template = new MessagingTemplate();
-		template.setDefaultChannel(requests());
-		template.setReceiveTimeout(RECEIVE_TIMEOUT);
-		partitionHandler.setMessagingOperations(template);
-
-		return partitionHandler;
-	}
-
 	@Bean
 	public Job remotePartitioningJob() {
 		return this.jobBuilderFactory.get("remotePartitioningJob")

spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java

@@ -19,24 +19,18 @@ import org.apache.activemq.ActiveMQConnectionFactory;
 
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
-import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
 import org.springframework.batch.core.configuration.annotation.StepScope;
-import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.step.tasklet.Tasklet;
-import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
-import org.springframework.batch.integration.partition.StepExecutionRequestHandler;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.NullChannel;
-import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.jms.dsl.Jms;
@@ -49,24 +43,15 @@ import org.springframework.integration.jms.dsl.Jms;
  */
 @Configuration
 @EnableBatchProcessing
-@EnableIntegration
+@EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
 public class WorkerConfiguration {
 
-	private final StepBuilderFactory stepBuilderFactory;
-
-	private final ApplicationContext applicationContext;
-
-	private final JobExplorer jobExplorer;
+	private final RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
 
 
-	public WorkerConfiguration(StepBuilderFactory stepBuilderFactory,
-								JobExplorer jobExplorer,
-								ApplicationContext applicationContext) {
-
-		this.stepBuilderFactory = stepBuilderFactory;
-		this.applicationContext = applicationContext;
-		this.jobExplorer = jobExplorer;
+	public WorkerConfiguration(RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory) {
+		this.workerStepBuilderFactory = workerStepBuilderFactory;
 	}
 
 	/*
@@ -86,37 +71,19 @@ public class WorkerConfiguration {
 	}
 
 	/*
-	 * Configure outbound flow (replies going to the master)
+	 * Configure the worker step
 	 */
 	@Bean
-	public NullChannel replies() {
-		return new NullChannel(); // replies are discarded (since the master is polling the job repository)
-	}
-
-	/*
-	 * Configure worker components
-	 */
-	@Bean
-	@ServiceActivator(inputChannel = "requests", outputChannel = "replies")
-	public StepExecutionRequestHandler stepExecutionRequestHandler() {
-		StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
-		stepExecutionRequestHandler.setJobExplorer(this.jobExplorer);
-		BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator();
-		stepLocator.setBeanFactory(this.applicationContext);
-		stepExecutionRequestHandler.setStepLocator(stepLocator);
-		return stepExecutionRequestHandler;
-	}
-
-	@Bean
-	public Step slaveStep() {
-		return this.stepBuilderFactory.get("slaveStep")
-				.tasklet(getTasklet(null))
+	public Step workerStep() {
+		return this.workerStepBuilderFactory.get("workerStep")
+				.inputChannel(requests())
+				.tasklet(tasklet(null))
 				.build();
 	}
 
 	@Bean
 	@StepScope
-	public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
+	public Tasklet tasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
 		return (contribution, chunkContext) -> {
 			System.out.println("processing " + partition);
 			return RepeatStatus.FINISHED;