diff --git a/spring-batch-docs/asciidoc/images/remote-partitioning-aggregation-config.png b/spring-batch-docs/asciidoc/images/remote-partitioning-aggregation-config.png new file mode 100644 index 000000000..c78447d0e Binary files /dev/null and b/spring-batch-docs/asciidoc/images/remote-partitioning-aggregation-config.png differ diff --git a/spring-batch-docs/asciidoc/images/remote-partitioning-polling-config.png b/spring-batch-docs/asciidoc/images/remote-partitioning-polling-config.png new file mode 100644 index 000000000..cc48e824f Binary files /dev/null and b/spring-batch-docs/asciidoc/images/remote-partitioning-polling-config.png differ diff --git a/spring-batch-docs/asciidoc/spring-batch-integration.adoc b/spring-batch-docs/asciidoc/spring-batch-integration.adoc index 11a46ed39..78cf0eaaf 100644 --- a/spring-batch-docs/asciidoc/spring-batch-integration.adoc +++ b/spring-batch-docs/asciidoc/spring-batch-integration.adoc @@ -945,7 +945,7 @@ two beans that can be autowired in the application context: * `RemoteChunkingMasterStepBuilderFactory`: used to configure the master step * `RemoteChunkingWorkerBuilder`: used to configure the remote worker integration flow -These APIs will take care of configuring a number of components as described in the following diagram: +These APIs take care of configuring a number of components as described in the following diagram: .Remote Chunking Configuration image::{batch-asciidoc}images/remote-chunking-config.png[Remote Chunking Configuration, scaledwidth="80%"] @@ -1261,3 +1261,92 @@ You must also ensure that the partition `handler` attribute maps to the `partiti You can find a complete example of a remote partitioning job link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-partitioning-sample$$[here]. + +The `@EnableBatchIntegration` annotation that can be used to simplify a remote + partitioning setup. This annotation provides two beans useful for remote partitioning: + +* `RemotePartitioningMasterStepBuilderFactory`: used to configure the master step +* `RemotePartitioningWorkerStepBuilderFactory`: used to configure the worker step + +These APIs take care of configuring a number of components as described in the following diagram: + +.Remote Partitioning Configuration (with job repository polling) +image::{batch-asciidoc}images/remote-partitioning-polling-config.png[Remote Partitioning Configuration (with job repository polling), scaledwidth="80%"] + +.Remote Partitioning Configuration (with replies aggregation) +image::{batch-asciidoc}images/remote-partitioning-aggregation-config.png[Remote Partitioning Configuration (with replies aggregation), scaledwidth="80%"] + +On the master side, the `RemotePartitioningMasterStepBuilderFactory` allows you to +configure a master step by declaring: + +* the `Partitioner` used to partition data +* the output channel ("Outgoing requests") to send requests to workers +* the input channel ("Incoming replies") to receive replies from workers (when configuring replies aggregation) +* the poll interval and timeout parameters (when configuring job repository polling) + +The `MessageChannelPartitionHandler` and the `MessagingTemplate` are not needed to be explicitly configured +(Those can still be explicitly configured if required). + +On the worker side, the `RemotePartitioningWorkerStepBuilderFactory` allows you to configure a worker to: + +* listen to requests sent by the master on the input channel ("Incoming requests") +* call the `handle` method of `StepExecutionRequestHandler` for each request +* send replies on the output channel ("Outgoing replies") to the master + +There is no need to explicitly configure the `StepExecutionRequestHandler` (which can be explicitly configured if required). + +The following example shows how to use these APIs: + +[source, java] +---- +@Configuration +@EnableBatchProcessing +@EnableBatchIntegration +public class RemotePartitioningJobConfiguration { + + @Configuration + public static class MasterConfiguration { + + @Autowired + private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory; + + @Bean + public Step masterStep() { + return this.masterStepBuilderFactory + .get("masterStep") + .partitioner("workerStep", partitioner()) + .gridSize(10) + .outputChannel(outgoingRequestsToWorkers()) + .inputChannel(incomingRepliesFromWorkers()) + .build(); + } + + // Middleware beans setup omitted + + } + + @Configuration + public static class WorkerConfiguration { + + @Autowired + private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory; + + @Bean + public Step workerStep() { + return this.workerStepBuilderFactory + .get("workerStep") + .inputChannel(incomingRequestsFromMaster()) + .outputChannel(outgoingRepliesToMaster()) + .chunk(100) + .reader(itemReader()) + .processor(itemProcessor()) + .writer(itemWriter()) + .build(); + } + + // Middleware beans setup omitted + + } + +} +---- diff --git a/spring-batch-docs/asciidoc/whatsnew.adoc b/spring-batch-docs/asciidoc/whatsnew.adoc index 655138f17..191b1ed71 100644 --- a/spring-batch-docs/asciidoc/whatsnew.adoc +++ b/spring-batch-docs/asciidoc/whatsnew.adoc @@ -9,7 +9,7 @@ The Spring Batch 4.1 release adds the following features: * A new `@SpringBatchTest` annotation to simplify testing batch components -* A new `@EnableBatchIntegration` annotation to simplify remote chunking configuration +* A new `@EnableBatchIntegration` annotation to simplify remote chunking and partitioning configuration * A new `JsonItemReader` and `JsonFileItemWriter` to support the JSON format * Add support for validating items with the Bean Validation API @@ -125,6 +125,55 @@ that uses these new APIs in the link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-chunking-sample$$[samples module] as well as more details in the <> chapter. +Just like the remote chunking configuration simplification, this version also +introduces new APIs to simplify a remote partitioning setup: +`RemotePartitioningMasterStepBuilder` and `RemotePartitioningWorkerStepBuilder`. +Those can be autowired in your configuration class if the +`@EnableBatchIntegration` is present as shown in the following example: + +[source, java] +---- +@Configuration +@EnableBatchProcessing +@EnableBatchIntegration +public class RemotePartitioningAppConfig { + + @Autowired + private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory; + + @Autowired + private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory; + + @Bean + public Step masterStep() { + return this.masterStepBuilderFactory + .get("masterStep") + .partitioner("workerStep", partitioner()) + .gridSize(10) + .outputChannel(outgoingRequestsToWorkers()) + .inputChannel(incomingRepliesFromWorkers()) + .build(); + } + + @Bean + public Step workerStep() { + return this.workerStepBuilderFactory + .get("workerStep") + .inputChannel(incomingRequestsFromMaster()) + .outputChannel(outgoingRepliesToMaster()) + .chunk(100) + .reader(itemReader()) + .processor(itemProcessor()) + .writer(itemWriter()) + .build(); + } + + // Middleware beans setup omitted +} +---- + +You can find more details about these new APIs in the <> chapter. + [[whatsNewJson]] === JSON support 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 087e6401d..2176a6147 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 @@ -15,9 +15,12 @@ */ package org.springframework.batch.integration.config.annotation; +import org.springframework.batch.core.explore.JobExplorer; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory; import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder; +import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory; +import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -32,6 +35,8 @@ import org.springframework.transaction.PlatformTransactionManager; @Configuration public class BatchIntegrationConfiguration { + private JobExplorer jobExplorer; + private JobRepository jobRepository; private PlatformTransactionManager transactionManager; @@ -39,9 +44,11 @@ public class BatchIntegrationConfiguration { @Autowired public BatchIntegrationConfiguration( JobRepository jobRepository, + JobExplorer jobExplorer, PlatformTransactionManager transactionManager) { this.jobRepository = jobRepository; + this.jobExplorer = jobExplorer; this.transactionManager = transactionManager; } @@ -56,4 +63,16 @@ public class BatchIntegrationConfiguration { return new RemoteChunkingWorkerBuilder<>(); } + @Bean + public RemotePartitioningMasterStepBuilderFactory remotePartitioningMasterStepBuilderFactory() { + return new RemotePartitioningMasterStepBuilderFactory(this.jobRepository, + this.jobExplorer, this.transactionManager); + } + + @Bean + public RemotePartitioningWorkerStepBuilderFactory remotePartitioningWorkerStepBuilderFactory() { + return new RemotePartitioningWorkerStepBuilderFactory(this.jobRepository, + this.jobExplorer, this.transactionManager); + } + } 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 0b3acf697..b36370100 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 @@ -23,24 +23,33 @@ import java.lang.annotation.Target; import org.springframework.batch.integration.chunk.RemoteChunkingWorkerBuilder; import org.springframework.batch.integration.chunk.RemoteChunkingMasterStepBuilderFactory; +import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory; +import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory; import org.springframework.context.annotation.Import; +import org.springframework.integration.config.EnableIntegration; /** * Enable Spring Batch Integration features and provide a base configuration for - * setting up remote chunking infrastructure beans. + * setting up remote chunking or partitioning infrastructure beans. * * By adding this annotation on a {@link org.springframework.context.annotation.Configuration} * class, it will be possible to autowire the following beans: * * * - * For example: + * For remote chunking, an example of a configuration class would be: * * 
  * @Configuration
@@ -75,15 +84,60 @@ import org.springframework.context.annotation.Import;
  *       		.build();
  * 	}
  *
+ * 	// Middleware beans omitted
+ *
  * }
  *
* + * For remote partitioning, an example of a configuration class would be: + * + * 
+ * @Configuration
+ * @EnableBatchIntegration
+ * @EnableBatchProcessing
+ * public class RemotePartitioningAppConfig {
+ *
+ * 	@Autowired
+ * 	private RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory;
+ *
+ * 	@Autowired
+ * 	private RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory;
+ *
+ * 	@Bean
+ * 	public Step masterStep() {
+ *       	 return this.masterStepBuilderFactory
+ *       		.get("masterStep")
+ *       		.partitioner("workerStep", partitioner())
+ *       		.gridSize(10)
+ *       		.outputChannel(outgoingRequestsToWorkers())
+ *       		.inputChannel(incomingRepliesFromWorkers())
+ *       		.build();
+ * 	}
+ *
+ * 	@Bean
+ * 	public Step workerStep() {
+ *       	 return this.workerStepBuilderFactory
+ *       		.get("workerStep")
+ *       		.inputChannel(incomingRequestsFromMaster())
+ *       		.outputChannel(outgoingRepliesToMaster())
+ *       		.chunk(100)
+ *       		.reader(itemReader())
+ *       		.processor(itemProcessor())
+ *       		.writer(itemWriter())
+ *       		.build();
+ * 	}
+ *
+ * 	// Middleware beans omitted
+ *
+ * }
+ *
* @since 4.1 * @author Mahmoud Ben Hassine */ @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented +@EnableIntegration @Import(BatchIntegrationConfiguration.class) public @interface EnableBatchIntegration { } 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 new file mode 100644 index 000000000..84e323e19 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilder.java @@ -0,0 +1,297 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.core.partition.support.StepExecutionAggregator; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.builder.PartitionStepBuilder; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.StandardIntegrationFlow; +import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * Builder for a master step in a remote partitioning setup. This builder creates and + * sets a {@link MessageChannelPartitionHandler} on the master step. + * + *

If no {@code messagingTemplate} is provided through + * {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)}, + * this builder will create one. The {@code outputChannel} set with + * {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)} will be + * used as a default channel of the messaging template and will override any default + * channel already set on the messaging template. + * + * @since 4.1 + * @author Mahmoud Ben Hassine + */ +public class RemotePartitioningMasterStepBuilder extends PartitionStepBuilder { + + private static final long DEFAULT_POLL_INTERVAL = 10000L; + private static final long DEFAULT_TIMEOUT = -1L; + + private MessagingTemplate messagingTemplate; + private MessageChannel inputChannel; + private MessageChannel outputChannel; + private JobExplorer jobExplorer; + private BeanFactory beanFactory; + private long pollInterval = DEFAULT_POLL_INTERVAL; + private long timeout = DEFAULT_TIMEOUT; + + /** + * Create a new {@link RemotePartitioningMasterStepBuilder}. + * @param stepName name of the master step + */ + public RemotePartitioningMasterStepBuilder(String stepName) { + super(new StepBuilder(stepName)); + } + + /** + * Set the input channel on which replies from workers will be received. + * @param inputChannel the input channel + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder inputChannel(MessageChannel inputChannel) { + Assert.notNull(inputChannel, "inputChannel must not be null"); + this.inputChannel = inputChannel; + return this; + } + + /** + * Set the output channel on which requests to workers will be sent. + * The output channel will be set as a default channel on the provided + * {@link MessagingTemplate} trough + * {@link RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate)} + * (or the one created by this builder if no messaging template is provided). + * @param outputChannel the output channel. + * @return this builder instance for fluent chaining + * + * @see RemotePartitioningMasterStepBuilder#messagingTemplate(MessagingTemplate) + */ + public RemotePartitioningMasterStepBuilder outputChannel(MessageChannel outputChannel) { + Assert.notNull(outputChannel, "outputChannel must not be null"); + this.outputChannel = outputChannel; + return this; + } + + /** + * Set the {@link MessagingTemplate} to use to send data to workers. + *

The default destination of the messaging template will be + * overridden by the output channel provided through + * {@link RemotePartitioningMasterStepBuilder#outputChannel(MessageChannel)}.

+ * + * @param messagingTemplate the messaging template to use + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder messagingTemplate(MessagingTemplate messagingTemplate) { + Assert.notNull(messagingTemplate, "messagingTemplate must not be null"); + this.messagingTemplate = messagingTemplate; + return this; + } + + /** + * Set the job explorer. + * @param jobExplorer the job explorer to use. + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder jobExplorer(JobExplorer jobExplorer) { + Assert.notNull(jobExplorer, "jobExplorer must not be null"); + this.jobExplorer = jobExplorer; + return this; + } + + /** + * How often to poll the job repository for the status of the workers. Defaults to 10 seconds. + * @param pollInterval the poll interval value in milliseconds + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder pollInterval(long pollInterval) { + Assert.isTrue(pollInterval > 0, "The poll interval must be greater than zero"); + this.pollInterval = pollInterval; + return this; + } + + /** + * When using job repository polling, the time limit to wait. Defaults to -1 (no timeout). + * @param timeout the timeout value in milliseconds + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder timeout(long timeout) { + this.timeout = timeout; + return this; + } + + /** + * Set the bean factory. + * @param beanFactory the bean factory to use + * @return this builder instance for fluent chaining + */ + public RemotePartitioningMasterStepBuilder beanFactory(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + return this; + } + + public Step build() { + Assert.notNull(this.outputChannel, "An outputChannel must be provided"); + + // configure messaging template + if (this.messagingTemplate == null) { + this.messagingTemplate = new MessagingTemplate(); + } + this.messagingTemplate.setDefaultChannel(this.outputChannel); + + // Configure the partition handler + final MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); + partitionHandler.setStepName(getStepName()); + partitionHandler.setGridSize(getGridSize()); + partitionHandler.setMessagingOperations(this.messagingTemplate); + + if (isPolling()) { + partitionHandler.setJobExplorer(this.jobExplorer); + partitionHandler.setPollInterval(this.pollInterval); + partitionHandler.setTimeout(this.timeout); + } + else { + PollableChannel replies = new QueueChannel(); + partitionHandler.setReplyChannel(replies); + StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows + .from(this.inputChannel) + .aggregate(aggregatorSpec -> aggregatorSpec.processor(partitionHandler)) + .channel(replies) + .get(); + IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class); + integrationFlowContext.registration(standardIntegrationFlow) + .autoStartup(false) + .register(); + } + + try { + partitionHandler.afterPropertiesSet(); + super.partitionHandler(partitionHandler); + } + catch (Exception e) { + throw new BeanCreationException("Unable to create a master step for remote partitioning", e); + } + + return super.build(); + } + + private boolean isPolling() { + return this.inputChannel == null; + } + + @Override + public RemotePartitioningMasterStepBuilder repository(JobRepository jobRepository) { + super.repository(jobRepository); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder transactionManager(PlatformTransactionManager transactionManager) { + super.transactionManager(transactionManager); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder partitioner(String slaveStepName, Partitioner partitioner) { + super.partitioner(slaveStepName, partitioner); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder gridSize(int gridSize) { + super.gridSize(gridSize); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder step(Step step) { + super.step(step); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder splitter(StepExecutionSplitter splitter) { + super.splitter(splitter); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder aggregator(StepExecutionAggregator aggregator) { + super.aggregator(aggregator); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder startLimit(int startLimit) { + super.startLimit(startLimit); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder listener(Object listener) { + super.listener(listener); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder listener(StepExecutionListener listener) { + super.listener(listener); + return this; + } + + @Override + public RemotePartitioningMasterStepBuilder allowStartIfComplete(boolean allowStartIfComplete) { + super.allowStartIfComplete(allowStartIfComplete); + return this; + } + + /** + * This method will throw a {@link UnsupportedOperationException} since + * the partition handler of the master step will be automatically set to an + * instance of {@link MessageChannelPartitionHandler}. + * + * When building a master step for remote partitioning using this builder, + * no partition handler must be provided. + * + * @param partitionHandler a partition handler + * @return this builder instance for fluent chaining + * @throws UnsupportedOperationException if a partition handler is provided + */ + @Override + public RemotePartitioningMasterStepBuilder partitionHandler(PartitionHandler partitionHandler) throws UnsupportedOperationException { + throw new UnsupportedOperationException("When configuring a master step " + + "for remote partitioning using the RemotePartitioningMasterStepBuilder, " + + "the partition handler will be automatically set to an instance " + + "of MessageChannelPartitionHandler. The partition handler must " + + "not be provided in this case."); + } + +} 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 new file mode 100644 index 000000000..85a420d6e --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderFactory.java @@ -0,0 +1,75 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.transaction.PlatformTransactionManager; + +/** + * Convenient factory for a {@link RemotePartitioningMasterStepBuilder} which sets + * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and + * {@link PlatformTransactionManager} automatically. + * + * @since 4.1 + * @author Mahmoud Ben Hassine + */ +public class RemotePartitioningMasterStepBuilderFactory implements BeanFactoryAware { + + private BeanFactory beanFactory; + final private JobExplorer jobExplorer; + final private JobRepository jobRepository; + final private PlatformTransactionManager transactionManager; + + + /** + * Create a new {@link RemotePartitioningMasterStepBuilderFactory}. + * @param jobRepository the job repository to use + * @param jobExplorer the job explorer to use + * @param transactionManager the transaction manager to use + */ + public RemotePartitioningMasterStepBuilderFactory(JobRepository jobRepository, + JobExplorer jobExplorer, PlatformTransactionManager transactionManager) { + + this.jobRepository = jobRepository; + this.jobExplorer = jobExplorer; + this.transactionManager = transactionManager; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + /** + * Creates a {@link RemotePartitioningMasterStepBuilder} and initializes its job + * repository, job explorer, bean factory and transaction manager. + * @param name the name of the step + * @return a {@link RemotePartitioningMasterStepBuilder} + */ + public RemotePartitioningMasterStepBuilder get(String name) { + return new RemotePartitioningMasterStepBuilder(name) + .repository(this.jobRepository) + .jobExplorer(this.jobExplorer) + .beanFactory(this.beanFactory) + .transactionManager(this.transactionManager); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java new file mode 100644 index 000000000..c812ae13e --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java @@ -0,0 +1,258 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.StepLocator; +import org.springframework.batch.core.step.builder.FlowStepBuilder; +import org.springframework.batch.core.step.builder.JobStepBuilder; +import org.springframework.batch.core.step.builder.PartitionStepBuilder; +import org.springframework.batch.core.step.builder.SimpleStepBuilder; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.core.step.builder.TaskletStepBuilder; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.repeat.CompletionPolicy; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.StandardIntegrationFlow; +import org.springframework.integration.dsl.context.IntegrationFlowContext; +import org.springframework.messaging.MessageChannel; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * Builder for a worker step in a remote partitioning setup. This builder + * creates an {@link IntegrationFlow} that: + * + * + * + * @since 4.1 + * @author Mahmoud Ben Hassine + */ +public class RemotePartitioningWorkerStepBuilder extends StepBuilder { + + private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handle"; + private static final Log logger = LogFactory.getLog(RemotePartitioningWorkerStepBuilder.class); + + private MessageChannel inputChannel; + private MessageChannel outputChannel; + private JobExplorer jobExplorer; + private StepLocator stepLocator; + private BeanFactory beanFactory; + + /** + * Initialize a step builder for a step with the given name. + * @param name the name of the step + */ + public RemotePartitioningWorkerStepBuilder(String name) { + super(name); + } + + /** + * Set the input channel on which step execution requests sent by the master + * are received. + * @param inputChannel the input channel + * @return this builder instance for fluent chaining + */ + public RemotePartitioningWorkerStepBuilder inputChannel(MessageChannel inputChannel) { + Assert.notNull(inputChannel, "inputChannel must not be null"); + this.inputChannel = inputChannel; + return this; + } + + /** + * Set the output channel on which replies will be sent to the master step. + * @param outputChannel the input channel + * @return this builder instance for fluent chaining + */ + public RemotePartitioningWorkerStepBuilder outputChannel(MessageChannel outputChannel) { + Assert.notNull(outputChannel, "outputChannel must not be null"); + this.outputChannel = outputChannel; + return this; + } + + /** + * Set the job explorer. + * @param jobExplorer the job explorer to use + * @return this builder instance for fluent chaining + */ + public RemotePartitioningWorkerStepBuilder jobExplorer(JobExplorer jobExplorer) { + Assert.notNull(jobExplorer, "jobExplorer must not be null"); + this.jobExplorer = jobExplorer; + return this; + } + + /** + * Set the step locator used to locate the worker step to execute. + * @param stepLocator the step locator to use + * @return this builder instance for fluent chaining + */ + public RemotePartitioningWorkerStepBuilder stepLocator(StepLocator stepLocator) { + Assert.notNull(stepLocator, "stepLocator must not be null"); + this.stepLocator = stepLocator; + return this; + } + + /** + * Set the bean factory. + * @param beanFactory the bean factory + * @return this builder instance for fluent chaining + */ + public RemotePartitioningWorkerStepBuilder beanFactory(BeanFactory beanFactory) { + Assert.notNull(beanFactory, "beanFactory must not be null"); + this.beanFactory = beanFactory; + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder repository(JobRepository jobRepository) { + super.repository(jobRepository); + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder transactionManager(PlatformTransactionManager transactionManager) { + super.transactionManager(transactionManager); + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder startLimit(int startLimit) { + super.startLimit(startLimit); + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder listener(Object listener) { + super.listener(listener); + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder listener(StepExecutionListener listener) { + super.listener(listener); + return this; + } + + @Override + public RemotePartitioningWorkerStepBuilder allowStartIfComplete(boolean allowStartIfComplete) { + super.allowStartIfComplete(allowStartIfComplete); + return this; + } + + @Override + public TaskletStepBuilder tasklet(Tasklet tasklet) { + configureWorkerIntegrationFlow(); + return super.tasklet(tasklet); + } + + @Override + public SimpleStepBuilder chunk(int chunkSize) { + configureWorkerIntegrationFlow(); + return super.chunk(chunkSize); + } + + @Override + public SimpleStepBuilder chunk(CompletionPolicy completionPolicy) { + configureWorkerIntegrationFlow(); + return super.chunk(completionPolicy); + } + + @Override + public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) { + configureWorkerIntegrationFlow(); + return super.partitioner(stepName, partitioner); + } + + @Override + public PartitionStepBuilder partitioner(Step step) { + configureWorkerIntegrationFlow(); + return super.partitioner(step); + } + + @Override + public JobStepBuilder job(Job job) { + configureWorkerIntegrationFlow(); + return super.job(job); + } + + @Override + public FlowStepBuilder flow(Flow flow) { + configureWorkerIntegrationFlow(); + return super.flow(flow); + } + + /** + * Create an {@link IntegrationFlow} with a {@link StepExecutionRequestHandler} + * configured as a service activator listening to the input channel and replying + * on the output channel. + */ + private void configureWorkerIntegrationFlow() { + Assert.notNull(this.inputChannel, "An InputChannel must be provided"); + Assert.notNull(this.jobExplorer, "A JobExplorer must be provided"); + + if (this.stepLocator == null) { + BeanFactoryStepLocator beanFactoryStepLocator = new BeanFactoryStepLocator(); + beanFactoryStepLocator.setBeanFactory(this.beanFactory); + this.stepLocator = beanFactoryStepLocator; + } + if (this.outputChannel == null) { + if (logger.isDebugEnabled()) { + logger.debug("The output channel is set to a NullChannel. " + + "The master step must poll the job repository for workers status."); + } + this.outputChannel = new NullChannel(); + } + + StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); + stepExecutionRequestHandler.setJobExplorer(this.jobExplorer); + stepExecutionRequestHandler.setStepLocator(this.stepLocator); + + StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows + .from(this.inputChannel) + .handle(stepExecutionRequestHandler, SERVICE_ACTIVATOR_METHOD_NAME) + .channel(this.outputChannel) + .get(); + IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class); + integrationFlowContext.registration(standardIntegrationFlow) + .autoStartup(false) + .register(); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java new file mode 100644 index 000000000..920f0bc58 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java @@ -0,0 +1,76 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.transaction.PlatformTransactionManager; + +/** + * Convenient factory for a {@link RemotePartitioningWorkerStepBuilder} which sets + * the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and + * {@link PlatformTransactionManager} automatically. + * + * @since 4.1 + * @author Mahmoud Ben Hassine + */ +public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAware { + + private BeanFactory beanFactory; + final private JobExplorer jobExplorer; + final private JobRepository jobRepository; + final private PlatformTransactionManager transactionManager; + + + /** + * Create a new {@link RemotePartitioningWorkerStepBuilderFactory}. + * @param jobRepository the job repository to use + * @param jobExplorer the job explorer to use + * @param transactionManager the transaction manager to use + */ + public RemotePartitioningWorkerStepBuilderFactory(JobRepository jobRepository, + JobExplorer jobExplorer, + PlatformTransactionManager transactionManager) { + + this.jobExplorer = jobExplorer; + this.jobRepository = jobRepository; + this.transactionManager = transactionManager; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + /** + * Creates a {@link RemotePartitioningWorkerStepBuilder} and initializes its job + * repository, job explorer, bean factory and transaction manager. + * @param name the name of the step + * @return a {@link RemotePartitioningWorkerStepBuilder} + */ + public RemotePartitioningWorkerStepBuilder get(String name) { + return new RemotePartitioningWorkerStepBuilder(name) + .repository(this.jobRepository) + .jobExplorer(this.jobExplorer) + .beanFactory(this.beanFactory) + .transactionManager(this.transactionManager); + } + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java new file mode 100644 index 000000000..c104d0476 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningMasterStepBuilderTests.java @@ -0,0 +1,246 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.runner.RunWith; +import org.mockito.Mockito; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.core.partition.support.StepExecutionAggregator; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.springframework.test.util.ReflectionTestUtils.getField; + +/** + * @author Mahmoud Ben Hassine + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = {RemotePartitioningMasterStepBuilderTests.BatchConfiguration.class}) +public class RemotePartitioningMasterStepBuilderTests { + + @Rule + public ExpectedException expectedException = ExpectedException.none(); + + @Autowired + private JobRepository jobRepository; + + @Test + public void inputChannelMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("inputChannel must not be null"); + + // when + new RemotePartitioningMasterStepBuilder("step").inputChannel(null); + + // then + // expected exception + } + + @Test + public void outputChannelMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("outputChannel must not be null"); + + // when + new RemotePartitioningMasterStepBuilder("step").outputChannel(null); + + // then + // expected exception + } + + @Test + public void messagingTemplateMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("messagingTemplate must not be null"); + + // when + new RemotePartitioningMasterStepBuilder("step").messagingTemplate(null); + + // then + // expected exception + } + + @Test + public void jobExplorerMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("jobExplorer must not be null"); + + // when + new RemotePartitioningMasterStepBuilder("step").jobExplorer(null); + + // then + // expected exception + } + + @Test + public void pollIntervalMustBeGreaterThanZero() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("The poll interval must be greater than zero"); + + // when + new RemotePartitioningMasterStepBuilder("step").pollInterval(-1); + + // then + // expected exception + } + + @Test + public void testMandatoryOutputChannel() { + // given + RemotePartitioningMasterStepBuilder builder = new RemotePartitioningMasterStepBuilder("step"); + + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("An outputChannel must be provided"); + + // when + Step step = builder.build(); + + // then + // expected exception + } + + @Test + public void testUnsupportedOperationExceptionWhenSpecifyingPartitionHandler() { + // given + PartitionHandler partitionHandler = Mockito.mock(PartitionHandler.class); + this.expectedException.expect(UnsupportedOperationException.class); + this.expectedException.expectMessage("When configuring a master step " + + "for remote partitioning using the RemotePartitioningMasterStepBuilder, " + + "the partition handler will be automatically set to an instance " + + "of MessageChannelPartitionHandler. The partition handler must " + + "not be provided in this case."); + + // when + new RemotePartitioningMasterStepBuilder("step").partitionHandler(partitionHandler); + + // then + // expected exception + } + + @Test + public void testMasterStepCreationWhenPollingRepository() { + // given + int gridSize = 5; + int startLimit = 3; + long timeout = 1000L; + long pollInterval = 5000L; + DirectChannel outputChannel = new DirectChannel(); + Partitioner partitioner = Mockito.mock(Partitioner.class); + StepExecutionAggregator stepExecutionAggregator = (result, executions) -> { }; + + // when + Step step = new RemotePartitioningMasterStepBuilder("masterStep") + .repository(jobRepository) + .outputChannel(outputChannel) + .partitioner("workerStep", partitioner) + .gridSize(gridSize) + .pollInterval(pollInterval) + .timeout(timeout) + .startLimit(startLimit) + .aggregator(stepExecutionAggregator) + .allowStartIfComplete(true) + .build(); + + // then + Assert.assertNotNull(step); + Assert.assertEquals(getField(step, "startLimit"), startLimit); + Assert.assertEquals(getField(step, "jobRepository"), this.jobRepository); + Assert.assertEquals(getField(step, "stepExecutionAggregator"), stepExecutionAggregator); + Assert.assertTrue((Boolean) getField(step, "allowStartIfComplete")); + + Object partitionHandler = getField(step, "partitionHandler"); + Assert.assertNotNull(partitionHandler); + Assert.assertTrue(partitionHandler instanceof MessageChannelPartitionHandler); + MessageChannelPartitionHandler messageChannelPartitionHandler = (MessageChannelPartitionHandler) partitionHandler; + Assert.assertEquals(getField(messageChannelPartitionHandler, "gridSize"), gridSize); + Assert.assertEquals(getField(messageChannelPartitionHandler, "pollInterval"), pollInterval); + Assert.assertEquals(getField(messageChannelPartitionHandler, "timeout"), timeout); + + Object messagingGateway = getField(messageChannelPartitionHandler, "messagingGateway"); + Assert.assertNotNull(messagingGateway); + MessagingTemplate messagingTemplate = (MessagingTemplate) messagingGateway; + Assert.assertEquals(getField(messagingTemplate, "defaultDestination"), outputChannel); + } + + @Test + public void testMasterStepCreationWhenAggregatingReplies() { + // given + int gridSize = 5; + int startLimit = 3; + DirectChannel outputChannel = new DirectChannel(); + Partitioner partitioner = Mockito.mock(Partitioner.class); + StepExecutionAggregator stepExecutionAggregator = (result, executions) -> { }; + + // when + Step step = new RemotePartitioningMasterStepBuilder("masterStep") + .repository(jobRepository) + .outputChannel(outputChannel) + .partitioner("workerStep", partitioner) + .gridSize(gridSize) + .startLimit(startLimit) + .aggregator(stepExecutionAggregator) + .allowStartIfComplete(true) + .build(); + + // then + Assert.assertNotNull(step); + Assert.assertEquals(getField(step, "startLimit"), startLimit); + Assert.assertEquals(getField(step, "jobRepository"), this.jobRepository); + Assert.assertEquals(getField(step, "stepExecutionAggregator"), stepExecutionAggregator); + Assert.assertTrue((Boolean) getField(step, "allowStartIfComplete")); + + Object partitionHandler = getField(step, "partitionHandler"); + Assert.assertNotNull(partitionHandler); + Assert.assertTrue(partitionHandler instanceof MessageChannelPartitionHandler); + MessageChannelPartitionHandler messageChannelPartitionHandler = (MessageChannelPartitionHandler) partitionHandler; + Assert.assertEquals(getField(messageChannelPartitionHandler, "gridSize"), gridSize); + + Object replyChannel = getField(messageChannelPartitionHandler, "replyChannel"); + Assert.assertNotNull(replyChannel); + Assert.assertTrue(replyChannel instanceof QueueChannel); + + Object messagingGateway = getField(messageChannelPartitionHandler, "messagingGateway"); + Assert.assertNotNull(messagingGateway); + MessagingTemplate messagingTemplate = (MessagingTemplate) messagingGateway; + Assert.assertEquals(getField(messagingTemplate, "defaultDestination"), outputChannel); + } + + @Configuration + @EnableBatchProcessing + public static class BatchConfiguration { + + } +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderTests.java new file mode 100644 index 000000000..1f478c0ae --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderTests.java @@ -0,0 +1,132 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.integration.partition; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.mockito.Mock; + +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.integration.channel.DirectChannel; + +/** + * @author Mahmoud Ben Hassine + */ +public class RemotePartitioningWorkerStepBuilderTests { + + @Rule + public ExpectedException expectedException = ExpectedException.none(); + + @Mock + private Tasklet tasklet; + + @Test + public void inputChannelMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("inputChannel must not be null"); + + // when + new RemotePartitioningWorkerStepBuilder("step").inputChannel(null); + + // then + // expected exception + } + + @Test + public void outputChannelMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("outputChannel must not be null"); + + // when + new RemotePartitioningWorkerStepBuilder("step").outputChannel(null); + + // then + // expected exception + } + + @Test + public void jobExplorerMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("jobExplorer must not be null"); + + // when + new RemotePartitioningWorkerStepBuilder("step").jobExplorer(null); + + // then + // expected exception + } + + @Test + public void stepLocatorMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("stepLocator must not be null"); + + // when + new RemotePartitioningWorkerStepBuilder("step").stepLocator(null); + + // then + // expected exception + } + + @Test + public void beanFactoryMustNotBeNull() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("beanFactory must not be null"); + + // when + new RemotePartitioningWorkerStepBuilder("step").beanFactory(null); + + // then + // expected exception + } + + @Test + public void testMandatoryInputChannel() { + // given + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("An InputChannel must be provided"); + + // when + new RemotePartitioningWorkerStepBuilder("step").tasklet(this.tasklet); + + // then + // expected exception + } + + @Test + public void testMandatoryJobExplorer() { + // given + DirectChannel inputChannel = new DirectChannel(); + this.expectedException.expect(IllegalArgumentException.class); + this.expectedException.expectMessage("A JobExplorer must be provided"); + + // when + new RemotePartitioningWorkerStepBuilder("step") + .inputChannel(inputChannel) + .tasklet(this.tasklet); + + // then + // expected exception + } + +} 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/MasterConfiguration.java index 48e8656b2..fb3b52f89 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/MasterConfiguration.java @@ -21,22 +21,15 @@ import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; -import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; -import org.springframework.batch.core.partition.PartitionHandler; -import org.springframework.batch.core.partition.support.Partitioner; -import org.springframework.batch.integration.partition.MessageChannelPartitionHandler; +import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; +import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory; import org.springframework.batch.sample.remotepartitioning.BasicPartitioner; import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.config.AggregatorFactoryBean; -import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.jms.dsl.Jms; @@ -49,24 +42,22 @@ import org.springframework.integration.jms.dsl.Jms; */ @Configuration @EnableBatchProcessing -@EnableIntegration +@EnableBatchIntegration @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) public class MasterConfiguration { private static final int GRID_SIZE = 3; - private static final long RECEIVE_TIMEOUT = 600000L; - private final JobBuilderFactory jobBuilderFactory; - private final StepBuilderFactory stepBuilderFactory; + private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory; public MasterConfiguration(JobBuilderFactory jobBuilderFactory, - StepBuilderFactory stepBuilderFactory) { + RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) { this.jobBuilderFactory = jobBuilderFactory; - this.stepBuilderFactory = stepBuilderFactory; + this.masterStepBuilderFactory = masterStepBuilderFactory; } /* @@ -89,64 +80,31 @@ public class MasterConfiguration { * Configure inbound flow (replies coming from workers) */ @Bean - public QueueChannel replies() { - return new QueueChannel(); - } - - @Bean - public DirectChannel inboundStaging() { + public DirectChannel replies() { return new DirectChannel(); } @Bean - public IntegrationFlow inboundStagingFlow(ActiveMQConnectionFactory connectionFactory) { + public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) { return IntegrationFlows .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies")) - .channel(inboundStaging()) + .channel(replies()) .get(); } /* - * Configure master step components + * Configure the master step */ @Bean public Step masterStep() { - return this.stepBuilderFactory.get("masterStep") - .partitioner("slaveStep", partitioner()) - .partitionHandler(partitionHandler()) + return this.masterStepBuilderFactory.get("masterStep") + .partitioner("workerStep", new BasicPartitioner()) .gridSize(GRID_SIZE) + .outputChannel(requests()) + .inputChannel(replies()) .build(); } - @Bean - public Partitioner partitioner() { - return new BasicPartitioner(); - } - - @Bean - public PartitionHandler partitionHandler() { - MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); - partitionHandler.setStepName("slaveStep"); - partitionHandler.setGridSize(GRID_SIZE); - partitionHandler.setReplyChannel(replies()); - - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultChannel(requests()); - template.setReceiveTimeout(RECEIVE_TIMEOUT); - partitionHandler.setMessagingOperations(template); - - return partitionHandler; - } - - @Bean - @ServiceActivator(inputChannel = "inboundStaging") - public AggregatorFactoryBean partitioningMessageHandler() { - AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); - aggregatorFactoryBean.setProcessorBean(partitionHandler()); - aggregatorFactoryBean.setOutputChannel(replies()); - return aggregatorFactoryBean; - } - @Bean public Job remotePartitioningJob() { return this.jobBuilderFactory.get("remotePartitioningJob") 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 ec19b1cad..25b202898 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 @@ -19,23 +19,18 @@ import org.apache.activemq.ActiveMQConnectionFactory; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; -import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; -import org.springframework.batch.core.explore.JobExplorer; import org.springframework.batch.core.step.tasklet.Tasklet; -import org.springframework.batch.integration.partition.BeanFactoryStepLocator; -import org.springframework.batch.integration.partition.StepExecutionRequestHandler; +import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; +import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.jms.dsl.Jms; @@ -48,24 +43,15 @@ import org.springframework.integration.jms.dsl.Jms; */ @Configuration @EnableBatchProcessing -@EnableIntegration +@EnableBatchIntegration @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) public class WorkerConfiguration { - private final StepBuilderFactory stepBuilderFactory; - - private final ApplicationContext applicationContext; - - private final JobExplorer jobExplorer; + private final RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory; - public WorkerConfiguration(StepBuilderFactory stepBuilderFactory, - JobExplorer jobExplorer, - ApplicationContext applicationContext) { - - this.stepBuilderFactory = stepBuilderFactory; - this.applicationContext = applicationContext; - this.jobExplorer = jobExplorer; + public WorkerConfiguration(RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory) { + this.workerStepBuilderFactory = workerStepBuilderFactory; } /* @@ -101,29 +87,20 @@ public class WorkerConfiguration { } /* - * Configure worker components + * Configure the worker step */ @Bean - @ServiceActivator(inputChannel = "requests", outputChannel = "replies") - public StepExecutionRequestHandler stepExecutionRequestHandler() { - StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); - stepExecutionRequestHandler.setJobExplorer(this.jobExplorer); - BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator(); - stepLocator.setBeanFactory(this.applicationContext); - stepExecutionRequestHandler.setStepLocator(stepLocator); - return stepExecutionRequestHandler; - } - - @Bean - public Step slaveStep() { - return this.stepBuilderFactory.get("slaveStep") - .tasklet(getTasklet(null)) + public Step workerStep() { + return this.workerStepBuilderFactory.get("workerStep") + .inputChannel(requests()) + .outputChannel(replies()) + .tasklet(tasklet(null)) .build(); } @Bean @StepScope - public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) { + public Tasklet tasklet(@Value("#{stepExecutionContext['partition']}") String partition) { return (contribution, chunkContext) -> { System.out.println("processing " + partition); return RepeatStatus.FINISHED; 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/MasterConfiguration.java index d5cf60ee1..65c58e31e 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/MasterConfiguration.java @@ -15,18 +15,14 @@ */ package org.springframework.batch.sample.remotepartitioning.polling; -import javax.sql.DataSource; - import org.apache.activemq.ActiveMQConnectionFactory; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; -import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; -import org.springframework.batch.core.partition.PartitionHandler; -import org.springframework.batch.core.partition.support.Partitioner; -import org.springframework.batch.integration.partition.MessageChannelPartitionHandler; +import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; +import org.springframework.batch.integration.partition.RemotePartitioningMasterStepBuilderFactory; import org.springframework.batch.sample.remotepartitioning.BasicPartitioner; import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; @@ -34,8 +30,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.jms.dsl.Jms; @@ -48,26 +42,22 @@ import org.springframework.integration.jms.dsl.Jms; */ @Configuration @EnableBatchProcessing -@EnableIntegration +@EnableBatchIntegration @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) public class MasterConfiguration { private static final int GRID_SIZE = 3; - private static final long RECEIVE_TIMEOUT = 600000L; - - private static final long POLL_INTERVAL = 2000L; - private final JobBuilderFactory jobBuilderFactory; - private final StepBuilderFactory stepBuilderFactory; + private final RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory; public MasterConfiguration(JobBuilderFactory jobBuilderFactory, - StepBuilderFactory stepBuilderFactory) { + RemotePartitioningMasterStepBuilderFactory masterStepBuilderFactory) { this.jobBuilderFactory = jobBuilderFactory; - this.stepBuilderFactory = stepBuilderFactory; + this.masterStepBuilderFactory = masterStepBuilderFactory; } /* @@ -87,38 +77,17 @@ public class MasterConfiguration { } /* - * Configure master step components + * Configure the master step */ @Bean public Step masterStep() { - return this.stepBuilderFactory.get("masterStep") - .partitioner("slaveStep", partitioner()) - .partitionHandler(partitionHandler(null)) + return this.masterStepBuilderFactory.get("masterStep") + .partitioner("workerStep", new BasicPartitioner()) .gridSize(GRID_SIZE) + .outputChannel(requests()) .build(); } - @Bean - public Partitioner partitioner() { - return new BasicPartitioner(); - } - - @Bean - public PartitionHandler partitionHandler(DataSource dataSource) { - MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); - partitionHandler.setStepName("slaveStep"); - partitionHandler.setGridSize(GRID_SIZE); - partitionHandler.setDataSource(dataSource); - partitionHandler.setPollInterval(POLL_INTERVAL); - - MessagingTemplate template = new MessagingTemplate(); - template.setDefaultChannel(requests()); - template.setReceiveTimeout(RECEIVE_TIMEOUT); - partitionHandler.setMessagingOperations(template); - - return partitionHandler; - } - @Bean public Job remotePartitioningJob() { return this.jobBuilderFactory.get("remotePartitioningJob") diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java index 693859c4a..932b7ef21 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java @@ -19,24 +19,18 @@ import org.apache.activemq.ActiveMQConnectionFactory; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; -import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; -import org.springframework.batch.core.explore.JobExplorer; import org.springframework.batch.core.step.tasklet.Tasklet; -import org.springframework.batch.integration.partition.BeanFactoryStepLocator; -import org.springframework.batch.integration.partition.StepExecutionRequestHandler; +import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; +import org.springframework.batch.integration.partition.RemotePartitioningWorkerStepBuilderFactory; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.NullChannel; -import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.jms.dsl.Jms; @@ -49,24 +43,15 @@ import org.springframework.integration.jms.dsl.Jms; */ @Configuration @EnableBatchProcessing -@EnableIntegration +@EnableBatchIntegration @Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) public class WorkerConfiguration { - private final StepBuilderFactory stepBuilderFactory; - - private final ApplicationContext applicationContext; - - private final JobExplorer jobExplorer; + private final RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory; - public WorkerConfiguration(StepBuilderFactory stepBuilderFactory, - JobExplorer jobExplorer, - ApplicationContext applicationContext) { - - this.stepBuilderFactory = stepBuilderFactory; - this.applicationContext = applicationContext; - this.jobExplorer = jobExplorer; + public WorkerConfiguration(RemotePartitioningWorkerStepBuilderFactory workerStepBuilderFactory) { + this.workerStepBuilderFactory = workerStepBuilderFactory; } /* @@ -86,37 +71,19 @@ public class WorkerConfiguration { } /* - * Configure outbound flow (replies going to the master) + * Configure the worker step */ @Bean - public NullChannel replies() { - return new NullChannel(); // replies are discarded (since the master is polling the job repository) - } - - /* - * Configure worker components - */ - @Bean - @ServiceActivator(inputChannel = "requests", outputChannel = "replies") - public StepExecutionRequestHandler stepExecutionRequestHandler() { - StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); - stepExecutionRequestHandler.setJobExplorer(this.jobExplorer); - BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator(); - stepLocator.setBeanFactory(this.applicationContext); - stepExecutionRequestHandler.setStepLocator(stepLocator); - return stepExecutionRequestHandler; - } - - @Bean - public Step slaveStep() { - return this.stepBuilderFactory.get("slaveStep") - .tasklet(getTasklet(null)) + public Step workerStep() { + return this.workerStepBuilderFactory.get("workerStep") + .inputChannel(requests()) + .tasklet(tasklet(null)) .build(); } @Bean @StepScope - public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) { + public Tasklet tasklet(@Value("#{stepExecutionContext['partition']}") String partition) { return (contribution, chunkContext) -> { System.out.println("processing " + partition); return RepeatStatus.FINISHED;