Add remote partitioning samples

This commit also fixes the code examples in the remote partitioning section of the documentation Resolves BATCH-2730
2018-06-08 06:41:04 +02:00
parent 34cbcc2294
commit 2c5b7b59e8
13 changed files with 928 additions and 49 deletions
--- a/spring-batch-docs/asciidoc/spring-batch-integration.adoc
+++ b/spring-batch-docs/asciidoc/spring-batch-integration.adoc
@@ -1051,6 +1051,10 @@ configuration:
 .Java Configuration
 [source, java, role="javaContent"]
 ----
+
+/*
+ * Configuration of the master side
+ */
@Bean
 public PartitionHandler partitionHandler() {
    MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
@@ -1064,6 +1068,11 @@ public PartitionHandler partitionHandler() {
    return partitionHandler;
 }

+@Bean
+public QueueChannel outboundReplies() {
+    return new QueueChannel();
+}
+
@Bean
 public DirectChannel outboundRequests() {
    return new DirectChannel();
@@ -1078,44 +1087,13 @@ public IntegrationFlow outboundJmsRequests() {
 }

@Bean
-public DirectChannel inboundRequests() {
-    return new DirectChannel();
-}
-
-public IntegrationFlow inboundJmsRequests() {
-    return IntegrationFlows
-            .from(Jms.messageDrivenChannelAdapter(connectionFactory())
-                    .configureListenerContainer(c -> c.subscriptionDurable(false))
-                    .destination("requestsQueue"))
-            .channel(inboundRequests())
-            .get();
-	}
-
-@Bean
-public StepExecutionRequestHandler stepExecutionRequestHandler() {
-    StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
-    stepExecutionRequestHandler.setJobExplorer(jobExplorer);
-    stepExecutionRequestHandler.setStepLocator(stepLocator());
-    return stepExecutionRequestHandler;
-}
-
-@Bean
-@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging")
-public StepExecutionRequestHandler serviceActivator() throws Exception {
-    return stepExecutionRequestHandler();
-}
-
-@Bean
-public DirectChannel outboundStaging() {
-    return new DirectChannel();
-}
-
-@Bean
-public IntegrationFlow outboundJmsStaging() {
-    return IntegrationFlows.from("outboundStaging")
-            .handle(Jms.outboundGateway(connectionFactory())
-                    .requestDestination("stagingQueue"))
-            .get();
+@ServiceActivator(inputChannel = "inboundStaging")
+public AggregatorFactoryBean partitioningMessageHandler() throws Exception {
+    AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
+    aggregatorFactoryBean.setProcessorBean(partitionHandler());
+    aggregatorFactoryBean.setOutputChannel(outboundReplies());
+    // configure other propeties of the aggregatorFactoryBean
+    return aggregatorFactoryBean;
 }

@Bean
@@ -1133,19 +1111,48 @@ public IntegrationFlow inboundJmsStaging() {
            .get();
 }

+/*
+ * Configuration of the worker side
+ */
@Bean
-@ServiceActivator(inputChannel = "inboundStaging")
-public AggregatorFactoryBean partitioningMessageHandler() throws Exception {
-    AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
-    aggregatorFactoryBean.setProcessorBean(partitionHandler());
-    aggregatorFactoryBean.setOutputChannel(outboundReplies());
-    ...
-    return aggregatorFactoryBean;
+public StepExecutionRequestHandler stepExecutionRequestHandler() {
+    StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
+    stepExecutionRequestHandler.setJobExplorer(jobExplorer);
+    stepExecutionRequestHandler.setStepLocator(stepLocator());
+    return stepExecutionRequestHandler;
 }

@Bean
-public QueueChannel outboundReplies() {
-    return new QueueChannel();
+@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging")
+public StepExecutionRequestHandler serviceActivator() throws Exception {
+    return stepExecutionRequestHandler();
+}
+
+@Bean
+public DirectChannel inboundRequests() {
+    return new DirectChannel();
+}
+
+public IntegrationFlow inboundJmsRequests() {
+    return IntegrationFlows
+            .from(Jms.messageDrivenChannelAdapter(connectionFactory())
+                    .configureListenerContainer(c -> c.subscriptionDurable(false))
+                    .destination("requestsQueue"))
+            .channel(inboundRequests())
+            .get();
+}
+
+@Bean
+public DirectChannel outboundStaging() {
+    return new DirectChannel();
+}
+
+@Bean
+public IntegrationFlow outboundJmsStaging() {
+    return IntegrationFlows.from("outboundStaging")
+            .handle(Jms.outboundGateway(connectionFactory())
+                    .requestDestination("stagingQueue"))
+            .get();
 }
 ----

@@ -1168,8 +1175,12 @@ You must also ensure that the partition `handler` attribute maps to the `partiti
 	public Job personJob() {
 		return jobBuilderFactory.get("personJob")
 				.start(stepBuilderFactory.get("step1.master")
-						.partitioner(partitionHandler())
+						.partitioner("step1.worker", partitioner())
+						.partitionHandler(partitionHandler())
 						.build())
 				.build();
 	}
 ----
+
+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].
--- a/spring-batch-samples/README.md
+++ b/spring-batch-samples/README.md
@@ -638,13 +638,27 @@ happens inside the main step transaction).
 The purpose of this sample is to show multi-threaded step execution
 using the `PartitionHandler` SPI.  The example uses a
 `TaskExecutorPartitionHandler` to spread the work of reading
-some files acrosss multiple threads, with one `Step` execution
+some files across multiple threads, with one `Step` execution
 per thread.  The key components are the `PartitionStep` and the
 `MultiResourcePartitioner` which is responsible for dividing up
 the work.  Notice that the readers and writers in the `Step`
 that is being partitioned are step-scoped, so that their state does
 not get shared across threads of execution.

+### [Remote Partitioning Sample](id:remotePartitioning)
+
+This sample shows how to configure a remote partitioning job. The master step
+uses a `MessageChannelPartitionHandler` to send partitions to and receive 
+replies from workers. Two examples are shown:
+
+* A master step that polls the job repository to see if all workers have finished
+their work
+* A master step that aggregates replies from workers to notify work completion
+
+The sample uses an embedded JMS broker and an embedded database for simplicity
+but any option supported via Spring Integration for communication is technically
+acceptable. 
+
 ### [Remote Chunking Sample](id:remoteChunking)

 This sample shows how to configure a remote chunking job. The master step will
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java
@@ -0,0 +1,43 @@
+/*
+ * 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.sample.remotepartitioning;
+
+import java.util.Map;
+
+import org.springframework.batch.core.partition.support.SimplePartitioner;
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * Simple partitioner for demonstration purpose.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+public class BasicPartitioner extends SimplePartitioner {
+
+	private static final String PARTITION_KEY = "partition";
+
+	@Override
+	public Map<String, ExecutionContext> partition(int gridSize) {
+		Map<String, ExecutionContext> partitions = super.partition(gridSize);
+		int i = 0;
+		for (ExecutionContext context : partitions.values()) {
+			context.put(PARTITION_KEY, PARTITION_KEY + (i++));
+		}
+		return partitions;
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java
@@ -0,0 +1,44 @@
+/*
+ * 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.sample.remotepartitioning;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+
+/**
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@PropertySource("classpath:remote-partitioning.properties")
+public class BrokerConfiguration {
+
+	@Value("${broker.url}")
+	private String brokerUrl;
+
+	@Bean
+	public ActiveMQConnectionFactory connectionFactory() {
+		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
+		connectionFactory.setBrokerURL(this.brokerUrl);
+		connectionFactory.setTrustAllPackages(true);
+		return connectionFactory;
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java
@@ -0,0 +1,53 @@
+/*
+ * 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.sample.remotepartitioning;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.dbcp2.BasicDataSource;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+
+/**
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@PropertySource("classpath:remote-partitioning.properties")
+public class DataSourceConfiguration {
+
+	@Value("${datasource.url}")
+	private String url;
+
+	@Value("${datasource.username}")
+	private String username;
+
+	@Value("${datasource.password}")
+	private String password;
+
+	@Bean
+	public DataSource dataSource() {
+		BasicDataSource dataSource = new BasicDataSource();
+		dataSource.setUrl(url);
+		dataSource.setUsername(username);
+		dataSource.setPassword(password);
+		return dataSource;
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java
@@ -0,0 +1,157 @@
+/*
+ * 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.sample.remotepartitioning.aggregating;
+
+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.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;
+
+/**
+ * This configuration class is for the master side of the remote partitioning sample.
+ * The master step will create 3 partitions for workers to process.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@EnableBatchProcessing
+@EnableIntegration
+@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;
+
+
+	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
+								StepBuilderFactory stepBuilderFactory) {
+
+		this.jobBuilderFactory = jobBuilderFactory;
+		this.stepBuilderFactory = stepBuilderFactory;
+	}
+
+	/*
+	 * Configure outbound flow (requests going to workers)
+	 */
+	@Bean
+	public DirectChannel requests() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(requests())
+				.handle(Jms.outboundAdapter(connectionFactory).destination("requests"))
+				.get();
+	}
+
+	/*
+	 * Configure inbound flow (replies coming from workers)
+	 */
+	@Bean
+	public QueueChannel replies() {
+		return new QueueChannel();
+	}
+
+	@Bean
+	public DirectChannel inboundStaging() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow inboundStagingFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies"))
+				.channel(inboundStaging())
+				.get();
+	}
+
+	/*
+	 * Configure master step components
+	 */
+	@Bean
+	public Step masterStep() {
+		return this.stepBuilderFactory.get("masterStep")
+				.partitioner("slaveStep", partitioner())
+				.partitionHandler(partitionHandler())
+				.gridSize(GRID_SIZE)
+				.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")
+				.start(masterStep())
+				.build();
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/WorkerConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/WorkerConfiguration.java
@@ -0,0 +1,133 @@
+/*
+ * 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.sample.remotepartitioning.aggregating;
+
+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.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;
+
+/**
+ * This configuration class is for the worker side of the remote partitioning sample.
+ * Each worker will process a partition sent by the master step.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@EnableBatchProcessing
+@EnableIntegration
+@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
+public class WorkerConfiguration {
+
+	private final StepBuilderFactory stepBuilderFactory;
+
+	private final ApplicationContext applicationContext;
+
+	private final JobExplorer jobExplorer;
+
+
+	public WorkerConfiguration(StepBuilderFactory stepBuilderFactory,
+								JobExplorer jobExplorer,
+								ApplicationContext applicationContext) {
+
+		this.stepBuilderFactory = stepBuilderFactory;
+		this.applicationContext = applicationContext;
+		this.jobExplorer = jobExplorer;
+	}
+
+	/*
+	 * Configure inbound flow (requests coming from the master)
+	 */
+	@Bean
+	public DirectChannel requests() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests"))
+				.channel(requests())
+				.get();
+	}
+
+	/*
+	 * Configure outbound flow (replies going to the master)
+	 */
+	@Bean
+	public DirectChannel replies() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(replies())
+				.handle(Jms.outboundAdapter(connectionFactory).destination("replies"))
+				.get();
+	}
+
+	/*
+	 * 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))
+				.build();
+	}
+
+	@Bean
+	@StepScope
+	public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
+		return (contribution, chunkContext) -> {
+			System.out.println("processing " + partition);
+			return RepeatStatus.FINISHED;
+		};
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java
@@ -0,0 +1,129 @@
+/*
+ * 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.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.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.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;
+
+/**
+ * This configuration class is for the master side of the remote partitioning sample.
+ * The master step will create 3 partitions for workers to process.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@EnableBatchProcessing
+@EnableIntegration
+@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;
+
+
+	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
+								StepBuilderFactory stepBuilderFactory) {
+
+		this.jobBuilderFactory = jobBuilderFactory;
+		this.stepBuilderFactory = stepBuilderFactory;
+	}
+
+	/*
+	 * Configure outbound flow (requests going to workers)
+	 */
+	@Bean
+	public DirectChannel requests() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(requests())
+				.handle(Jms.outboundAdapter(connectionFactory).destination("requests"))
+				.get();
+	}
+
+	/*
+	 * Configure master step components
+	 */
+	@Bean
+	public Step masterStep() {
+		return this.stepBuilderFactory.get("masterStep")
+				.partitioner("slaveStep", partitioner())
+				.partitionHandler(partitionHandler(null))
+				.gridSize(GRID_SIZE)
+				.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")
+				.start(masterStep())
+				.build();
+	}
+
+}
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java
@@ -0,0 +1,126 @@
+/*
+ * 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.sample.remotepartitioning.polling;
+
+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.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;
+
+/**
+ * This configuration class is for the worker side of the remote partitioning sample.
+ * Each worker will process a partition sent by the master step.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@Configuration
+@EnableBatchProcessing
+@EnableIntegration
+@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
+public class WorkerConfiguration {
+
+	private final StepBuilderFactory stepBuilderFactory;
+
+	private final ApplicationContext applicationContext;
+
+	private final JobExplorer jobExplorer;
+
+
+	public WorkerConfiguration(StepBuilderFactory stepBuilderFactory,
+								JobExplorer jobExplorer,
+								ApplicationContext applicationContext) {
+
+		this.stepBuilderFactory = stepBuilderFactory;
+		this.applicationContext = applicationContext;
+		this.jobExplorer = jobExplorer;
+	}
+
+	/*
+	 * Configure inbound flow (requests coming from the master)
+	 */
+	@Bean
+	public DirectChannel requests() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
+		return IntegrationFlows
+				.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests"))
+				.channel(requests())
+				.get();
+	}
+
+	/*
+	 * Configure outbound flow (replies going to the master)
+	 */
+	@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))
+				.build();
+	}
+
+	@Bean
+	@StepScope
+	public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) {
+		return (contribution, chunkContext) -> {
+			System.out.println("processing " + partition);
+			return RepeatStatus.FINISHED;
+		};
+	}
+
+}
--- a/spring-batch-samples/src/main/resources/remote-partitioning.properties
+++ b/spring-batch-samples/src/main/resources/remote-partitioning.properties
@@ -0,0 +1,4 @@
+broker.url=tcp://localhost:61617
+datasource.url=jdbc:hsqldb:mem:testdb
+datasource.username=sa
+datasource.password=
--- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java
+++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java
@@ -0,0 +1,91 @@
+/*
+ * 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.sample;
+
+import org.apache.activemq.broker.BrokerService;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Base class for remote partitioning tests.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@RunWith(SpringRunner.class)
+@PropertySource("classpath:remote-partitioning.properties")
+public abstract class RemotePartitioningJobFunctionalTests {
+
+	private static final String BROKER_DATA_DIRECTORY = "build/activemq-data";
+
+	@Value("${broker.url}")
+	private String brokerUrl;
+
+	@Autowired
+	protected JobLauncherTestUtils jobLauncherTestUtils;
+
+	private BrokerService brokerService;
+
+	private EmbeddedDatabase embeddedDatabase;
+
+	private AnnotationConfigApplicationContext workerApplicationContext;
+
+	protected abstract Class<?> getWorkerConfigurationClass();
+
+	@Before
+	public void setUp() throws Exception {
+		this.brokerService = new BrokerService();
+		this.brokerService.addConnector(this.brokerUrl);
+		this.brokerService.setDataDirectory(BROKER_DATA_DIRECTORY);
+		this.brokerService.start();
+		this.embeddedDatabase = new EmbeddedDatabaseBuilder()
+				.addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql")
+				.addScript("/org/springframework/batch/core/schema-hsqldb.sql")
+				.build();
+		this.workerApplicationContext = new AnnotationConfigApplicationContext(getWorkerConfigurationClass());
+	}
+
+	@Test
+	public void testRemotePartitioningJob() throws Exception {
+		// when
+		JobExecution jobExecution = this.jobLauncherTestUtils.launchJob();
+
+		// then
+		Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
+		Assert.assertEquals(4, jobExecution.getStepExecutions().size()); // master + 3 workers
+	}
+
+	@After
+	public void tearDown() throws Exception {
+		this.workerApplicationContext.close();
+		this.brokerService.stop();
+		this.embeddedDatabase.shutdown();
+	}
+
+}
--- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithMessageAggregationFunctionalTests.java
+++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithMessageAggregationFunctionalTests.java
@@ -0,0 +1,37 @@
+/*
+ * 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.sample;
+
+import org.springframework.batch.sample.config.JobRunnerConfiguration;
+import org.springframework.batch.sample.remotepartitioning.aggregating.MasterConfiguration;
+import org.springframework.batch.sample.remotepartitioning.aggregating.WorkerConfiguration;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * The master step of the job under test will create 3 partitions for workers
+ * to process.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
+public class RemotePartitioningJobWithMessageAggregationFunctionalTests extends RemotePartitioningJobFunctionalTests {
+
+	@Override
+	protected Class<WorkerConfiguration> getWorkerConfigurationClass() {
+		return WorkerConfiguration.class;
+	}
+
+}
--- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithRepositoryPollingFunctionalTests.java
+++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithRepositoryPollingFunctionalTests.java
@@ -0,0 +1,37 @@
+/*
+ * 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.sample;
+
+import org.springframework.batch.sample.config.JobRunnerConfiguration;
+import org.springframework.batch.sample.remotepartitioning.polling.MasterConfiguration;
+import org.springframework.batch.sample.remotepartitioning.polling.WorkerConfiguration;
+import org.springframework.test.context.ContextConfiguration;
+
+/**
+ * The master step of the job under test will create 3 partitions for workers
+ * to process.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
+public class RemotePartitioningJobWithRepositoryPollingFunctionalTests extends RemotePartitioningJobFunctionalTests {
+
+	@Override
+	protected Class<WorkerConfiguration> getWorkerConfigurationClass() {
+		return WorkerConfiguration.class;
+	}
+
+}