From 2c5b7b59e84384b3ea3520a1e01cc46e59a7b3ca Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Fri, 8 Jun 2018 06:41:04 +0200 Subject: [PATCH] Add remote partitioning samples This commit also fixes the code examples in the remote partitioning section of the documentation Resolves BATCH-2730 --- .../asciidoc/spring-batch-integration.adoc | 107 ++++++------ spring-batch-samples/README.md | 16 +- .../remotepartitioning/BasicPartitioner.java | 43 +++++ .../BrokerConfiguration.java | 44 +++++ .../DataSourceConfiguration.java | 53 ++++++ .../aggregating/MasterConfiguration.java | 157 ++++++++++++++++++ .../aggregating/WorkerConfiguration.java | 133 +++++++++++++++ .../polling/MasterConfiguration.java | 129 ++++++++++++++ .../polling/WorkerConfiguration.java | 126 ++++++++++++++ .../resources/remote-partitioning.properties | 4 + .../RemotePartitioningJobFunctionalTests.java | 91 ++++++++++ ...WithMessageAggregationFunctionalTests.java | 37 +++++ ...bWithRepositoryPollingFunctionalTests.java | 37 +++++ 13 files changed, 928 insertions(+), 49 deletions(-) create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/WorkerConfiguration.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java create mode 100644 spring-batch-samples/src/main/resources/remote-partitioning.properties create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithMessageAggregationFunctionalTests.java create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithRepositoryPollingFunctionalTests.java diff --git a/spring-batch-docs/asciidoc/spring-batch-integration.adoc b/spring-batch-docs/asciidoc/spring-batch-integration.adoc index 5eba5b80f..752552f90 100644 --- a/spring-batch-docs/asciidoc/spring-batch-integration.adoc +++ b/spring-batch-docs/asciidoc/spring-batch-integration.adoc @@ -1051,6 +1051,10 @@ configuration: .Java Configuration [source, java, role="javaContent"] ---- + +/* + * Configuration of the master side + */ @Bean public PartitionHandler partitionHandler() { MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); @@ -1064,6 +1068,11 @@ public PartitionHandler partitionHandler() { return partitionHandler; } +@Bean +public QueueChannel outboundReplies() { + return new QueueChannel(); +} + @Bean public DirectChannel outboundRequests() { return new DirectChannel(); @@ -1078,44 +1087,13 @@ public IntegrationFlow outboundJmsRequests() { } @Bean -public DirectChannel inboundRequests() { - return new DirectChannel(); -} - -public IntegrationFlow inboundJmsRequests() { - return IntegrationFlows - .from(Jms.messageDrivenChannelAdapter(connectionFactory()) - .configureListenerContainer(c -> c.subscriptionDurable(false)) - .destination("requestsQueue")) - .channel(inboundRequests()) - .get(); - } - -@Bean -public StepExecutionRequestHandler stepExecutionRequestHandler() { - StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); - stepExecutionRequestHandler.setJobExplorer(jobExplorer); - stepExecutionRequestHandler.setStepLocator(stepLocator()); - return stepExecutionRequestHandler; -} - -@Bean -@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging") -public StepExecutionRequestHandler serviceActivator() throws Exception { - return stepExecutionRequestHandler(); -} - -@Bean -public DirectChannel outboundStaging() { - return new DirectChannel(); -} - -@Bean -public IntegrationFlow outboundJmsStaging() { - return IntegrationFlows.from("outboundStaging") - .handle(Jms.outboundGateway(connectionFactory()) - .requestDestination("stagingQueue")) - .get(); +@ServiceActivator(inputChannel = "inboundStaging") +public AggregatorFactoryBean partitioningMessageHandler() throws Exception { + AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); + aggregatorFactoryBean.setProcessorBean(partitionHandler()); + aggregatorFactoryBean.setOutputChannel(outboundReplies()); + // configure other propeties of the aggregatorFactoryBean + return aggregatorFactoryBean; } @Bean @@ -1133,19 +1111,48 @@ public IntegrationFlow inboundJmsStaging() { .get(); } +/* + * Configuration of the worker side + */ @Bean -@ServiceActivator(inputChannel = "inboundStaging") -public AggregatorFactoryBean partitioningMessageHandler() throws Exception { - AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); - aggregatorFactoryBean.setProcessorBean(partitionHandler()); - aggregatorFactoryBean.setOutputChannel(outboundReplies()); - ... - return aggregatorFactoryBean; +public StepExecutionRequestHandler stepExecutionRequestHandler() { + StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); + stepExecutionRequestHandler.setJobExplorer(jobExplorer); + stepExecutionRequestHandler.setStepLocator(stepLocator()); + return stepExecutionRequestHandler; } @Bean -public QueueChannel outboundReplies() { - return new QueueChannel(); +@ServiceActivator(inputChannel = "inboundRequests", outputChannel = "outboundStaging") +public StepExecutionRequestHandler serviceActivator() throws Exception { + return stepExecutionRequestHandler(); +} + +@Bean +public DirectChannel inboundRequests() { + return new DirectChannel(); +} + +public IntegrationFlow inboundJmsRequests() { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory()) + .configureListenerContainer(c -> c.subscriptionDurable(false)) + .destination("requestsQueue")) + .channel(inboundRequests()) + .get(); +} + +@Bean +public DirectChannel outboundStaging() { + return new DirectChannel(); +} + +@Bean +public IntegrationFlow outboundJmsStaging() { + return IntegrationFlows.from("outboundStaging") + .handle(Jms.outboundGateway(connectionFactory()) + .requestDestination("stagingQueue")) + .get(); } ---- @@ -1168,8 +1175,12 @@ You must also ensure that the partition `handler` attribute maps to the `partiti public Job personJob() { return jobBuilderFactory.get("personJob") .start(stepBuilderFactory.get("step1.master") - .partitioner(partitionHandler()) + .partitioner("step1.worker", partitioner()) + .partitionHandler(partitionHandler()) .build()) .build(); } ---- + +You can find a complete example of a remote partitioning job +link:$$https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#remote-partitioning-sample$$[here]. diff --git a/spring-batch-samples/README.md b/spring-batch-samples/README.md index 9daf02c45..7bcc4bc9e 100644 --- a/spring-batch-samples/README.md +++ b/spring-batch-samples/README.md @@ -638,13 +638,27 @@ happens inside the main step transaction). The purpose of this sample is to show multi-threaded step execution using the `PartitionHandler` SPI. The example uses a `TaskExecutorPartitionHandler` to spread the work of reading -some files acrosss multiple threads, with one `Step` execution +some files across multiple threads, with one `Step` execution per thread. The key components are the `PartitionStep` and the `MultiResourcePartitioner` which is responsible for dividing up the work. Notice that the readers and writers in the `Step` that is being partitioned are step-scoped, so that their state does not get shared across threads of execution. +### [Remote Partitioning Sample](id:remotePartitioning) + +This sample shows how to configure a remote partitioning job. The master step +uses a `MessageChannelPartitionHandler` to send partitions to and receive +replies from workers. Two examples are shown: + +* A master step that polls the job repository to see if all workers have finished +their work +* A master step that aggregates replies from workers to notify work completion + +The sample uses an embedded JMS broker and an embedded database for simplicity +but any option supported via Spring Integration for communication is technically +acceptable. + ### [Remote Chunking Sample](id:remoteChunking) This sample shows how to configure a remote chunking job. The master step will diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java new file mode 100644 index 000000000..fbf28b021 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BasicPartitioner.java @@ -0,0 +1,43 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.sample.remotepartitioning; + +import java.util.Map; + +import org.springframework.batch.core.partition.support.SimplePartitioner; +import org.springframework.batch.item.ExecutionContext; + +/** + * Simple partitioner for demonstration purpose. + * + * @author Mahmoud Ben Hassine + */ +public class BasicPartitioner extends SimplePartitioner { + + private static final String PARTITION_KEY = "partition"; + + @Override + public Map partition(int gridSize) { + Map partitions = super.partition(gridSize); + int i = 0; + for (ExecutionContext context : partitions.values()) { + context.put(PARTITION_KEY, PARTITION_KEY + (i++)); + } + return partitions; + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java new file mode 100644 index 000000000..372512a0f --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/BrokerConfiguration.java @@ -0,0 +1,44 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.sample.remotepartitioning; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.PropertySource; + +/** + * @author Mahmoud Ben Hassine + */ +@Configuration +@PropertySource("classpath:remote-partitioning.properties") +public class BrokerConfiguration { + + @Value("${broker.url}") + private String brokerUrl; + + @Bean + public ActiveMQConnectionFactory connectionFactory() { + ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(); + connectionFactory.setBrokerURL(this.brokerUrl); + connectionFactory.setTrustAllPackages(true); + return connectionFactory; + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java new file mode 100644 index 000000000..bb30a4988 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/DataSourceConfiguration.java @@ -0,0 +1,53 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.sample.remotepartitioning; + +import javax.sql.DataSource; + +import org.apache.commons.dbcp2.BasicDataSource; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.PropertySource; + +/** + * @author Mahmoud Ben Hassine + */ +@Configuration +@PropertySource("classpath:remote-partitioning.properties") +public class DataSourceConfiguration { + + @Value("${datasource.url}") + private String url; + + @Value("${datasource.username}") + private String username; + + @Value("${datasource.password}") + private String password; + + @Bean + public DataSource dataSource() { + BasicDataSource dataSource = new BasicDataSource(); + dataSource.setUrl(url); + dataSource.setUsername(username); + dataSource.setPassword(password); + return dataSource; + } + +} 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 new file mode 100644 index 000000000..48e8656b2 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/MasterConfiguration.java @@ -0,0 +1,157 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample.remotepartitioning.aggregating; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.integration.partition.MessageChannelPartitionHandler; +import org.springframework.batch.sample.remotepartitioning.BasicPartitioner; +import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; +import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.AggregatorFactoryBean; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.jms.dsl.Jms; + +/** + * This configuration class is for the master side of the remote partitioning sample. + * The master step will create 3 partitions for workers to process. + * + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +@EnableIntegration +@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) +public class MasterConfiguration { + + private static final int GRID_SIZE = 3; + + private static final long RECEIVE_TIMEOUT = 600000L; + + private final JobBuilderFactory jobBuilderFactory; + + private final StepBuilderFactory stepBuilderFactory; + + + public MasterConfiguration(JobBuilderFactory jobBuilderFactory, + StepBuilderFactory stepBuilderFactory) { + + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + } + + /* + * Configure outbound flow (requests going to workers) + */ + @Bean + public DirectChannel requests() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(requests()) + .handle(Jms.outboundAdapter(connectionFactory).destination("requests")) + .get(); + } + + /* + * Configure inbound flow (replies coming from workers) + */ + @Bean + public QueueChannel replies() { + return new QueueChannel(); + } + + @Bean + public DirectChannel inboundStaging() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow inboundStagingFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies")) + .channel(inboundStaging()) + .get(); + } + + /* + * Configure master step components + */ + @Bean + public Step masterStep() { + return this.stepBuilderFactory.get("masterStep") + .partitioner("slaveStep", partitioner()) + .partitionHandler(partitionHandler()) + .gridSize(GRID_SIZE) + .build(); + } + + @Bean + public Partitioner partitioner() { + return new BasicPartitioner(); + } + + @Bean + public PartitionHandler partitionHandler() { + MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); + partitionHandler.setStepName("slaveStep"); + partitionHandler.setGridSize(GRID_SIZE); + partitionHandler.setReplyChannel(replies()); + + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(requests()); + template.setReceiveTimeout(RECEIVE_TIMEOUT); + partitionHandler.setMessagingOperations(template); + + return partitionHandler; + } + + @Bean + @ServiceActivator(inputChannel = "inboundStaging") + public AggregatorFactoryBean partitioningMessageHandler() { + AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); + aggregatorFactoryBean.setProcessorBean(partitionHandler()); + aggregatorFactoryBean.setOutputChannel(replies()); + return aggregatorFactoryBean; + } + + @Bean + public Job remotePartitioningJob() { + return this.jobBuilderFactory.get("remotePartitioningJob") + .start(masterStep()) + .build(); + } + +} 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 new file mode 100644 index 000000000..ec19b1cad --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/aggregating/WorkerConfiguration.java @@ -0,0 +1,133 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample.remotepartitioning.aggregating; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepScope; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.integration.partition.BeanFactoryStepLocator; +import org.springframework.batch.integration.partition.StepExecutionRequestHandler; +import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; +import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.jms.dsl.Jms; + +/** + * This configuration class is for the worker side of the remote partitioning sample. + * Each worker will process a partition sent by the master step. + * + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +@EnableIntegration +@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) +public class WorkerConfiguration { + + private final StepBuilderFactory stepBuilderFactory; + + private final ApplicationContext applicationContext; + + private final JobExplorer jobExplorer; + + + public WorkerConfiguration(StepBuilderFactory stepBuilderFactory, + JobExplorer jobExplorer, + ApplicationContext applicationContext) { + + this.stepBuilderFactory = stepBuilderFactory; + this.applicationContext = applicationContext; + this.jobExplorer = jobExplorer; + } + + /* + * Configure inbound flow (requests coming from the master) + */ + @Bean + public DirectChannel requests() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests")) + .channel(requests()) + .get(); + } + + /* + * Configure outbound flow (replies going to the master) + */ + @Bean + public DirectChannel replies() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(replies()) + .handle(Jms.outboundAdapter(connectionFactory).destination("replies")) + .get(); + } + + /* + * Configure worker components + */ + @Bean + @ServiceActivator(inputChannel = "requests", outputChannel = "replies") + public StepExecutionRequestHandler stepExecutionRequestHandler() { + StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); + stepExecutionRequestHandler.setJobExplorer(this.jobExplorer); + BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator(); + stepLocator.setBeanFactory(this.applicationContext); + stepExecutionRequestHandler.setStepLocator(stepLocator); + return stepExecutionRequestHandler; + } + + @Bean + public Step slaveStep() { + return this.stepBuilderFactory.get("slaveStep") + .tasklet(getTasklet(null)) + .build(); + } + + @Bean + @StepScope + public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) { + return (contribution, chunkContext) -> { + System.out.println("processing " + partition); + return RepeatStatus.FINISHED; + }; + } + +} 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 new file mode 100644 index 000000000..d5cf60ee1 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/MasterConfiguration.java @@ -0,0 +1,129 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample.remotepartitioning.polling; + +import javax.sql.DataSource; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.support.Partitioner; +import org.springframework.batch.integration.partition.MessageChannelPartitionHandler; +import org.springframework.batch.sample.remotepartitioning.BasicPartitioner; +import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; +import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.jms.dsl.Jms; + +/** + * This configuration class is for the master side of the remote partitioning sample. + * The master step will create 3 partitions for workers to process. + * + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +@EnableIntegration +@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) +public class MasterConfiguration { + + private static final int GRID_SIZE = 3; + + private static final long RECEIVE_TIMEOUT = 600000L; + + private static final long POLL_INTERVAL = 2000L; + + private final JobBuilderFactory jobBuilderFactory; + + private final StepBuilderFactory stepBuilderFactory; + + + public MasterConfiguration(JobBuilderFactory jobBuilderFactory, + StepBuilderFactory stepBuilderFactory) { + + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + } + + /* + * Configure outbound flow (requests going to workers) + */ + @Bean + public DirectChannel requests() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(requests()) + .handle(Jms.outboundAdapter(connectionFactory).destination("requests")) + .get(); + } + + /* + * Configure master step components + */ + @Bean + public Step masterStep() { + return this.stepBuilderFactory.get("masterStep") + .partitioner("slaveStep", partitioner()) + .partitionHandler(partitionHandler(null)) + .gridSize(GRID_SIZE) + .build(); + } + + @Bean + public Partitioner partitioner() { + return new BasicPartitioner(); + } + + @Bean + public PartitionHandler partitionHandler(DataSource dataSource) { + MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler(); + partitionHandler.setStepName("slaveStep"); + partitionHandler.setGridSize(GRID_SIZE); + partitionHandler.setDataSource(dataSource); + partitionHandler.setPollInterval(POLL_INTERVAL); + + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(requests()); + template.setReceiveTimeout(RECEIVE_TIMEOUT); + partitionHandler.setMessagingOperations(template); + + return partitionHandler; + } + + @Bean + public Job remotePartitioningJob() { + return this.jobBuilderFactory.get("remotePartitioningJob") + .start(masterStep()) + .build(); + } + +} 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 new file mode 100644 index 000000000..693859c4a --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/remotepartitioning/polling/WorkerConfiguration.java @@ -0,0 +1,126 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample.remotepartitioning.polling; + +import org.apache.activemq.ActiveMQConnectionFactory; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepScope; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.integration.partition.BeanFactoryStepLocator; +import org.springframework.batch.integration.partition.StepExecutionRequestHandler; +import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.batch.sample.remotepartitioning.BrokerConfiguration; +import org.springframework.batch.sample.remotepartitioning.DataSourceConfiguration; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.jms.dsl.Jms; + +/** + * This configuration class is for the worker side of the remote partitioning sample. + * Each worker will process a partition sent by the master step. + * + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +@EnableIntegration +@Import(value = {DataSourceConfiguration.class, BrokerConfiguration.class}) +public class WorkerConfiguration { + + private final StepBuilderFactory stepBuilderFactory; + + private final ApplicationContext applicationContext; + + private final JobExplorer jobExplorer; + + + public WorkerConfiguration(StepBuilderFactory stepBuilderFactory, + JobExplorer jobExplorer, + ApplicationContext applicationContext) { + + this.stepBuilderFactory = stepBuilderFactory; + this.applicationContext = applicationContext; + this.jobExplorer = jobExplorer; + } + + /* + * Configure inbound flow (requests coming from the master) + */ + @Bean + public DirectChannel requests() { + return new DirectChannel(); + } + + @Bean + public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) { + return IntegrationFlows + .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests")) + .channel(requests()) + .get(); + } + + /* + * Configure outbound flow (replies going to the master) + */ + @Bean + public NullChannel replies() { + return new NullChannel(); // replies are discarded (since the master is polling the job repository) + } + + /* + * Configure worker components + */ + @Bean + @ServiceActivator(inputChannel = "requests", outputChannel = "replies") + public StepExecutionRequestHandler stepExecutionRequestHandler() { + StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler(); + stepExecutionRequestHandler.setJobExplorer(this.jobExplorer); + BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator(); + stepLocator.setBeanFactory(this.applicationContext); + stepExecutionRequestHandler.setStepLocator(stepLocator); + return stepExecutionRequestHandler; + } + + @Bean + public Step slaveStep() { + return this.stepBuilderFactory.get("slaveStep") + .tasklet(getTasklet(null)) + .build(); + } + + @Bean + @StepScope + public Tasklet getTasklet(@Value("#{stepExecutionContext['partition']}") String partition) { + return (contribution, chunkContext) -> { + System.out.println("processing " + partition); + return RepeatStatus.FINISHED; + }; + } + +} diff --git a/spring-batch-samples/src/main/resources/remote-partitioning.properties b/spring-batch-samples/src/main/resources/remote-partitioning.properties new file mode 100644 index 000000000..ed040df33 --- /dev/null +++ b/spring-batch-samples/src/main/resources/remote-partitioning.properties @@ -0,0 +1,4 @@ +broker.url=tcp://localhost:61617 +datasource.url=jdbc:hsqldb:mem:testdb +datasource.username=sa +datasource.password= diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java new file mode 100644 index 000000000..9631f7e66 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobFunctionalTests.java @@ -0,0 +1,91 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample; + +import org.apache.activemq.broker.BrokerService; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.test.JobLauncherTestUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.PropertySource; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * Base class for remote partitioning tests. + * + * @author Mahmoud Ben Hassine + */ +@RunWith(SpringRunner.class) +@PropertySource("classpath:remote-partitioning.properties") +public abstract class RemotePartitioningJobFunctionalTests { + + private static final String BROKER_DATA_DIRECTORY = "build/activemq-data"; + + @Value("${broker.url}") + private String brokerUrl; + + @Autowired + protected JobLauncherTestUtils jobLauncherTestUtils; + + private BrokerService brokerService; + + private EmbeddedDatabase embeddedDatabase; + + private AnnotationConfigApplicationContext workerApplicationContext; + + protected abstract Class getWorkerConfigurationClass(); + + @Before + public void setUp() throws Exception { + this.brokerService = new BrokerService(); + this.brokerService.addConnector(this.brokerUrl); + this.brokerService.setDataDirectory(BROKER_DATA_DIRECTORY); + this.brokerService.start(); + this.embeddedDatabase = new EmbeddedDatabaseBuilder() + .addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql") + .addScript("/org/springframework/batch/core/schema-hsqldb.sql") + .build(); + this.workerApplicationContext = new AnnotationConfigApplicationContext(getWorkerConfigurationClass()); + } + + @Test + public void testRemotePartitioningJob() throws Exception { + // when + JobExecution jobExecution = this.jobLauncherTestUtils.launchJob(); + + // then + Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + Assert.assertEquals(4, jobExecution.getStepExecutions().size()); // master + 3 workers + } + + @After + public void tearDown() throws Exception { + this.workerApplicationContext.close(); + this.brokerService.stop(); + this.embeddedDatabase.shutdown(); + } + +} 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 new file mode 100644 index 000000000..0bff862d2 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithMessageAggregationFunctionalTests.java @@ -0,0 +1,37 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample; + +import org.springframework.batch.sample.config.JobRunnerConfiguration; +import org.springframework.batch.sample.remotepartitioning.aggregating.MasterConfiguration; +import org.springframework.batch.sample.remotepartitioning.aggregating.WorkerConfiguration; +import org.springframework.test.context.ContextConfiguration; + +/** + * The master step of the job under test will create 3 partitions for workers + * to process. + * + * @author Mahmoud Ben Hassine + */ +@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class}) +public class RemotePartitioningJobWithMessageAggregationFunctionalTests extends RemotePartitioningJobFunctionalTests { + + @Override + protected Class getWorkerConfigurationClass() { + return WorkerConfiguration.class; + } + +} 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 new file mode 100644 index 000000000..44dc1a401 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/RemotePartitioningJobWithRepositoryPollingFunctionalTests.java @@ -0,0 +1,37 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.sample; + +import org.springframework.batch.sample.config.JobRunnerConfiguration; +import org.springframework.batch.sample.remotepartitioning.polling.MasterConfiguration; +import org.springframework.batch.sample.remotepartitioning.polling.WorkerConfiguration; +import org.springframework.test.context.ContextConfiguration; + +/** + * The master step of the job under test will create 3 partitions for workers + * to process. + * + * @author Mahmoud Ben Hassine + */ +@ContextConfiguration(classes = {JobRunnerConfiguration.class, MasterConfiguration.class}) +public class RemotePartitioningJobWithRepositoryPollingFunctionalTests extends RemotePartitioningJobFunctionalTests { + + @Override + protected Class getWorkerConfigurationClass() { + return WorkerConfiguration.class; + } + +}