Add remote partitioning samples

This commit also fixes the code examples in the remote partitioning
section of the documentation

Resolves BATCH-2730

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;
+	}
+
+}

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;
+	}
+
+}

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;
+	}
+
+}

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();
+	}
+
+}

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;
+		};
+	}
+
+}

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();
+	}
+
+}

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;
+		};
+	}
+
+}

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=

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();
+	}
+
+}

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;
+	}
+
+}

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;
+	}
+
+}