diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilder.java
new file mode 100644
index 000000000..c3c35e9dd
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilder.java
@@ -0,0 +1,420 @@
+/*
+ * Copyright 2019 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
+ *
+ *       https://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.chunk;
+
+import org.springframework.batch.core.ChunkListener;
+import org.springframework.batch.core.ItemReadListener;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.core.step.item.KeyGenerator;
+import org.springframework.batch.core.step.skip.SkipPolicy;
+import org.springframework.batch.core.step.tasklet.TaskletStep;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.repeat.CompletionPolicy;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.repeat.exception.ExceptionHandler;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.retry.RetryPolicy;
+import org.springframework.retry.backoff.BackOffPolicy;
+import org.springframework.retry.policy.RetryContextCache;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttribute;
+import org.springframework.util.Assert;
+
+/**
+ * Builder for a manager step in a remote chunking setup. This builder creates and
+ * sets a {@link ChunkMessageChannelItemWriter} on the manager step.
+ *
+ * <p>If no {@code messagingTemplate} is provided through
+ * {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)},
+ * this builder will create one and set its default channel to the {@code outputChannel}
+ * provided through {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)}.</p>
+ *
+ * <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
+ * and that its default channel is set to an output channel on which requests to workers
+ * will be sent.</p>
+ *
+ * @param <I> type of input items
+ * @param <O> type of output items
+ *
+ * @since 4.2
+ * @author Mahmoud Ben Hassine
+ */
+public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
+
+	private MessagingTemplate messagingTemplate;
+	private PollableChannel inputChannel;
+	private MessageChannel outputChannel;
+
+	private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
+	private static final long DEFAULT_THROTTLE_LIMIT = 6;
+	private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
+	private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
+
+	/**
+	 * Create a new {@link RemoteChunkingManagerStepBuilder}.
+	 *
+	 * @param stepName name of the manager step
+	 */
+	public RemoteChunkingManagerStepBuilder(String stepName) {
+		super(new StepBuilder(stepName));
+	}
+
+	/**
+	 * Set the input channel on which replies from workers will be received.
+	 * The provided input channel will be set as a reply channel on the
+	 * {@link ChunkMessageChannelItemWriter} created by this builder.
+	 *
+	 * @param inputChannel the input channel
+	 * @return this builder instance for fluent chaining
+	 *
+	 * @see ChunkMessageChannelItemWriter#setReplyChannel
+	 */
+	public RemoteChunkingManagerStepBuilder<I, O> inputChannel(PollableChannel 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. By using
+	 * this setter, a default messaging template will be created and the output
+	 * channel will be set as its default channel.
+	 * <p>Use either this setter or {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)}
+	 * to provide a fully configured messaging template.</p>
+	 *
+	 * @param outputChannel the output channel.
+	 * @return this builder instance for fluent chaining
+	 *
+	 * @see RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)
+	 */
+	public RemoteChunkingManagerStepBuilder<I, O> 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.
+	 * <strong>The default channel of the messaging template must be set</strong>.
+	 * <p>Use either this setter to provide a fully configured messaging template or
+	 * provide an output channel through {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)}
+	 * and a default messaging template will be created.</p>
+	 *
+	 * @param messagingTemplate the messaging template to use
+	 * @return this builder instance for fluent chaining
+	 * @see RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)
+	 */
+	public RemoteChunkingManagerStepBuilder<I, O> messagingTemplate(MessagingTemplate messagingTemplate) {
+		Assert.notNull(messagingTemplate, "messagingTemplate must not be null");
+		this.messagingTemplate = messagingTemplate;
+		return this;
+	}
+
+	/**
+	 * The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
+	 * multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
+	 * slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
+	 *
+	 * @param maxWaitTimeouts the maximum number of wait timeouts
+	 * @return this builder instance for fluent chaining
+	 * @see ChunkMessageChannelItemWriter#setMaxWaitTimeouts(int)
+	 */
+	public RemoteChunkingManagerStepBuilder<I, O> maxWaitTimeouts(int maxWaitTimeouts) {
+		Assert.isTrue(maxWaitTimeouts > 0, "maxWaitTimeouts must be greater than zero");
+		this.maxWaitTimeouts = maxWaitTimeouts;
+		return this;
+	}
+
+	/**
+	 * Public setter for the throttle limit. This limits the number of pending requests for chunk processing to avoid
+	 * overwhelming the receivers.
+	 *
+	 * @param throttleLimit the throttle limit to set
+	 * @return this builder instance for fluent chaining
+	 * @see ChunkMessageChannelItemWriter#setThrottleLimit(long)
+	 */
+	public RemoteChunkingManagerStepBuilder<I, O> throttleLimit(long throttleLimit) {
+		Assert.isTrue(throttleLimit > 0, "throttleLimit must be greater than zero");
+		this.throttleLimit = throttleLimit;
+		return this;
+	}
+
+	/**
+	 * Build a manager {@link TaskletStep}.
+	 *
+	 * @return the configured manager step
+	 * @see RemoteChunkHandlerFactoryBean
+	 */
+	public TaskletStep build() {
+		Assert.notNull(this.inputChannel, "An InputChannel must be provided");
+		Assert.state(this.outputChannel == null || this.messagingTemplate == null,
+				"You must specify either an outputChannel or a messagingTemplate but not both.");
+
+		// configure messaging template
+		if (this.messagingTemplate == null) {
+			this.messagingTemplate = new MessagingTemplate();
+			this.messagingTemplate.setDefaultChannel(this.outputChannel);
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("No messagingTemplate was provided, using a default one");
+			}
+		}
+
+		// configure item writer
+		ChunkMessageChannelItemWriter<O> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();
+		chunkMessageChannelItemWriter.setMessagingOperations(this.messagingTemplate);
+		chunkMessageChannelItemWriter.setMaxWaitTimeouts(this.maxWaitTimeouts);
+		chunkMessageChannelItemWriter.setThrottleLimit(this.throttleLimit);
+		chunkMessageChannelItemWriter.setReplyChannel(this.inputChannel);
+		super.writer(chunkMessageChannelItemWriter);
+
+		return super.build();
+	}
+
+	/*
+	 * The following methods override those from parent builders and return
+	 * the current builder type.
+	 * FIXME: Change parent builders to be generic and return current builder
+	  * type in each method.
+	 */
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> reader(ItemReader<? extends I> reader) {
+		super.reader(reader);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> repository(JobRepository jobRepository) {
+		super.repository(jobRepository);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> transactionManager(PlatformTransactionManager transactionManager) {
+		super.transactionManager(transactionManager);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(Object listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(SkipListener<? super I, ? super O> listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(ChunkListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> transactionAttribute(TransactionAttribute transactionAttribute) {
+		super.transactionAttribute(transactionAttribute);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(org.springframework.retry.RetryListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> keyGenerator(KeyGenerator keyGenerator) {
+		super.keyGenerator(keyGenerator);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> retryLimit(int retryLimit) {
+		super.retryLimit(retryLimit);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> retryPolicy(RetryPolicy retryPolicy) {
+		super.retryPolicy(retryPolicy);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> backOffPolicy(BackOffPolicy backOffPolicy) {
+		super.backOffPolicy(backOffPolicy);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> retryContextCache(RetryContextCache retryContextCache) {
+		super.retryContextCache(retryContextCache);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> skipLimit(int skipLimit) {
+		super.skipLimit(skipLimit);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> noSkip(Class<? extends Throwable> type) {
+		super.noSkip(type);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> skip(Class<? extends Throwable> type) {
+		super.skip(type);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> skipPolicy(SkipPolicy skipPolicy) {
+		super.skipPolicy(skipPolicy);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> noRollback(Class<? extends Throwable> type) {
+		super.noRollback(type);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> noRetry(Class<? extends Throwable> type) {
+		super.noRetry(type);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> retry(Class<? extends Throwable> type) {
+		super.retry(type);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> stream(ItemStream stream) {
+		super.stream(stream);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> chunk(int chunkSize) {
+		super.chunk(chunkSize);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
+		super.chunk(completionPolicy);
+		return this;
+	}
+
+	/**
+	 * This method will throw a {@link UnsupportedOperationException} since
+	 * the item writer of the manager step in a remote chunking setup will be
+	 * automatically set to an instance of {@link ChunkMessageChannelItemWriter}.
+	 *
+	 * When building a manager step for remote chunking, no item writer must be
+	 * provided.
+	 *
+	 * @throws UnsupportedOperationException if an item writer is provided
+	 * @see ChunkMessageChannelItemWriter
+	 * @see RemoteChunkHandlerFactoryBean#setChunkWriter(ItemWriter)
+	 */
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> writer(ItemWriter<? super O> writer) throws UnsupportedOperationException {
+		throw new UnsupportedOperationException("When configuring a manager step " +
+				"for remote chunking, the item writer will be automatically set " +
+				"to an instance of ChunkMessageChannelItemWriter. The item writer " +
+				"must not be provided in this case.");
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> readerIsTransactionalQueue() {
+		super.readerIsTransactionalQueue();
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(ItemReadListener<? super I> listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(ItemWriteListener<? super O> listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> chunkOperations(RepeatOperations repeatTemplate) {
+		super.chunkOperations(repeatTemplate);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> exceptionHandler(ExceptionHandler exceptionHandler) {
+		super.exceptionHandler(exceptionHandler);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> stepOperations(RepeatOperations repeatTemplate) {
+		super.stepOperations(repeatTemplate);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> startLimit(int startLimit) {
+		super.startLimit(startLimit);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> listener(StepExecutionListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> allowStartIfComplete(boolean allowStartIfComplete) {
+		super.allowStartIfComplete(allowStartIfComplete);
+		return this;
+	}
+
+	@Override
+	public RemoteChunkingManagerStepBuilder<I, O> processor(ItemProcessor<? super I, ? extends O> itemProcessor) {
+		super.processor(itemProcessor);
+		return this;
+	}
+}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderFactory.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderFactory.java
new file mode 100644
index 000000000..f83149512
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderFactory.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2019 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
+ *
+ *       https://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.chunk;
+
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.transaction.PlatformTransactionManager;
+
+/**
+ * Convenient factory for a {@link RemoteChunkingManagerStepBuilder} which sets
+ * the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
+ *
+ * @since 4.2
+ * @author Mahmoud Ben Hassine
+ */
+public class RemoteChunkingManagerStepBuilderFactory {
+
+	private JobRepository jobRepository;
+
+	private PlatformTransactionManager transactionManager;
+
+	/**
+	 * Create a new {@link RemoteChunkingManagerStepBuilderFactory}.
+	 *
+	 * @param jobRepository the job repository to use
+	 * @param transactionManager the transaction manager to use
+	 */
+	public RemoteChunkingManagerStepBuilderFactory(
+			JobRepository jobRepository,
+			PlatformTransactionManager transactionManager) {
+
+		this.jobRepository = jobRepository;
+		this.transactionManager = transactionManager;
+	}
+
+	/**
+	 * Creates a {@link RemoteChunkingManagerStepBuilder} and initializes its job
+	 * repository and transaction manager.
+	 * 
+	 * @param name the name of the step
+	 * @param <I> type of input items
+	 * @param <O> type of output items
+	 * @return a {@link RemoteChunkingManagerStepBuilder}
+	 */
+	public <I, O> RemoteChunkingManagerStepBuilder<I, O> get(String name) {
+		return new RemoteChunkingManagerStepBuilder<I, O>(name)
+				.repository(this.jobRepository)
+				.transactionManager(this.transactionManager);
+	}
+
+}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java
index a8c41dd17..8d88f0ea3 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -59,9 +59,12 @@ import org.springframework.util.Assert;
  * @param <I> type of input items
  * @param <O> type of output items
  *
+ * @deprecated Use {@link RemoteChunkingManagerStepBuilder} instead.
+ *
  * @since 4.1
  * @author Mahmoud Ben Hassine
  */
+@Deprecated
 public class RemoteChunkingMasterStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
 
 	private MessagingTemplate messagingTemplate;
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderFactory.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderFactory.java
index 410b1e269..44405a216 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderFactory.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -22,9 +22,12 @@ import org.springframework.transaction.PlatformTransactionManager;
  * Convenient factory for a {@link RemoteChunkingMasterStepBuilder} which sets
  * the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
  *
+ * @deprecated Use {@link RemoteChunkingManagerStepBuilderFactory} instead.
+ *
  * @since 4.1
  * @author Mahmoud Ben Hassine
  */
+@Deprecated
 public class RemoteChunkingMasterStepBuilderFactory {
 
 	private JobRepository jobRepository;
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java
index e5fa2e898..d377da942 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -18,8 +18,10 @@ 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.RemoteChunkingManagerStepBuilderFactory;
 import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder;
 import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -52,23 +54,37 @@ public class BatchIntegrationConfiguration {
 		this.transactionManager = transactionManager;
 	}
 
+	@Deprecated
 	@Bean
 	public RemoteChunkingMasterStepBuilderFactory remoteChunkingMasterStepBuilderFactory() {
 		return new RemoteChunkingMasterStepBuilderFactory(this.jobRepository,
 				this.transactionManager);
 	}
 
+	@Bean
+	public RemoteChunkingManagerStepBuilderFactory remoteChunkingManagerStepBuilderFactory() {
+		return new RemoteChunkingManagerStepBuilderFactory(this.jobRepository,
+				this.transactionManager);
+	}
+
 	@Bean
 	public <I,O> RemoteChunkingWorkerBuilder<I, O> remoteChunkingWorkerBuilder() {
 		return new RemoteChunkingWorkerBuilder<>();
 	}
 
+	@Deprecated
 	@Bean
 	public RemotePartitioningMasterStepBuilderFactory remotePartitioningMasterStepBuilderFactory() {
 		return new RemotePartitioningMasterStepBuilderFactory(this.jobRepository,
 				this.jobExplorer, this.transactionManager);
 	}
 
+	@Bean
+	public RemotePartitioningManagerStepBuilderFactory remotePartitioningManagerStepBuilderFactory() {
+		return new RemotePartitioningManagerStepBuilderFactory(this.jobRepository,
+				this.jobExplorer, this.transactionManager);
+	}
+
 	@Bean
 	public RemotePartitioningWorkerStepBuilderFactory remotePartitioningWorkerStepBuilderFactory() {
 		return new RemotePartitioningWorkerStepBuilderFactory(this.jobRepository,
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java
index f9020bd8e..8d5897ec4 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -22,8 +22,8 @@ import java.lang.annotation.RetentionPolicy;
 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.chunk.RemoteChunkingManagerStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.config.EnableIntegration;
@@ -36,13 +36,13 @@ import org.springframework.integration.config.EnableIntegration;
  * class, it will be possible to autowire the following beans:
  *
  * <ul>
- *     <li>{@link RemoteChunkingMasterStepBuilderFactory}:
- *     used to create a master step of a remote chunking setup by automatically
+ *     <li>{@link RemoteChunkingManagerStepBuilderFactory}:
+ *     used to create a manager 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 of a remote chunking setup.</li>
- *     <li>{@link RemotePartitioningMasterStepBuilderFactory}: used to create
- *     a master step of a remote partitioning setup by automatically setting
+ *     <li>{@link RemotePartitioningManagerStepBuilderFactory}: used to create
+ *     a manager 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
@@ -58,15 +58,15 @@ import org.springframework.integration.config.EnableIntegration;
  * public class RemoteChunkingAppConfig {
  *
  * 	&#064;Autowired
- * 	private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
+ * 	private RemoteChunkingManagerStepBuilderFactory managerStepBuilderFactory;
  *
  * 	&#064;Autowired
  * 	private RemoteChunkingWorkerBuilder workerBuilder;
  *
  * 	&#064;Bean
- * 	public TaskletStep masterStep() {
- *       	 return this.masterStepBuilderFactory
- *       		.get("masterStep")
+ * 	public TaskletStep managerStep() {
+ *       	 return this.managerStepBuilderFactory
+ *       		.get("managerStep")
  *       		.chunk(100)
  *       		.reader(itemReader())
  *       		.outputChannel(outgoingRequestsToWorkers())
@@ -79,8 +79,8 @@ import org.springframework.integration.config.EnableIntegration;
  *       	 return this.workerBuilder
  *       		.itemProcessor(itemProcessor())
  *       		.itemWriter(itemWriter())
- *       		.inputChannel(incomingRequestsFromMaster())
- *       		.outputChannel(outgoingRepliesToMaster())
+ *       		.inputChannel(incomingRequestsFromManager())
+ *       		.outputChannel(outgoingRepliesToManager())
  *       		.build();
  * 	}
  *
@@ -98,15 +98,15 @@ import org.springframework.integration.config.EnableIntegration;
  * public class RemotePartitioningAppConfig {
  *
  * 	&#064;Autowired
- * 	private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+ * 	private RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory;
  *
  * 	&#064;Autowired
  * 	private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
  *
  * 	&#064;Bean
- * 	public Step masterStep() {
- *       	 return this.masterStepBuilderFactory
- *       		.get("masterStep")
+ * 	public Step managerStep() {
+ *       	 return this.managerStepBuilderFactory
+ *       		.get("managerStep")
  *       		.partitioner("workerStep", partitioner())
  *       		.gridSize(10)
  *       		.outputChannel(outgoingRequestsToWorkers())
@@ -118,8 +118,8 @@ import org.springframework.integration.config.EnableIntegration;
  * 	public Step workerStep() {
  *       	 return this.workerStepBuilderFactory
  *       		.get("workerStep")
- *       		.inputChannel(incomingRequestsFromMaster())
- *       		.outputChannel(outgoingRepliesToMaster())
+ *       		.inputChannel(incomingRequestsFromManager())
+ *       		.outputChannel(outgoingRepliesToManager())
  *       		.chunk(100)
  *       		.reader(itemReader())
  *       		.processor(itemProcessor())
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java
index 7dd5b0b1b..e870b693c 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -22,6 +22,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
  *
  * @author Gunnar Hillert
  * @author Chris Schaefer
+ * @author Mahmoud Ben Hassine
  * @since 1.3
  */
 public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -30,7 +31,13 @@ public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespa
 	 */
 	public void init() {
 		this.registerBeanDefinitionParser("job-launching-gateway",  new JobLaunchingGatewayParser());
-		this.registerBeanDefinitionParser("remote-chunking-master", new RemoteChunkingMasterParser());
-		this.registerBeanDefinitionParser("remote-chunking-slave", new RemoteChunkingSlaveParser());
+		RemoteChunkingManagerParser remoteChunkingManagerParser = new RemoteChunkingManagerParser();
+		this.registerBeanDefinitionParser("remote-chunking-manager", remoteChunkingManagerParser);
+		RemoteChunkingWorkerParser remoteChunkingWorkerParser = new RemoteChunkingWorkerParser();
+		this.registerBeanDefinitionParser("remote-chunking-worker", remoteChunkingWorkerParser);
+
+		// TODO remove the following when related deprecated APIs are removed
+		this.registerBeanDefinitionParser("remote-chunking-master", remoteChunkingManagerParser);
+		this.registerBeanDefinitionParser("remote-chunking-slave", remoteChunkingWorkerParser);
 	}
 }
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParser.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParser.java
similarity index 93%
rename from spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParser.java
rename to spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParser.java
index 438f0fd2a..0959fa4e0 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParser.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -28,13 +28,14 @@ import org.w3c.dom.Element;
 
 /**
  * <p>
- * Parser for the remote-chunking-master namespace element.
+ * Parser for the remote-chunking-manager namespace element.
  * </p>
  *
  * @author Chris Schaefer
+ * @author Mahmoud Ben Hassine
  * @since 3.1
  */
-public class RemoteChunkingMasterParser extends AbstractBeanDefinitionParser {
+public class RemoteChunkingManagerParser extends AbstractBeanDefinitionParser {
 	private static final String MESSAGE_TEMPLATE_ATTRIBUTE = "message-template";
 	private static final String STEP_ATTRIBUTE = "step";
 	private static final String REPLY_CHANNEL_ATTRIBUTE = "reply-channel";
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParser.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParser.java
similarity index 95%
rename from spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParser.java
rename to spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParser.java
index 55560fa96..f0052f4ad 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParser.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -34,16 +34,17 @@ import org.springframework.util.StringUtils;
 
 /**
  * <p>
- * Parser for the remote-chunking-slave namespace element. If an
+ * Parser for the remote-chunking-worker namespace element. If an
  * {@link org.springframework.batch.item.ItemProcessor} is not provided, an
  * {@link org.springframework.batch.item.support.PassThroughItemProcessor} will be
  * configured.
  * </p>
  *
  * @author Chris Schaefer
+ * @author Mahmoud Ben Hassine
  * @since 3.1
  */
-public class RemoteChunkingSlaveParser extends AbstractBeanDefinitionParser {
+public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
 	private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
 	private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
 	private static final String ITEM_PROCESSOR_ATTRIBUTE = "item-processor";
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilder.java
new file mode 100644
index 000000000..9151f4f55
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilder.java
@@ -0,0 +1,305 @@
+/*
+ * Copyright 2019 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
+ *
+ *       https://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 manager step in a remote partitioning setup. This builder creates and
+ * sets a {@link MessageChannelPartitionHandler} on the manager step.
+ *
+ * <p>If no {@code messagingTemplate} is provided through
+ * {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)},
+ * this builder will create one and set its default channel to the {@code outputChannel}
+ * provided through {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)}.</p>
+ *
+ * <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
+ * and that its default channel is set to an output channel on which requests to workers
+ * will be sent.</p>
+ *
+ * @since 4.2
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningManagerStepBuilder 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 RemotePartitioningManagerStepBuilder}.
+	 * @param stepName name of the manager step
+	 */
+	public RemotePartitioningManagerStepBuilder(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 RemotePartitioningManagerStepBuilder 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. By using
+	 * this setter, a default messaging template will be created and the output
+	 * channel will be set as its default channel.
+	 * <p>Use either this setter or {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)}
+	 * to provide a fully configured messaging template.</p>
+	 *
+	 * @param outputChannel the output channel.
+	 * @return this builder instance for fluent chaining
+	 * @see RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)
+	 */
+	public RemotePartitioningManagerStepBuilder 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.
+	 * <strong>The default channel of the messaging template must be set</strong>.
+	 * <p>Use either this setter to provide a fully configured messaging template or
+	 * provide an output channel through {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)}
+	 * and a default messaging template will be created.</p>
+	 *
+	 * @param messagingTemplate the messaging template to use
+	 * @return this builder instance for fluent chaining
+	 * @see RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)
+	 */
+	public RemotePartitioningManagerStepBuilder 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 RemotePartitioningManagerStepBuilder 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 RemotePartitioningManagerStepBuilder 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 RemotePartitioningManagerStepBuilder 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 RemotePartitioningManagerStepBuilder beanFactory(BeanFactory beanFactory) {
+		this.beanFactory = beanFactory;
+		return this;
+	}
+
+	public Step build() {
+		Assert.state(this.outputChannel == null || this.messagingTemplate == null,
+				"You must specify either an outputChannel or a messagingTemplate but not both.");
+
+		// configure messaging template
+		if (this.messagingTemplate == null) {
+			this.messagingTemplate = new MessagingTemplate();
+			this.messagingTemplate.setDefaultChannel(this.outputChannel);
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("No messagingTemplate was provided, using a default one");
+			}
+		}
+
+		// 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 manager step for remote partitioning", e);
+		}
+
+		return super.build();
+	}
+
+	private boolean isPolling() {
+		return this.inputChannel == null;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder repository(JobRepository jobRepository) {
+		super.repository(jobRepository);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder transactionManager(PlatformTransactionManager transactionManager) {
+		super.transactionManager(transactionManager);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder partitioner(String slaveStepName, Partitioner partitioner) {
+		super.partitioner(slaveStepName, partitioner);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder gridSize(int gridSize) {
+		super.gridSize(gridSize);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder step(Step step) {
+		super.step(step);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder splitter(StepExecutionSplitter splitter) {
+		super.splitter(splitter);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder aggregator(StepExecutionAggregator aggregator) {
+		super.aggregator(aggregator);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder startLimit(int startLimit) {
+		super.startLimit(startLimit);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder listener(Object listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder listener(StepExecutionListener listener) {
+		super.listener(listener);
+		return this;
+	}
+
+	@Override
+	public RemotePartitioningManagerStepBuilder allowStartIfComplete(boolean allowStartIfComplete) {
+		super.allowStartIfComplete(allowStartIfComplete);
+		return this;
+	}
+
+	/**
+	 * This method will throw a {@link UnsupportedOperationException} since
+	 * the partition handler of the manager step will be automatically set to an
+	 * instance of {@link MessageChannelPartitionHandler}.
+	 *
+	 * When building a manager 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 RemotePartitioningManagerStepBuilder partitionHandler(PartitionHandler partitionHandler) throws UnsupportedOperationException {
+		throw new UnsupportedOperationException("When configuring a manager step " +
+				"for remote partitioning using the RemotePartitioningManagerStepBuilder, " +
+				"the partition handler will be automatically set to an instance " +
+				"of MessageChannelPartitionHandler. The partition handler must " +
+				"not be provided in this case.");
+	}
+
+}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderFactory.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderFactory.java
new file mode 100644
index 000000000..7697eeca4
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderFactory.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2019 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
+ *
+ *       https://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 RemotePartitioningManagerStepBuilder} which sets
+ * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
+ * {@link PlatformTransactionManager} automatically.
+ *
+ * @since 4.2
+ * @author Mahmoud Ben Hassine
+ */
+public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryAware {
+
+	private BeanFactory beanFactory;
+	final private JobExplorer jobExplorer;
+	final private JobRepository jobRepository;
+	final private PlatformTransactionManager transactionManager;
+
+
+	/**
+	 * Create a new {@link RemotePartitioningManagerStepBuilderFactory}.
+	 * @param jobRepository the job repository to use
+	 * @param jobExplorer the job explorer to use
+	 * @param transactionManager the transaction manager to use
+	 */
+	public RemotePartitioningManagerStepBuilderFactory(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 RemotePartitioningManagerStepBuilder} and initializes its job
+	 * repository, job explorer, bean factory and transaction manager.
+	 * @param name the name of the step
+	 * @return a {@link RemotePartitioningManagerStepBuilder}
+	 */
+	public RemotePartitioningManagerStepBuilder get(String name) {
+		return new RemotePartitioningManagerStepBuilder(name)
+				.repository(this.jobRepository)
+				.jobExplorer(this.jobExplorer)
+				.beanFactory(this.beanFactory)
+				.transactionManager(this.transactionManager);
+	}
+
+}
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java
index 8bf3076b3..ae3397b5d 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -51,9 +51,12 @@ import org.springframework.util.Assert;
  * and that its default channel is set to an output channel on which requests to workers
  * will be sent.</p>
  *
+ * @deprecated Use {@link RemotePartitioningManagerStepBuilder} instead.
+ *
  * @since 4.1
  * @author Mahmoud Ben Hassine
  */
+@Deprecated
 public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder {
 
 	private static final long DEFAULT_POLL_INTERVAL = 10000L;
diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java
index 40022a1cc..7d8249c82 100644
--- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -28,9 +28,12 @@ import org.springframework.transaction.PlatformTransactionManager;
  * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
  * {@link PlatformTransactionManager} automatically.
  *
+ * @deprecated Use {@link RemotePartitioningManagerStepBuilderFactory} instead
+ *
  * @since 4.1
  * @author Mahmoud Ben Hassine
  */
+@Deprecated
 public class RemotePartitioningMasterStepBuilderFactory implements BeanFactoryAware {
 
 	private BeanFactory beanFactory;
diff --git a/spring-batch-integration/src/main/resources/META-INF/spring.schemas b/spring-batch-integration/src/main/resources/META-INF/spring.schemas
index 5f1da865f..3c3012614 100644
--- a/spring-batch-integration/src/main/resources/META-INF/spring.schemas
+++ b/spring-batch-integration/src/main/resources/META-INF/spring.schemas
@@ -1,3 +1,4 @@
 http\://www.springframework.org/schema/batch-integration/spring-batch-integration-1.3.xsd=org/springframework/batch/integration/config/xml/spring-batch-integration-1.3.xsd
 http\://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd=org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd
-http\://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd=org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd
+http\://www.springframework.org/schema/batch-integration/spring-batch-integration-4.2.xsd=org/springframework/batch/integration/config/xml/spring-batch-integration-4.2.xsd
+http\://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd=org/springframework/batch/integration/config/xml/spring-batch-integration-4.2.xsd
diff --git a/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd b/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd
index 6d5e061d0..f1474b8a5 100644
--- a/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd
+++ b/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-3.1.xsd
@@ -137,6 +137,8 @@
 			<xsd:documentation><![CDATA[
 						The remote chunking master is used as the writer in batch jobs
 						to communicate with the middleware used to send chunks remotely.
+						This element is deprecated in favor of "remote-chunking-manager"
+						defined in "spring-batch-integration-4.2.xsd" and above.
 					]]></xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
@@ -186,6 +188,8 @@
 						The remote chunking slave receives chunks sent by the master and
 						provides the processor and writer to handle the chunks, returning
 						status back to the master.
+						This element is deprecated in favor of "remote-chunking-worker"
+						defined in "spring-batch-integration-4.2.xsd" and above.
 					]]></xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
diff --git a/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-4.2.xsd b/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-4.2.xsd
new file mode 100644
index 000000000..0794e1129
--- /dev/null
+++ b/spring-batch-integration/src/main/resources/org/springframework/batch/integration/config/xml/spring-batch-integration-4.2.xsd
@@ -0,0 +1,244 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/batch-integration"
+			xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+			xmlns:tool="http://www.springframework.org/schema/tool"
+			xmlns:integration="http://www.springframework.org/schema/integration"
+			targetNamespace="http://www.springframework.org/schema/batch-integration"
+			elementFormDefault="qualified" attributeFormDefault="unqualified">
+
+	<xsd:import namespace="http://www.springframework.org/schema/beans"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"/>
+	<xsd:import namespace="http://www.springframework.org/schema/integration"
+				schemaLocation="https://www.springframework.org/schema/integration/spring-integration.xsd"/>
+
+	<xsd:annotation>
+		<xsd:documentation><![CDATA[
+			Defines the configuration elements for the Spring Batch Integration
+			Support.
+		]]></xsd:documentation>
+	</xsd:annotation>
+
+	<xsd:element name="job-launching-gateway">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+						This Outbound Gateway is used to launch Batch Jobs. The
+						payload of Messages to be processed MUST be an instance
+						of JobLaunchRequest.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+			<xsd:attributeGroup ref="corespringBatchIntegrationComponentAttributes"/>
+			<xsd:attribute name="request-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The input Message Channel of this endpoint.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								Message Channel to which the resulting JobExecution
+								payload will be sent.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-timeout" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								Allows you to specify how long this gateway will wait for
+								the reply message to be sent successfully to the reply channel
+								before throwing an exception. This attribute only applies when the
+								channel might block, for example when using a bounded queue channel that
+								is currently full.
+
+								Also, keep in mind that when sending to a DirectChannel, the
+								invocation will occur in the sender's thread. Therefore,
+								the failing of the send operation may be caused by other
+								components further downstream.
+
+								The "reply-timeout" attribute maps to the "sendTimeout" property of the
+								underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
+
+								The attribute will default, if not specified, to '-1', meaning that
+								by default, the Gateway will wait indefinitely. The value is
+								specified in milliseconds.
+							]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="job-launcher" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<xsd:documentation><![CDATA[
+								Pass in a custom JobLauncher bean reference.
+								This attribute is optional. If not specified the
+								adapter will re-use the default instance (under
+								the id 'jobLauncher', e.g. when using the
+								@EnableBatchProcessing annotation via JavaConfig).
+								If no default instance exists an exception is
+								thrown.
+							]]></xsd:documentation>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.batch.core.launch.JobLauncher"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="order">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+							Specifies the order for invocation when this endpoint
+							is connected as a subscriber to a SubscribableChannel.
+						]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:attributeGroup name="corespringBatchIntegrationComponentAttributes">
+		<xsd:attribute name="id" type="xsd:string" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Identifies the underlying Spring bean definition, which is an
+					instance of either 'EventDrivenConsumer' or 'PollingConsumer',
+					depending on whether the component's input channel is a
+					'SubscribableChannel' or 'PollableChannel'.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="auto-startup" default="true" use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Flag to indicate that the component should start automatically
+					on startup (default true).
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
+	<xsd:element name="remote-chunking-manager">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+						The remote chunking manager is used as the writer in batch jobs
+						to communicate with the middleware used to send chunks remotely.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:attribute name="id" type="xsd:ID" use="required"/>
+			<xsd:attribute name="message-template" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The messaging template to use.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.core.MessagingTemplate"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="step" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The step to be remotely chunked.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.batch.core.Step"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="reply-channel" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The channel to use for reply messages from workers.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="remote-chunking-worker">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+						The remote chunking worker receives chunks sent by the manager and
+						provides the processor and writer to handle the chunks, returning
+						status back to the manager.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:attribute name="id" type="xsd:ID" use="required"/>
+			<xsd:attribute name="input-channel" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The channel to use for receiving messages from the manager.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="output-channel" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The channel to use for sending messages to the manager.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.MessageChannel"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="item-processor" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The ItemProcessor implementation to use for processing items. If none
+								provided a PassThroughItemProcessor will be automatically used.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.batch.item.ItemProcessor"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="item-writer" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+								The ItemWriter implementation to use for writing items.
+							]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.batch.item.ItemWriter"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+</xsd:schema>
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderTest.java
similarity index 91%
rename from spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java
rename to spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderTest.java
index 9dd2a683a..0efbc4719 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingMasterStepBuilderTest.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -66,12 +66,13 @@ import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+
 /**
  * @author Mahmoud Ben Hassine
  */
 @RunWith(SpringRunner.class)
-@ContextConfiguration(classes = {RemoteChunkingMasterStepBuilderTest.BatchConfiguration.class})
-public class RemoteChunkingMasterStepBuilderTest {
+@ContextConfiguration(classes = {RemoteChunkingManagerStepBuilderTest.BatchConfiguration.class})
+public class RemoteChunkingManagerStepBuilderTest {
 
 	@Rule
 	public ExpectedException expectedException = ExpectedException.none();
@@ -92,7 +93,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 		this.expectedException.expectMessage("inputChannel must not be null");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.inputChannel(null)
 				.build();
 
@@ -107,7 +108,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 		this.expectedException.expectMessage("outputChannel must not be null");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.outputChannel(null)
 				.build();
 
@@ -122,7 +123,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 		this.expectedException.expectMessage("messagingTemplate must not be null");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.messagingTemplate(null)
 				.build();
 
@@ -137,7 +138,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 		this.expectedException.expectMessage("maxWaitTimeouts must be greater than zero");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.maxWaitTimeouts(-1)
 				.build();
 
@@ -152,7 +153,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 		this.expectedException.expectMessage("throttleLimit must be greater than zero");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.throttleLimit(-1L)
 				.build();
 
@@ -163,7 +164,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 	@Test
 	public void testMandatoryInputChannel() {
 		// given
-		RemoteChunkingMasterStepBuilder<String, String> builder = new RemoteChunkingMasterStepBuilder<>("step");
+		RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<>("step");
 
 		this.expectedException.expect(IllegalArgumentException.class);
 		this.expectedException.expectMessage("An InputChannel must be provided");
@@ -178,7 +179,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 	@Test
 	public void eitherOutputChannelOrMessagingTemplateMustBeProvided() {
 		// given
-		RemoteChunkingMasterStepBuilder<String, String> builder = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		RemoteChunkingManagerStepBuilder<String, String> builder = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.inputChannel(this.inputChannel)
 				.outputChannel(new DirectChannel())
 				.messagingTemplate(new MessagingTemplate());
@@ -197,13 +198,13 @@ public class RemoteChunkingMasterStepBuilderTest {
 	public void testUnsupportedOperationExceptionWhenSpecifyingAnItemWriter() {
 		// given
 		this.expectedException.expect(UnsupportedOperationException.class);
-		this.expectedException.expectMessage("When configuring a master " +
+		this.expectedException.expectMessage("When configuring a manager " +
 				"step for remote chunking, the item writer will be automatically " +
 				"set to an instance of ChunkMessageChannelItemWriter. " +
 				"The item writer must not be provided in this case.");
 
 		// when
-		TaskletStep step = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep step = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.reader(this.itemReader)
 				.writer(items -> { })
 				.repository(this.jobRepository)
@@ -217,9 +218,9 @@ public class RemoteChunkingMasterStepBuilderTest {
 	}
 
 	@Test
-	public void testMasterStepCreation() {
+	public void testManagerStepCreation() {
 		// when
-		TaskletStep taskletStep = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep taskletStep = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.reader(this.itemReader)
 				.repository(this.jobRepository)
 				.transactionManager(this.transactionManager)
@@ -292,7 +293,7 @@ public class RemoteChunkingMasterStepBuilderTest {
 			}
 		};
 
-		TaskletStep taskletStep = new RemoteChunkingMasterStepBuilder<String, String>("step")
+		TaskletStep taskletStep = new RemoteChunkingManagerStepBuilder<String, String>("step")
 				.reader(itemReader)
 				.readerIsTransactionalQueue()
 				.processor(itemProcessor)
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/config/xml/RemoteChunkingParserTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/config/xml/RemoteChunkingParserTests.java
index 6a2311960..282078bec 100644
--- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/config/xml/RemoteChunkingParserTests.java
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/config/xml/RemoteChunkingParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,16 +42,21 @@ import static org.junit.Assert.fail;
 
 /**
  * <p>
- * Test cases for the {@link org.springframework.batch.integration.config.xml.RemoteChunkingSlaveParser}
- * and {@link org.springframework.batch.integration.config.xml.RemoteChunkingMasterParser}.
+ * Test cases for the {@link RemoteChunkingWorkerParser}
+ * and {@link RemoteChunkingManagerParser}.
  * </p>
  *
  * @author Chris Schaefer
+ * @author Mahmoud Ben Hassine
  * @since 3.1
  */
 @SuppressWarnings("unchecked")
 public class RemoteChunkingParserTests {
 
+	/* TODO delete the following deprecated tests when related APIs are removed
+	 * /!\ Deliberately not using parametrized tests as it will be easier to delete
+	 * the following tests afterwards
+	 */
 	@SuppressWarnings("rawtypes")
 	@Test
 	public void testRemoteChunkingSlaveParserWithProcessorDefined() {
@@ -267,6 +272,223 @@ public class RemoteChunkingParserTests {
 		}
 	}
 
+	/* TODO end of deprecated tests to remove */
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testRemoteChunkingWorkerParserWithProcessorDefined() {
+		ApplicationContext applicationContext =
+				new ClassPathXmlApplicationContext("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserTests.xml");
+
+		ChunkHandler chunkHandler = applicationContext.getBean(ChunkProcessorChunkHandler.class);
+		ChunkProcessor chunkProcessor = (SimpleChunkProcessor) TestUtils.getPropertyValue(chunkHandler, "chunkProcessor");
+		assertNotNull("ChunkProcessor must not be null", chunkProcessor);
+
+		ItemWriter<String> itemWriter = (ItemWriter<String>) TestUtils.getPropertyValue(chunkProcessor, "itemWriter");
+		assertNotNull("ChunkProcessor ItemWriter must not be null", itemWriter);
+		assertTrue("Got wrong instance of ItemWriter", itemWriter instanceof Writer);
+
+		ItemProcessor<String, String> itemProcessor = (ItemProcessor<String, String>) TestUtils.getPropertyValue(chunkProcessor, "itemProcessor");
+		assertNotNull("ChunkProcessor ItemWriter must not be null", itemProcessor);
+		assertTrue("Got wrong instance of ItemProcessor", itemProcessor instanceof Processor);
+
+		FactoryBean serviceActivatorFactoryBean = applicationContext.getBean(ServiceActivatorFactoryBean.class);
+		assertNotNull("ServiceActivatorFactoryBean must not be null", serviceActivatorFactoryBean);
+		assertNotNull("Output channel name must not be null", TestUtils.getPropertyValue(serviceActivatorFactoryBean, "outputChannelName"));
+
+		MessageChannel inputChannel = applicationContext.getBean("requests", MessageChannel.class);
+		assertNotNull("Input channel must not be null", inputChannel);
+
+		String targetMethodName = (String) TestUtils.getPropertyValue(serviceActivatorFactoryBean, "targetMethodName");
+		assertNotNull("Target method name must not be null", targetMethodName);
+		assertTrue("Target method name must be handleChunk, got: " + targetMethodName, "handleChunk".equals(targetMethodName));
+
+		ChunkHandler targetObject = (ChunkHandler) TestUtils.getPropertyValue(serviceActivatorFactoryBean, "targetObject");
+		assertNotNull("Target object must not be null", targetObject);
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testRemoteChunkingWorkerParserWithProcessorNotDefined() {
+		ApplicationContext applicationContext =
+				new ClassPathXmlApplicationContext("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserNoProcessorTests.xml");
+
+		ChunkHandler chunkHandler = applicationContext.getBean(ChunkProcessorChunkHandler.class);
+		ChunkProcessor chunkProcessor = (SimpleChunkProcessor) TestUtils.getPropertyValue(chunkHandler, "chunkProcessor");
+		assertNotNull("ChunkProcessor must not be null", chunkProcessor);
+
+		ItemProcessor<String, String> itemProcessor = (ItemProcessor<String, String>) TestUtils.getPropertyValue(chunkProcessor, "itemProcessor");
+		assertNotNull("ChunkProcessor ItemWriter must not be null", itemProcessor);
+		assertTrue("Got wrong instance of ItemProcessor", itemProcessor instanceof PassThroughItemProcessor);
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testRemoteChunkingManagerParser() {
+		ApplicationContext applicationContext =
+				new ClassPathXmlApplicationContext("/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserTests.xml");
+
+		ItemWriter itemWriter = applicationContext.getBean("itemWriter", ChunkMessageChannelItemWriter.class);
+		assertNotNull("Messaging template must not be null", TestUtils.getPropertyValue(itemWriter, "messagingGateway"));
+		assertNotNull("Reply channel must not be null", TestUtils.getPropertyValue(itemWriter, "replyChannel"));
+
+		FactoryBean<ChunkHandler> remoteChunkingHandlerFactoryBean = applicationContext.getBean(RemoteChunkHandlerFactoryBean.class);
+		assertNotNull("Chunk writer must not be null", TestUtils.getPropertyValue(remoteChunkingHandlerFactoryBean, "chunkWriter"));
+		assertNotNull("Step must not be null", TestUtils.getPropertyValue(remoteChunkingHandlerFactoryBean, "step"));
+	}
+
+	@Test
+	public void testRemoteChunkingManagerIdAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingIdAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The id attribute must be specified" + " but got: " + iae.getMessage(),
+					"The id attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingManagerMessageTemplateAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingMessageTemplateAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The message-template attribute must be specified" + " but got: " + iae.getMessage(),
+					"The message-template attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingManagerStepAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingStepAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The step attribute must be specified" + " but got: " + iae.getMessage(),
+					"The step attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingManagerReplyChannelAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingReplyChannelAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The reply-channel attribute must be specified" + " but got: " + iae.getMessage(),
+					"The reply-channel attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingWorkerIdAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingIdAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The id attribute must be specified" + " but got: " + iae.getMessage(),
+					"The id attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingWorkerInputChannelAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingInputChannelAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The input-channel attribute must be specified" + " but got: " + iae.getMessage(),
+					"The input-channel attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingWorkerItemWriterAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingItemWriterAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The item-writer attribute must be specified" + " but got: " + iae.getMessage(),
+					"The item-writer attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
+	@Test
+	public void testRemoteChunkingWorkerOutputChannelAttrAssert() throws Exception {
+		ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext();
+		applicationContext.setValidating(false);
+		applicationContext.setConfigLocation("/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingOutputChannelAttrTests.xml");
+
+		try {
+			applicationContext.refresh();
+			fail();
+		} catch (BeanDefinitionStoreException e) {
+			assertTrue("Nested exception must be of type IllegalArgumentException", e.getCause() instanceof IllegalArgumentException);
+
+			IllegalArgumentException iae = (IllegalArgumentException) e.getCause();
+
+			assertTrue("Expected: " + "The output-channel attribute must be specified" + " but got: " + iae.getMessage(),
+					"The output-channel attribute must be specified".equals(iae.getMessage()));
+		}
+	}
+
 	private static class Writer implements ItemWriter<String> {
 		@Override
 		public void write(List<? extends String> items) throws Exception {
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingIdAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingIdAttrTests.xml
new file mode 100644
index 000000000..d977fe0a9
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingIdAttrTests.xml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-manager message-template="messagingTemplate" step="process" reply-channel="replies"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingMessageTemplateAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingMessageTemplateAttrTests.xml
new file mode 100644
index 000000000..0d349d5c9
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingMessageTemplateAttrTests.xml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-manager id="itemWriter" step="process" reply-channel="replies"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingReplyChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingReplyChannelAttrTests.xml
new file mode 100644
index 000000000..6187eebc2
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingReplyChannelAttrTests.xml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-manager id="itemWriter" message-template="messagingTemplate" step="process"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingStepAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingStepAttrTests.xml
new file mode 100644
index 000000000..be4f74893
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserMissingStepAttrTests.xml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-manager id="itemWriter" message-template="messagingTemplate" reply-channel="replies"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserTests.xml
new file mode 100644
index 000000000..1c1a58f29
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParserTests.xml
@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:batch="http://www.springframework.org/schema/batch"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:p="http://www.springframework.org/schema/p"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch
+		 https://www.springframework.org/schema/batch/spring-batch.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd
+		 http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<batch:job id="processingJob">
+		<batch:step id="process">
+			<batch:tasklet>
+				<batch:chunk reader="itemReader" writer="itemWriter" commit-interval="6"/>
+			</batch:tasklet>
+		</batch:step>
+	</batch:job>
+
+	<batch:job-repository/>
+
+	<bean id="itemReader" class="org.springframework.batch.item.support.ListItemReader">
+		<constructor-arg>
+			<list>
+				<value>1</value>
+				<value>2</value>
+				<value>3</value>
+				<value>4</value>
+				<value>5</value>
+				<value>6</value>
+			</list>
+		</constructor-arg>
+	</bean>
+
+	<batch-int:remote-chunking-manager id="itemWriter" message-template="messagingTemplate" step="process" reply-channel="replies"/>
+
+	<bean id="messagingTemplate" class="org.springframework.integration.core.MessagingTemplate"/>
+
+	<int:channel id="replies">
+		<int:queue/>
+	</int:channel>
+
+	<bean id="placeholderProperties" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"
+		  p:location="classpath:batch-${ENVIRONMENT:hsql}.properties"/>
+
+	<bean id="dataSource" class="org.apache.commons.dbcp2.BasicDataSource"
+		  p:driverClassName="${batch.jdbc.driver}" p:url="${batch.jdbc.url}"
+		  p:username="${batch.jdbc.user}" p:password="${batch.jdbc.password}"/>
+
+	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager"
+		  p:dataSource-ref="dataSource"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingIdAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingIdAttrTests.xml
index 3cde4ad00..b7e92bc00 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingIdAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingIdAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-master message-template="messagingTemplate" step="process" reply-channel="replies"/>
-</beans>
\ No newline at end of file
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingMessageTemplateAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingMessageTemplateAttrTests.xml
index badcdb37e..ccf1bd943 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingMessageTemplateAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingMessageTemplateAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-master id="itemWriter" step="process" reply-channel="replies"/>
-</beans>
\ No newline at end of file
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingReplyChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingReplyChannelAttrTests.xml
index 32c299ec1..82d102e6e 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingReplyChannelAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingReplyChannelAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-master id="itemWriter" message-template="messagingTemplate" step="process"/>
-</beans>
\ No newline at end of file
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingStepAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingStepAttrTests.xml
index d9abc1684..fa17d9c8b 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingStepAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserMissingStepAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-master id="itemWriter" message-template="messagingTemplate" reply-channel="replies"/>
-</beans>
\ No newline at end of file
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserTests.xml
index 9cac991df..74af85c3e 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingMasterParserTests.xml
@@ -11,7 +11,7 @@
 		 http://www.springframework.org/schema/batch
 		 https://www.springframework.org/schema/batch/spring-batch.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd
 		 http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
 
 	<batch:job id="processingJob">
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingIdAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingIdAttrTests.xml
index bbd80ecfb..5b201bfff 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingIdAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingIdAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave input-channel="requests" output-channel="replies"
 									 item-processor="itemProcessor" item-writer="itemWriter"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingInputChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingInputChannelAttrTests.xml
index fdaae8d00..bcc5becca 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingInputChannelAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingInputChannelAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave id="remote-chunking-slave" output-channel="replies"
 									 item-processor="itemProcessor" item-writer="itemWriter"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingItemWriterAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingItemWriterAttrTests.xml
index f86348b08..a7af0eef6 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingItemWriterAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingItemWriterAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave id="remote-chunking-slave" input-channel="requests" output-channel="replies"
 									 item-processor="itemProcessor"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingOutputChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingOutputChannelAttrTests.xml
index 2e6b3fe85..2284e3c84 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingOutputChannelAttrTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserMissingOutputChannelAttrTests.xml
@@ -6,7 +6,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave id="remote-chunking-slave" input-channel="requests"
 									 item-processor="itemProcessor" item-writer="itemWriter"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserNoProcessorTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserNoProcessorTests.xml
index 417cd25c0..05b96dda5 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserNoProcessorTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserNoProcessorTests.xml
@@ -7,7 +7,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave id="remote-chunking-slave" input-channel="requests" output-channel="replies"
 									 item-writer="itemWriter"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserTests.xml
index cc998991f..f5cd56468 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserTests.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingSlaveParserTests.xml
@@ -7,7 +7,7 @@
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans
 		 https://www.springframework.org/schema/beans/spring-beans.xsd
 		 http://www.springframework.org/schema/batch-integration
-		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration-3.1.xsd">
 
 	<batch-int:remote-chunking-slave id="remote-chunking-slave" input-channel="requests" output-channel="replies"
 									 item-processor="itemProcessor" item-writer="itemWriter"/>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingIdAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingIdAttrTests.xml
new file mode 100644
index 000000000..e071d2e5a
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingIdAttrTests.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker input-channel="requests" output-channel="replies"
+									 item-processor="itemProcessor" item-writer="itemWriter"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingInputChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingInputChannelAttrTests.xml
new file mode 100644
index 000000000..8b5ea7750
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingInputChannelAttrTests.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker id="remote-chunking-worker" output-channel="replies"
+									 item-processor="itemProcessor" item-writer="itemWriter"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingItemWriterAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingItemWriterAttrTests.xml
new file mode 100644
index 000000000..b490becd1
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingItemWriterAttrTests.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker id="remote-chunking-worker" input-channel="requests" output-channel="replies"
+									 item-processor="itemProcessor"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingOutputChannelAttrTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingOutputChannelAttrTests.xml
new file mode 100644
index 000000000..75b1d89b0
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserMissingOutputChannelAttrTests.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker id="remote-chunking-worker" input-channel="requests"
+									 item-processor="itemProcessor" item-writer="itemWriter"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserNoProcessorTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserNoProcessorTests.xml
new file mode 100644
index 000000000..3c1842e34
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserNoProcessorTests.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker id="remote-chunking-worker" input-channel="requests" output-channel="replies"
+									 item-writer="itemWriter"/>
+
+	<bean id="itemProcessor" class="org.springframework.batch.integration.config.xml.RemoteChunkingParserTests$Processor"/>
+
+	<bean id="itemWriter" class="org.springframework.batch.integration.config.xml.RemoteChunkingParserTests$Writer"/>
+
+	<int:channel id="requests"/>
+	<int:channel id="replies"/>
+</beans>
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserTests.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserTests.xml
new file mode 100644
index 000000000..1254a6ed3
--- /dev/null
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParserTests.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+		 https://www.springframework.org/schema/beans/spring-beans.xsd
+		 http://www.springframework.org/schema/batch-integration
+		 https://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">
+
+	<batch-int:remote-chunking-worker id="remote-chunking-worker" input-channel="requests" output-channel="replies"
+									 item-processor="itemProcessor" item-writer="itemWriter"/>
+
+	<bean id="itemProcessor" class="org.springframework.batch.integration.config.xml.RemoteChunkingParserTests$Processor"/>
+
+	<bean id="itemWriter" class="org.springframework.batch.integration.config.xml.RemoteChunkingParserTests$Writer"/>
+
+	<int:channel id="requests"/>
+	<int:channel id="replies"/>
+</beans>
diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/MasterConfiguration.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/ManagerConfiguration.java
similarity index 88%
rename from spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/MasterConfiguration.java
rename to spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/ManagerConfiguration.java
index 9f2c99814..861557de5 100644
--- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/MasterConfiguration.java
+++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotechunking/ManagerConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -23,7 +23,7 @@ import org.springframework.batch.core.Job;
 import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
-import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory;
+import org.springframework.batch.integration.chunk.RemoteChunkingManagerStepBuilderFactory;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -39,8 +39,8 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.jms.dsl.Jms;
 
 /**
- * This configuration class is for the master side of the remote chunking sample.
- * The master step reads numbers from 1 to 6 and sends 2 chunks {1, 2, 3} and
+ * This configuration class is for the manager side of the remote chunking sample.
+ * The manager step reads numbers from 1 to 6 and sends 2 chunks {1, 2, 3} and
  * {4, 5, 6} to workers for processing and writing.
  *
  * @author Mahmoud Ben Hassine
@@ -50,7 +50,7 @@ import org.springframework.integration.jms.dsl.Jms;
 @EnableBatchIntegration
 @EnableIntegration
 @PropertySource("classpath:remote-chunking.properties")
-public class MasterConfiguration {
+public class ManagerConfiguration {
 
 	@Value("${broker.url}")
 	private String brokerUrl;
@@ -59,7 +59,7 @@ public class MasterConfiguration {
 	private JobBuilderFactory jobBuilderFactory;
 
 	@Autowired
-	private RemoteChunkingMasterStepBuilderFactory masterStepBuilderFactory;
+	private RemoteChunkingManagerStepBuilderFactory managerStepBuilderFactory;
 
 	@Bean
 	public ActiveMQConnectionFactory connectionFactory() {
@@ -110,8 +110,8 @@ public class MasterConfiguration {
 	}
 
 	@Bean
-	public TaskletStep masterStep() {
-		return this.masterStepBuilderFactory.get("masterStep")
+	public TaskletStep managerStep() {
+		return this.managerStepBuilderFactory.get("managerStep")
 				.<Integer, Integer>chunk(3)
 				.reader(itemReader())
 				.outputChannel(requests())
@@ -122,7 +122,7 @@ public class MasterConfiguration {
 	@Bean
 	public Job remoteChunkingJob() {
 		return this.jobBuilderFactory.get("remoteChunkingJob")
-				.start(masterStep())
+				.start(managerStep())
 				.build();
 	}
 
diff --git 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/ManagerConfiguration.java
similarity index 81%
rename from spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java
rename to spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/ManagerConfiguration.java
index eceb2b760..07078272b 100644
--- 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/ManagerConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -22,7 +22,7 @@ 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.integration.config.annotation.EnableBatchIntegration;
-import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.batch.sample.remotepartitioning.BasicPartitioner;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
@@ -35,8 +35,8 @@ 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.
+ * This configuration class is for the manager side of the remote partitioning sample.
+ * The manager step will create 3 partitions for workers to process.
  *
  * @author Mahmoud Ben Hassine
  */
@@ -44,20 +44,20 @@ import org.springframework.integration.jms.dsl.Jms;
 @EnableBatchProcessing
 @EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
-public class MasterConfiguration {
+public class ManagerConfiguration {
 
 	private static final int GRID_SIZE = 3;
 
 	private final JobBuilderFactory jobBuilderFactory;
 
-	private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+	private final RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory;
 
 
-	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
-							   RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) {
+	public ManagerConfiguration(JobBuilderFactory jobBuilderFactory,
+								RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory) {
 
 		this.jobBuilderFactory = jobBuilderFactory;
-		this.masterStepBuilderFactory = masterStepBuilderFactory;
+		this.managerStepBuilderFactory = managerStepBuilderFactory;
 	}
 
 	/*
@@ -93,11 +93,11 @@ public class MasterConfiguration {
 	}
 
 	/*
-	 * Configure the master step
+	 * Configure the manager step
 	 */
 	@Bean
-	public Step masterStep() {
-		return this.masterStepBuilderFactory.get("masterStep")
+	public Step managerStep() {
+		return this.managerStepBuilderFactory.get("managerStep")
 				.partitioner("workerStep", new BasicPartitioner())
 				.gridSize(GRID_SIZE)
 				.outputChannel(requests())
@@ -108,7 +108,7 @@ public class MasterConfiguration {
 	@Bean
 	public Job remotePartitioningJob() {
 		return this.jobBuilderFactory.get("remotePartitioningJob")
-				.start(masterStep())
+				.start(managerStep())
 				.build();
 	}
 
diff --git 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
index fe491e2b6..ad756f3c0 100644
--- 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -37,7 +37,7 @@ 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.
+ * Each worker will process a partition sent by the manager step.
  *
  * @author Mahmoud Ben Hassine
  */
@@ -55,7 +55,7 @@ public class WorkerConfiguration {
 	}
 
 	/*
-	 * Configure inbound flow (requests coming from the master)
+	 * Configure inbound flow (requests coming from the manager)
 	 */
 	@Bean
 	public DirectChannel requests() {
@@ -71,7 +71,7 @@ public class WorkerConfiguration {
 	}
 
 	/*
-	 * Configure outbound flow (replies going to the master)
+	 * Configure outbound flow (replies going to the manager)
 	 */
 	@Bean
 	public DirectChannel replies() {
diff --git 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/ManagerConfiguration.java
similarity index 79%
rename from spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java
rename to spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/ManagerConfiguration.java
index 96f5636ab..b5e3beff9 100644
--- 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/ManagerConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -22,7 +22,7 @@ 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.integration.config.annotation.EnableBatchIntegration;
-import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory;
+import org.springframework.batch.integration.partition.RemotePartitioningManagerStepBuilderFactory;
 import org.springframework.batch.sample.remotepartitioning.BasicPartitioner;
 import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration;
 import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration;
@@ -35,8 +35,8 @@ 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.
+ * This configuration class is for the manager side of the remote partitioning sample.
+ * The manager step will create 3 partitions for workers to process.
  *
  * @author Mahmoud Ben Hassine
  */
@@ -44,20 +44,20 @@ import org.springframework.integration.jms.dsl.Jms;
 @EnableBatchProcessing
 @EnableBatchIntegration
 @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class})
-public class MasterConfiguration {
+public class ManagerConfiguration {
 
 	private static final int GRID_SIZE = 3;
 
 	private final JobBuilderFactory jobBuilderFactory;
 
-	private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+	private final RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory;
 
 
-	public MasterConfiguration(JobBuilderFactory jobBuilderFactory,
-							   RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) {
+	public ManagerConfiguration(JobBuilderFactory jobBuilderFactory,
+								RemotePartitioningManagerStepBuilderFactory managerStepBuilderFactory) {
 
 		this.jobBuilderFactory = jobBuilderFactory;
-		this.masterStepBuilderFactory = masterStepBuilderFactory;
+		this.managerStepBuilderFactory = managerStepBuilderFactory;
 	}
 
 	/*
@@ -77,11 +77,11 @@ public class MasterConfiguration {
 	}
 
 	/*
-	 * Configure the master step
+	 * Configure the manager step
 	 */
 	@Bean
-	public Step masterStep() {
-		return this.masterStepBuilderFactory.get("masterStep")
+	public Step managerStep() {
+		return this.managerStepBuilderFactory.get("managerStep")
 				.partitioner("workerStep", new BasicPartitioner())
 				.gridSize(GRID_SIZE)
 				.outputChannel(requests())
@@ -91,7 +91,7 @@ public class MasterConfiguration {
 	@Bean
 	public Job remotePartitioningJob() {
 		return this.jobBuilderFactory.get("remotePartitioningJob")
-				.start(masterStep())
+				.start(managerStep())
 				.build();
 	}
 
diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemoteChunkingJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemoteChunkingJobFunctionalTests.java
index 928703155..5a9256e2f 100644
--- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemoteChunkingJobFunctionalTests.java
+++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemoteChunkingJobFunctionalTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -25,7 +25,7 @@ import org.junit.runner.RunWith;
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.sample.config.JobRunnerConfiguration;
-import org.springframework.batch.sample.remotechunking.MasterConfiguration;
+import org.springframework.batch.sample.remotechunking.ManagerConfiguration;
 import org.springframework.batch.sample.remotechunking.WorkerConfiguration;
 import org.springframework.batch.test.JobLauncherTestUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,13 +36,13 @@ import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
- * The master step of the job under test will read data and send chunks to the worker
+ * The manager step of the job under test will read data and send chunks to the worker
  * (started in {@link RemoteChunkingJobFunctionalTests#setUp()}) for processing and writing.
  *
  * @author Mahmoud Ben Hassine
  */
 @RunWith(SpringRunner.class)
-@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
+@ContextConfiguration(classes = {JobRunnerConfiguration.class, ManagerConfiguration.class})
 @PropertySource("classpath:remote-chunking.properties")
 public class RemoteChunkingJobFunctionalTests {
 
@@ -81,7 +81,7 @@ public class RemoteChunkingJobFunctionalTests {
 		// then
 		Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
 		Assert.assertEquals(
-				"Waited for 2 results.", // the master sent 2 chunks ({1, 2, 3} and {4, 5, 6}) to workers
+				"Waited for 2 results.", // the manager sent 2 chunks ({1, 2, 3} and {4, 5, 6}) to workers
 				jobExecution.getExitStatus().getExitDescription());
 	}
 
diff --git 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
index 02643fd8a..c15507c79 100644
--- 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -16,17 +16,17 @@
 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.ManagerConfiguration;
 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
+ * The manager step of the job under test will create 3 partitions for workers
  * to process.
  *
  * @author Mahmoud Ben Hassine
  */
-@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
+@ContextConfiguration(classes = {JobRunnerConfiguration.class, ManagerConfiguration.class})
 public class RemotePartitioningJobWithMessageAggregationFunctionalTests extends RemotePartitioningJobFunctionalTests {
 
 	@Override
diff --git 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
index 0140077a1..cca8cfb4b 100644
--- 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -16,17 +16,17 @@
 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.ManagerConfiguration;
 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
+ * The manager step of the job under test will create 3 partitions for workers
  * to process.
  *
  * @author Mahmoud Ben Hassine
  */
-@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class})
+@ContextConfiguration(classes = {JobRunnerConfiguration.class, ManagerConfiguration.class})
 public class RemotePartitioningJobWithRepositoryPollingFunctionalTests extends RemotePartitioningJobFunctionalTests {
 
 	@Override