From 102f614874a4abdfe39fcb126bfa15ab44974207 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Tue, 8 Oct 2013 15:08:53 -0500 Subject: [PATCH] BATCH-2003: Implemented: * Basic partitioning support per JSR-352 * Parameter injection at the partition level * PartitionCollector/PartitionAnalyzer functionality * PartitionReducer functionality --- .../xml/StepParserStepFactoryBean.java | 98 ++++- .../BatchPropertyBeanPostProcessor.java | 3 +- .../support/BatchPropertyContext.java | 47 ++- .../jsr/configuration/xml/ListenerParser.java | 2 +- .../configuration/xml/PartitionParser.java | 155 ++++++++ .../jsr/configuration/xml/PropertyParser.java | 32 +- .../configuration/xml/StepFactoryBean.java | 120 +++++- .../jsr/configuration/xml/StepParser.java | 5 + .../batch/core/jsr/launch/JsrJobOperator.java | 4 +- .../jsr/partition/JsrPartitionHandler.java | 267 +++++++++++++ .../partition/PartitionCollectorAdapter.java | 79 ++++ .../jsr/partition/PartitionException.java | 36 ++ .../batch/core/jsr/step/BatchletStep.java | 50 +++ .../batch/core/jsr/step/PartitionStep.java | 84 ++++ .../step/builder/JsrBatchletStepBuilder.java | 102 +++++ .../builder/JsrFaultTolerantStepBuilder.java | 71 +++- .../step/builder/JsrPartitionStepBuilder.java | 129 +++++++ .../step/builder/JsrSimpleStepBuilder.java | 71 ++++ .../partition/JsrStepExecutionSplitter.java | 87 +++++ .../core/partition/support/PartitionStep.java | 8 +- .../batch/core/scope/context/StepContext.java | 55 ++- .../context/StepSynchronizationManager.java | 36 +- .../batch/core/step/AbstractStep.java | 22 +- .../builder/AbstractTaskletStepBuilder.java | 53 ++- .../builder/FaultTolerantStepBuilder.java | 3 +- .../step/builder/PartitionStepBuilder.java | 54 ++- .../core/step/builder/SimpleStepBuilder.java | 5 +- .../support/BatchPropertyContextTests.java | 44 ++- .../xml/PartitionParserTests.java | 365 ++++++++++++++++++ .../core/jsr/launch/JsrJobOperatorTests.java | 44 ++- .../partition/JsrPartitionHandlerTests.java | 218 +++++++++++ .../JsrStepExecutionSplitterTests.java | 54 +++ .../PartitionCollectorAdapterTests.java | 99 +++++ .../jsr/step/item/JsrChunkProcessorTests.java | 2 + .../JsrFaultTolerantChunkProcessorTests.java | 2 + .../core/scope/context/StepContextTests.java | 23 ++ .../StepSynchronizationManagerTests.java | 27 ++ .../batch-jobs/fullPartitionParserTests.xml | 12 + ...tionParserWithHardcodedPropertiesTests.xml | 41 ++ ...rtitionParserWithMapperPropertiesTests.xml | 28 ++ ...fullPartitionParserWithPropertiesTests.xml | 32 ++ .../partitionParserTestsBatchlet.xml | 9 + .../batch-jobs/partitionParserTestsChunk.xml | 12 + 43 files changed, 2592 insertions(+), 98 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/BatchletStep.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrBatchletStepBuilder.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrPartitionStepBuilder.java create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/partition/JsrStepExecutionSplitter.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrStepExecutionSplitterTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserTests.xml create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithHardcodedPropertiesTests.xml create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithMapperPropertiesTests.xml create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithPropertiesTests.xml create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsBatchlet.xml create mode 100644 spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsChunk.xml diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 53550687a..489b4c4dc 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -16,11 +16,13 @@ package org.springframework.batch.core.configuration.xml; +import java.io.Serializable; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; import java.util.Map; +import java.util.Queue; import java.util.Set; import javax.batch.api.chunk.listener.RetryProcessListener; @@ -29,6 +31,8 @@ import javax.batch.api.chunk.listener.RetryWriteListener; import javax.batch.api.chunk.listener.SkipProcessListener; import javax.batch.api.chunk.listener.SkipReadListener; import javax.batch.api.chunk.listener.SkipWriteListener; +import javax.batch.api.partition.PartitionAnalyzer; +import javax.batch.api.partition.PartitionCollector; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemProcessListener; @@ -49,6 +53,7 @@ import org.springframework.batch.core.jsr.RetryReadListenerAdapter; import org.springframework.batch.core.jsr.RetryWriteListenerAdapter; import org.springframework.batch.core.jsr.SkipListenerAdapter; import org.springframework.batch.core.jsr.StepListenerAdapter; +import org.springframework.batch.core.jsr.partition.PartitionCollectorAdapter; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.partition.PartitionHandler; import org.springframework.batch.core.partition.support.Partitioner; @@ -158,6 +163,8 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa private int gridSize = DEFAULT_GRID_SIZE; + private Queue partitionQueue; + // // Tasklet Elements // @@ -238,13 +245,20 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa private StepExecutionAggregator stepExecutionAggregator; + /** + * @param queue The {@link Queue} that is used for communication between {@link PartitionCollector} and {@link PartitionAnalyzer} + */ + public void setPartitionQueue(Queue queue) { + this.partitionQueue = queue; + } + /** * Create a {@link Step} from the configuration provided. * * @see FactoryBean#getObject() */ @Override - public final Object getObject() throws Exception { + public Object getObject() throws Exception { if (hasChunkElement) { Assert.isNull(tasklet, "Step [" + name + "] has both a element and a 'ref' attribute referencing a Tasklet."); @@ -278,7 +292,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa return hasChunkElement || tasklet != null; } - private void enhanceCommonStep(StepBuilderHelper builder) { + /** + * @param builder + */ + protected void enhanceCommonStep(StepBuilderHelper builder) { if (allowStartIfComplete != null) { builder.allowStartIfComplete(allowStartIfComplete); } @@ -290,13 +307,13 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa for (Object listener : stepExecutionListeners) { if(listener instanceof StepExecutionListener) { builder.listener((StepExecutionListener) listener); - } else if(listener instanceof StepListener) { + } else if(listener instanceof javax.batch.api.listener.StepListener) { builder.listener(new StepListenerAdapter((javax.batch.api.listener.StepListener) listener)); } } } - private Step createPartitionStep() { + protected Step createPartitionStep() { PartitionStepBuilder builder; if (partitioner != null) { @@ -321,7 +338,7 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa } - private Step createFaultTolerantStep() { + protected Step createFaultTolerantStep() { FaultTolerantStepBuilder builder = getFaultTolerantStepBuilder(this.name); @@ -417,7 +434,7 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa } @SuppressWarnings("unchecked") - private Step createSimpleStep() { + protected Step createSimpleStep() { SimpleStepBuilder builder = getSimpleStepBuilder(this.name); if(timeout != null && commitInterval != null) { @@ -447,14 +464,17 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa return new SimpleStepBuilder(new StepBuilder(stepName)); } - private TaskletStep createTaskletStep() { + /** + * @return a new {@link TaskletStep} + */ + protected TaskletStep createTaskletStep() { TaskletStepBuilder builder = new StepBuilder(name).tasklet(tasklet); enhanceTaskletStepBuilder(builder); return builder.build(); } @SuppressWarnings("serial") - private void enhanceTaskletStepBuilder(AbstractTaskletStepBuilder builder) { + protected void enhanceTaskletStepBuilder(AbstractTaskletStepBuilder builder) { enhanceCommonStep(builder); for (ChunkListener listener : chunkListeners) { @@ -496,7 +516,7 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa } - private Step createFlowStep() { + protected Step createFlowStep() { FlowStepBuilder builder = new StepBuilder(name).flow(flow); enhanceCommonStep(builder); return builder.build(); @@ -512,7 +532,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa } - private void validateFaultTolerantSettings() { + /** + * Validates that all components required to build a fault tolerant step are set + */ + protected void validateFaultTolerantSettings() { validateDependency("skippable-exception-classes", skippableExceptionClasses, "skip-limit", skipLimit, true); validateDependency("retryable-exception-classes", retryableExceptionClasses, "retry-limit", retryLimit, true); validateDependency("retry-listeners", retryListeners, "retry-limit", retryLimit, false); @@ -565,7 +588,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa return o != null; } - private boolean isFaultTolerant() { + /** + * @return true if the step is configured with any components that require fault tolerance + */ + protected boolean isFaultTolerant() { return backOffPolicy != null || skipPolicy != null || retryPolicy != null || isPositive(skipLimit) || isPositive(retryLimit) || isPositive(cacheCapacity) || isTrue(readerTransactionalQueue); } @@ -611,6 +637,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa this.name = name; } + public String getName() { + return this.name; + } + // ========================================================= // Flow Attributes // ========================================================= @@ -656,6 +686,13 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa this.stepExecutionAggregator = stepExecutionAggregator; } + /** + * @return stepExecutionAggregator the current step's {@link StepExecutionAggregator} + */ + protected StepExecutionAggregator getStepExecutionAggergator() { + return this.stepExecutionAggregator; + } + /** * @param partitionHandler the partitionHandler to set */ @@ -663,6 +700,13 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa this.partitionHandler = partitionHandler; } + /** + * @return partitionHandler the current step's {@link PartitionHandler} + */ + protected PartitionHandler getPartitionHandler() { + return this.partitionHandler; + } + /** * @param gridSize the gridSize to set */ @@ -726,6 +770,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa this.tasklet = tasklet; } + protected Tasklet getTasklet() { + return this.tasklet; + } + /** * @return transactionManager */ @@ -752,7 +800,6 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa */ @SuppressWarnings("unchecked") public void setListeners(Object[] listeners) { - // this.listeners = listeners; // useful for testing for (Object listener : listeners) { if (listener instanceof SkipListener) { SkipListener skipListener = (SkipListener) listener; @@ -819,6 +866,13 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa if(listener instanceof RetryWriteListener) { jsrRetryListeners.add(new RetryWriteListenerAdapter((RetryWriteListener) listener)); } + if(listener instanceof PartitionCollector) { + PartitionCollectorAdapter adapter = new PartitionCollectorAdapter(); + adapter.setPartitionCollector((PartitionCollector) listener); + adapter.setPartitionQueue(partitionQueue); + chunkListeners.add(adapter); + stepExecutionListeners.add(adapter); + } } } @@ -1079,4 +1133,24 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa this.hasChunkElement = hasChunkElement; } + /** + * @return true if the defined step has a <chunk> element + */ + protected boolean hasChunkElement() { + return this.hasChunkElement; + } + + /** + * @return true if the defined step has a <tasklet> element + */ + protected boolean hasTasklet() { + return this.tasklet != null; + } + + /** + * @return true if the defined step has a <partition> element + */ + protected boolean hasPartitionElement() { + return this.partitionHandler != null; + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyBeanPostProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyBeanPostProcessor.java index 1eda5a89f..a6ccfbcd6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyBeanPostProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyBeanPostProcessor.java @@ -49,6 +49,7 @@ import org.springframework.util.ReflectionUtils; * @author Michael Minella * @since 3.0 */ +@SuppressWarnings("unchecked") public class BatchPropertyBeanPostProcessor implements BeanPostProcessor, BeanFactoryAware { private static final String SCOPED_TARGET_BEAN_PREFIX = "scopedTarget."; private static final Log LOGGER = LogFactory.getLog(BatchPropertyBeanPostProcessor.class); @@ -167,7 +168,7 @@ public class BatchPropertyBeanPostProcessor implements BeanPostProcessor, BeanFa ConfigurableListableBeanFactory configurableListableBeanFactory = (ConfigurableListableBeanFactory) beanFactory; BeanExpressionContext beanExpressionContext = new BeanExpressionContext(configurableListableBeanFactory, - configurableListableBeanFactory.getBean(StepScope.class)); + configurableListableBeanFactory.getBean(StepScope.class)); this.jsrExpressionParser = new JsrExpressionParser(new StandardBeanExpressionResolver(), beanExpressionContext); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContext.java index 24d50d357..7b6312591 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContext.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContext.java @@ -19,6 +19,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; + import org.springframework.util.Assert; /** @@ -33,6 +34,7 @@ import org.springframework.util.Assert; * @since 3.0 */ public class BatchPropertyContext { + private static final String PARTITION_INDICATOR = ":partition"; private Properties jobProperties = new Properties(); private Map stepProperties = new HashMap(); private Map artifactProperties = new HashMap(); @@ -58,9 +60,18 @@ public class BatchPropertyContext { * @return the {@link Properties} for the Step */ public Properties getStepProperties(String stepName) { - Properties properties = stepProperties.get(stepName); + Properties properties = new Properties(); - return properties != null ? properties : new Properties(); + if(stepProperties.containsKey(stepName)) { + properties.putAll(stepProperties.get(stepName)); + } + + if(stepName.contains(PARTITION_INDICATOR)) { + String parentStepName = stepName.substring(0, stepName.indexOf(PARTITION_INDICATOR)); + properties.putAll(getStepProperties(parentStepName)); + } + + return properties; } /** @@ -103,6 +114,17 @@ public class BatchPropertyContext { properties.putAll(artifactProperties.get(artifactName)); } + if(stepName.contains(PARTITION_INDICATOR)) { + String parentStepName = stepName.substring(0, stepName.indexOf(PARTITION_INDICATOR)); + properties.putAll(getStepProperties(parentStepName)); + + Map parentArtifactProperties = stepArtifactProperties.get(parentStepName); + + if (parentArtifactProperties != null && parentArtifactProperties.containsKey(artifactName)) { + properties.putAll(parentArtifactProperties.get(artifactName)); + } + } + return properties; } @@ -177,6 +199,7 @@ public class BatchPropertyContext { * * @param batchPropertyContextEntries the {@link BatchPropertyContextEntry} objects to add */ + @SuppressWarnings("serial") public void setStepArtifactPropertiesContextEntry(List batchPropertyContextEntries) { for (BatchPropertyContextEntry batchPropertyContextEntry : batchPropertyContextEntries) { Assert.hasText(batchPropertyContextEntry.getStepName(), "Step name must be defined"); @@ -212,16 +235,16 @@ public class BatchPropertyContext { */ public String getPropertyName(BatchArtifact.BatchArtifactType batchArtifactType) { switch (batchArtifactType) { - case STEP: - return "stepPropertiesContextEntry"; - case STEP_ARTIFACT: - return "stepArtifactPropertiesContextEntry"; - case ARTIFACT: - return "artifactPropertiesContextEntry"; - case JOB: - return "jobPropertiesContextEntry"; - default: - throw new IllegalStateException("Unhandled BatchArtifactType of: " + batchArtifactType); + case STEP: + return "stepPropertiesContextEntry"; + case STEP_ARTIFACT: + return "stepArtifactPropertiesContextEntry"; + case ARTIFACT: + return "artifactPropertiesContextEntry"; + case JOB: + return "jobPropertiesContextEntry"; + default: + throw new IllegalStateException("Unhandled BatchArtifactType of: " + batchArtifactType); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java index c57c817b4..82b0e1154 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java @@ -100,6 +100,6 @@ public class ListenerParser { private BatchArtifact.BatchArtifactType getBatchArtifactType(String stepName) { return (stepName != null && !"".equals(stepName)) ? BatchArtifact.BatchArtifactType.STEP_ARTIFACT - : BatchArtifact.BatchArtifactType.ARTIFACT; + : BatchArtifact.BatchArtifactType.ARTIFACT; } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java new file mode 100644 index 000000000..c99749003 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java @@ -0,0 +1,155 @@ +/* + * Copyright 2013 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.core.jsr.configuration.xml; + +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; + +import org.springframework.batch.core.jsr.configuration.support.BatchArtifact.BatchArtifactType; +import org.springframework.batch.core.jsr.partition.JsrPartitionHandler; +import org.springframework.beans.MutablePropertyValues; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; + +/** + * Parser for the <partition> element as defined by JSR-352. + * + * @author Michael Minella + * @since 3.0 + */ +public class PartitionParser { + + private static final String REF = "ref"; + private static final String MAPPER_ELEMENT = "mapper"; + private static final String PLAN_ELEMENT = "plan"; + private static final String PARTITIONS_ATTRIBUTE = "partitions"; + private static final String THREADS_ATTRIBUTE = "threads"; + private static final String PROPERTIES_ELEMENT = "properties"; + private static final String ANALYZER_ELEMENT = "analyzer"; + private static final String COLLECTOR_ELEMENT = "collector"; + private static final String REDUCER_ELEMENT = "reducer"; + private static final String PARTITION_CONTEXT_PROPERTY = "propertyContext"; + private static final String PARTITION_MAPPER_PROPERTY = "partitionMapper"; + private static final String PARTITION_ANALYZER_PROPERTY = "partitionAnalyzer"; + private static final String PARTITION_REDUCER_PROPERTY = "partitionReducer"; + private static final String PARTITION_QUEUE_PROPERTY = "partitionDataQueue"; + private static final String LISTENERS_PROPERTY = "listeners"; + private static final String THREADS_PROPERTY = "threads"; + private static final String PARTITIONS_PROPERTY = "partitions"; + + private final String name; + + /** + * @param stepName the name of the step that is being partitioned + */ + public PartitionParser(String stepName) { + this.name = stepName; + } + + public void parse(Element element, AbstractBeanDefinition bd, ParserContext parserContext, String stepName) { + BeanDefinitionRegistry registry = parserContext.getRegistry(); + MutablePropertyValues factoryBeanProperties = bd.getPropertyValues(); + + AbstractBeanDefinition partitionHandlerDefinition = BeanDefinitionBuilder.genericBeanDefinition(JsrPartitionHandler.class) + .getBeanDefinition(); + + MutablePropertyValues properties = partitionHandlerDefinition.getPropertyValues(); + properties.addPropertyValue(PARTITION_CONTEXT_PROPERTY, new RuntimeBeanReference("batchPropertyContext")); + + Element mapperElement = DomUtils.getChildElementByTagName(element, MAPPER_ELEMENT); + + if(mapperElement != null) { + String mapperName = mapperElement.getAttribute(REF); + properties.add(PARTITION_MAPPER_PROPERTY, new RuntimeBeanReference(mapperName)); + new PropertyParser(mapperName, parserContext, BatchArtifactType.STEP_ARTIFACT, name).parseProperties(mapperElement); + } + + parsePartitionPlan(element, parserContext, stepName, properties); + + Element analyzerElement = DomUtils.getChildElementByTagName(element, ANALYZER_ELEMENT); + + if(analyzerElement != null) { + String analyzerName = analyzerElement.getAttribute(REF); + properties.add(PARTITION_ANALYZER_PROPERTY, new RuntimeBeanReference(analyzerName)); + new PropertyParser(analyzerName, parserContext, BatchArtifactType.STEP_ARTIFACT, name).parseProperties(analyzerElement); + } + + Element reducerElement = DomUtils.getChildElementByTagName(element, REDUCER_ELEMENT); + + if(reducerElement != null) { + String reducerName = reducerElement.getAttribute(REF); + factoryBeanProperties.add(PARTITION_REDUCER_PROPERTY, new RuntimeBeanReference(reducerName)); + new PropertyParser(reducerName, parserContext, BatchArtifactType.STEP_ARTIFACT, name).parseProperties(reducerElement); + } + + Element collectorElement = DomUtils.getChildElementByTagName(element, COLLECTOR_ELEMENT); + + if(collectorElement != null) { + // Only needed if a collector is used + registerCollectorAnalyzerQueue(parserContext); + properties.add(PARTITION_QUEUE_PROPERTY, new RuntimeBeanReference(name + "PartitionQueue")); + factoryBeanProperties.add("partitionQueue", new RuntimeBeanReference(name + "PartitionQueue")); + String collectorName = collectorElement.getAttribute(REF); + factoryBeanProperties.add(LISTENERS_PROPERTY, new RuntimeBeanReference(collectorName)); + new PropertyParser(collectorName, parserContext, BatchArtifactType.STEP_ARTIFACT, name).parseProperties(collectorElement); + } + + String partitionHandlerBeanName = name + ".partitionHandler"; + registry.registerBeanDefinition(partitionHandlerBeanName, partitionHandlerDefinition); + factoryBeanProperties.add("partitionHandler", new RuntimeBeanReference(partitionHandlerBeanName)); + + } + + private void registerCollectorAnalyzerQueue(ParserContext parserContext) { + AbstractBeanDefinition partitionQueueDefinition = BeanDefinitionBuilder.genericBeanDefinition(ConcurrentLinkedQueue.class) + .getBeanDefinition(); + + parserContext.getRegistry().registerBeanDefinition(name + "PartitionQueue", partitionQueueDefinition); + } + + protected void parsePartitionPlan(Element element, + ParserContext parserContext, String stepName, + MutablePropertyValues properties) { + Element planElement = DomUtils.getChildElementByTagName(element, PLAN_ELEMENT); + + if(planElement != null) { + String partitions = planElement.getAttribute(PARTITIONS_ATTRIBUTE); + String threads = planElement.getAttribute(THREADS_ATTRIBUTE); + + if(!StringUtils.hasText(threads)) { + threads = partitions; + } + + List partitionProperties = DomUtils.getChildElementsByTagName(planElement, PROPERTIES_ELEMENT); + + if(partitionProperties != null) { + for (Element partition : partitionProperties) { + String partitionStepName = stepName + ":partition" + partition.getAttribute("partition"); + new PropertyParser(partitionStepName, parserContext, BatchArtifactType.STEP).parsePartitionProperties(partition); + } + } + + properties.add(THREADS_PROPERTY, threads); + properties.add(PARTITIONS_PROPERTY, partitions); + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PropertyParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PropertyParser.java index 3fd51a818..253df467a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PropertyParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PropertyParser.java @@ -15,10 +15,12 @@ */ package org.springframework.batch.core.jsr.configuration.xml; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; + import org.springframework.batch.core.jsr.configuration.support.BatchArtifact; import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.beans.factory.config.BeanDefinition; @@ -80,13 +82,7 @@ public class PropertyParser { Properties properties = new Properties(); if (propertiesElements.size() == 1) { - List propertyElements = DomUtils.getChildElementsByTagName(propertiesElements.get(0), PROPERTY_ELEMENT); - - for (Element propertyElement : propertyElements) { - properties.put(propertyElement.getAttribute(PROPERTY_NAME_ATTRIBUTE), propertyElement.getAttribute(PROPERTY_VALUE_ATTRIBUTE)); - } - - addProperties(properties); + parsePropertiesElement(propertiesElements, properties); } else if (propertiesElements.size() > 1) { parserContext.getReaderContext().error("The element may not appear more than once in a single .", element); } @@ -94,12 +90,32 @@ public class PropertyParser { setJobProperties(properties); } + public void parsePartitionProperties(Element element) { + Properties properties = new Properties(); + + List elements = new ArrayList(); + elements.add(element); + parsePropertiesElement(elements, properties); + + setJobProperties(properties); + } + + private void parsePropertiesElement(List propertiesElements, Properties properties) { + List propertyElements = DomUtils.getChildElementsByTagName(propertiesElements.get(0), PROPERTY_ELEMENT); + + for (Element propertyElement : propertyElements) { + properties.put(propertyElement.getAttribute(PROPERTY_NAME_ATTRIBUTE), propertyElement.getAttribute(PROPERTY_VALUE_ATTRIBUTE)); + } + + addProperties(properties); + } + private void addProperties(Properties properties) { BeanDefinition beanDefinition = parserContext.getRegistry().getBeanDefinition(BATCH_PROPERTY_CONTEXT_BEAN_NAME); BatchPropertyContext batchPropertyContext = new BatchPropertyContext(); BatchPropertyContext.BatchPropertyContextEntry batchPropertyContextEntry = - batchPropertyContext.new BatchPropertyContextEntry(beanName, properties, batchArtifactType); + batchPropertyContext.new BatchPropertyContextEntry(beanName, properties, batchArtifactType); if (StringUtils.hasText(stepName)) { batchPropertyContextEntry.setStepName(stepName); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java index da2b71210..986fc6619 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java @@ -20,22 +20,31 @@ import javax.batch.api.chunk.CheckpointAlgorithm; import javax.batch.api.chunk.ItemProcessor; import javax.batch.api.chunk.ItemReader; import javax.batch.api.chunk.ItemWriter; +import javax.batch.api.partition.PartitionReducer; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.xml.StepParserStepFactoryBean; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.partition.JsrPartitionHandler; import org.springframework.batch.core.jsr.step.batchlet.BatchletAdapter; +import org.springframework.batch.core.jsr.step.builder.JsrBatchletStepBuilder; import org.springframework.batch.core.jsr.step.builder.JsrFaultTolerantStepBuilder; +import org.springframework.batch.core.jsr.step.builder.JsrPartitionStepBuilder; import org.springframework.batch.core.jsr.step.builder.JsrSimpleStepBuilder; +import org.springframework.batch.core.partition.JsrStepExecutionSplitter; import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder; 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.core.step.tasklet.TaskletStep; import org.springframework.batch.jsr.item.ItemProcessorAdapter; import org.springframework.batch.jsr.item.ItemReaderAdapter; import org.springframework.batch.jsr.item.ItemWriterAdapter; import org.springframework.batch.jsr.repeat.CheckpointAlgorithmAdapter; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.beans.factory.FactoryBean; +import org.springframework.util.Assert; /** * This {@link FactoryBean} is used by the JSR-352 namespace parser to create @@ -48,6 +57,109 @@ import org.springframework.beans.factory.FactoryBean; @SuppressWarnings({"rawtypes", "unchecked"}) public class StepFactoryBean extends StepParserStepFactoryBean { + private int partitions; + private BatchPropertyContext batchPropertyContext; + + private PartitionReducer reducer; + + public void setPartitionReducer(PartitionReducer reducer) { + this.reducer = reducer; + } + + public void setBatchPropertyContext(BatchPropertyContext context) { + this.batchPropertyContext = context; + } + + public void setPartitions(int partitions) { + this.partitions = partitions; + } + + /** + * Create a {@link Step} from the configuration provided. + * + * @see FactoryBean#getObject() + */ + @Override + public Object getObject() throws Exception { + if(hasPartitionElement()) { + return createPartitionStep(); + } + else if (hasChunkElement()) { + Assert.isTrue(!hasTasklet(), "Step [" + getName() + + "] has both a element and a 'ref' attribute referencing a Tasklet."); + + validateFaultTolerantSettings(); + + if (isFaultTolerant()) { + return createFaultTolerantStep(); + } + else { + return createSimpleStep(); + } + } + else if (hasTasklet()) { + return createTaskletStep(); + } + else { + return createFlowStep(); + } + } + + /** + * @return a new {@link TaskletStep} + */ + @Override + protected TaskletStep createTaskletStep() { + JsrBatchletStepBuilder jsrBatchletStepBuilder = new JsrBatchletStepBuilder(new StepBuilder(getName())); + jsrBatchletStepBuilder.setBatchPropertyContext(batchPropertyContext); + TaskletStepBuilder builder = jsrBatchletStepBuilder.tasklet(getTasklet()); + enhanceTaskletStepBuilder(builder); + return builder.build(); + } + + @Override + protected Step createPartitionStep() { + // Creating a partitioned step for the JSR needs to create two steps...the partitioned step and the step being executed. + Step executedStep = null; + + if (hasChunkElement()) { + Assert.isTrue(!hasTasklet(), "Step [" + getName() + + "] has both a element and a 'ref' attribute referencing a Tasklet."); + + validateFaultTolerantSettings(); + + if (isFaultTolerant()) { + executedStep = createFaultTolerantStep(); + } + else { + executedStep = createSimpleStep(); + } + } + else if (hasTasklet()) { + executedStep = createTaskletStep(); + } + + ((JsrPartitionHandler) super.getPartitionHandler()).setStep(executedStep); + + JsrPartitionStepBuilder builder = new JsrSimpleStepBuilder(new StepBuilder(executedStep.getName())).partitioner(executedStep); + + enhanceCommonStep(builder); + + if (getPartitionHandler() != null) { + builder.partitionHandler(getPartitionHandler()); + } + + if(reducer != null) { + builder.reducer(reducer); + } + + builder.splitter(new JsrStepExecutionSplitter(getName(), getJobRepository())); + + builder.aggregator(getStepExecutionAggergator()); + + return builder.build(); + } + /** * Wraps a {@link Batchlet} in a {@link BatchletAdapter} if required for consumption * by the rest of the framework. @@ -140,11 +252,15 @@ public class StepFactoryBean extends StepParserStepFactoryBean { @Override protected FaultTolerantStepBuilder getFaultTolerantStepBuilder(String stepName) { - return new JsrFaultTolerantStepBuilder(new StepBuilder(stepName)); + JsrFaultTolerantStepBuilder jsrFaultTolerantStepBuilder = new JsrFaultTolerantStepBuilder(new StepBuilder(stepName)); + jsrFaultTolerantStepBuilder.setBatchPropertyContext(batchPropertyContext); + return jsrFaultTolerantStepBuilder; } @Override protected SimpleStepBuilder getSimpleStepBuilder(String stepName) { - return new JsrSimpleStepBuilder(new StepBuilder(stepName)); + JsrSimpleStepBuilder jsrSimpleStepBuilder = new JsrSimpleStepBuilder(new StepBuilder(stepName)); + jsrSimpleStepBuilder.setBatchPropertyContext(batchPropertyContext); + return jsrSimpleStepBuilder; } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java index 9bc4db1d3..64d36d473 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java @@ -21,6 +21,7 @@ import org.springframework.batch.core.job.flow.support.state.StepState; import org.springframework.batch.core.jsr.configuration.support.BatchArtifact; import org.springframework.batch.core.listener.StepListenerFactoryBean; import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -45,11 +46,13 @@ public class StepParser extends AbstractSingleBeanDefinitionParser { private static final String ALLOW_START_IF_COMPLETE_ATTRIBUTE = "allow-start-if-complete"; private static final String START_LIMIT_ATTRIBUTE = "start-limit"; private static final String SPLIT_ID_ATTRIBUTE = "id"; + private static final String PARTITION_ELEMENT = "partition"; protected Collection parse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { BeanDefinitionBuilder defBuilder = BeanDefinitionBuilder.genericBeanDefinition(); AbstractBeanDefinition bd = defBuilder.getRawBeanDefinition(); bd.setBeanClass(StepFactoryBean.class); + bd.getPropertyValues().addPropertyValue("batchPropertyContext", new RuntimeBeanReference("batchPropertyContext")); BeanDefinitionBuilder stateBuilder = BeanDefinitionBuilder.genericBeanDefinition(StepState.class); @@ -87,6 +90,8 @@ public class StepParser extends AbstractSingleBeanDefinitionParser { new BatchletParser().parseBatchlet(nestedElement, bd, parserContext, stepName); } else if(name.equals(CHUNK_ELEMENT)) { new ChunkParser().parse(nestedElement, bd, parserContext, stepName); + } else if(name.equals(PARTITION_ELEMENT)) { + new PartitionParser(stepName).parse(nestedElement, bd, parserContext, stepName); } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java index 90468713e..d0de1d8d2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java @@ -388,7 +388,9 @@ public class JsrJobOperator implements JobOperator, InitializingBean { if(executions != null) { for (org.springframework.batch.core.StepExecution stepExecution : executions) { - batchExecutions.add(new org.springframework.batch.core.jsr.StepExecution(jobExplorer.getStepExecution(executionId, stepExecution.getId()))); + if(!stepExecution.getStepName().contains(":partition")) { + batchExecutions.add(new org.springframework.batch.core.jsr.StepExecution(jobExplorer.getStepExecution(executionId, stepExecution.getId()))); + } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java new file mode 100644 index 000000000..634ba437b --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java @@ -0,0 +1,267 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Properties; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.LinkedBlockingQueue; + +import javax.batch.api.partition.PartitionAnalyzer; +import javax.batch.api.partition.PartitionCollector; +import javax.batch.api.partition.PartitionMapper; +import javax.batch.api.partition.PartitionPlan; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchArtifact.BatchArtifactType; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext.BatchPropertyContextEntry; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.task.TaskRejectedException; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.util.Assert; + +/** + * Executes a step instance per thread using a {@link ThreadPoolTaskExecutor} in + * accordance with JSR-352. The results from each step is aggregated into a + * cumulative result. + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrPartitionHandler implements PartitionHandler, InitializingBean { + + // TODO: Replace with proper Channel and Messages once minimum support level for Spring is 4 + private Queue partitionDataQueue; + private Step step; + private int partitions; + private PartitionAnalyzer analyzer; + private PartitionMapper mapper; + private int threads; + private BatchPropertyContext propertyContext; + + /** + * @param queue {@link Queue} to receive the output of the {@link PartitionCollector} + */ + public void setPartitionDataQueue(Queue queue) { + this.partitionDataQueue = queue; + } + + /** + * @param context {@link BatchPropertyContext} to resolve partition level step properties + */ + public void setPropertyContext(BatchPropertyContext context) { + this.propertyContext = context; + } + + /** + * @param mapper {@link PartitionMapper} used to configure partitioning + */ + public void setPartitionMapper(PartitionMapper mapper) { + this.mapper = mapper; + } + + /** + * @param step the step to be executed as a partitioned step + */ + public void setStep(Step step) { + this.step = step; + } + + /** + * @param analyzer {@link PartitionAnalyzer} + */ + public void setPartitionAnalyzer(PartitionAnalyzer analyzer) { + this.analyzer = analyzer; + } + + /** + * @param threads the number of threads to execute the partitions to be run + * within. The default is the number of partitions. + */ + public void setThreads(int threads) { + this.threads = threads; + } + + /** + * @param partitions the number of partitions to be executed + */ + public void setPartitions(int partitions) { + this.partitions = partitions; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.partition.PartitionHandler#handle(org.springframework.batch.core.partition.StepExecutionSplitter, org.springframework.batch.core.StepExecution) + */ + @Override + public Collection handle(StepExecutionSplitter stepSplitter, + StepExecution stepExecution) throws Exception { + final List> tasks = new ArrayList>(); + final Set result = new HashSet(); + final ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + + Set partitionStepExecutions; + + if(mapper != null) { + PartitionPlan plan = mapper.mapPartitions(); + if(plan.getThreads() > 0) { + threads = plan.getThreads(); + } else if(plan.getPartitions() > 0) { + threads = plan.getPartitions(); + } else { + throw new IllegalArgumentException("Either a number of threads or partitions are required"); + } + + partitionStepExecutions = stepSplitter.split(stepExecution, plan.getPartitions()); + registerPartitionProperties(partitionStepExecutions, plan); + + } else { + partitionStepExecutions = stepSplitter.split(stepExecution, partitions); + } + + taskExecutor.setCorePoolSize(threads); + taskExecutor.setMaxPoolSize(threads); + + taskExecutor.initialize(); + + for (final StepExecution curStepExecution : partitionStepExecutions) { + final FutureTask task = createTask(step, curStepExecution); + + try { + taskExecutor.execute(task); + tasks.add(task); + } catch (TaskRejectedException e) { + // couldn't execute one of the tasks + ExitStatus exitStatus = ExitStatus.FAILED + .addExitDescription("TaskExecutor rejected the task for this step."); + /* + * Set the status in case the caller is tracking it through the + * JobExecution. + */ + curStepExecution.setStatus(BatchStatus.FAILED); + curStepExecution.setExitStatus(exitStatus); + result.add(stepExecution); + } + } + + while(true) { + while(!partitionDataQueue.isEmpty()) { + analyzer.analyzeCollectorData(partitionDataQueue.remove()); + } + + processFinishedPartitions(tasks, result); + + if(tasks.size() == 0) { + break; + } + } + + return result; + } + + private void processFinishedPartitions( + final List> tasks, + final Set result) throws InterruptedException, + ExecutionException, Exception { + for(int i = 0; i < tasks.size(); i++) { + Future curTask = tasks.get(i); + + if(curTask.isDone()) { + StepExecution curStepExecution = curTask.get(); + + if(analyzer != null) { + analyzer.analyzeStatus(curStepExecution.getStatus().getBatchStatus(), curStepExecution.getExitStatus().getExitCode()); + } + + result.add(curStepExecution); + + tasks.remove(i); + i--; + } + } + } + + private void registerPartitionProperties( + Set partitionStepExecutions, PartitionPlan plan) { + Properties[] partitionProperties = plan.getPartitionProperties(); + if(partitionProperties != null) { + Iterator executions = partitionStepExecutions.iterator(); + + int i = 0; + while(executions.hasNext()) { + StepExecution curExecution = executions.next(); + + if(i < partitionProperties.length) { + Properties partitionPropertyValues = partitionProperties[i]; + if(partitionPropertyValues != null) { + List entries = new ArrayList(); + BatchPropertyContextEntry entry = propertyContext.new BatchPropertyContextEntry(curExecution.getStepName(), partitionPropertyValues, BatchArtifactType.STEP); + entries.add(entry); + + propertyContext.setStepPropertiesContextEntry(entries); + } + + i++; + } else { + break; + } + } + } + } + + /** + * Creates the task executing the given step in the context of the given execution. + * + * @param step the step to execute + * @param stepExecution the given execution + * @return the task executing the given step + */ + protected FutureTask createTask(final Step step, + final StepExecution stepExecution) { + return new FutureTask(new Callable() { + @Override + public StepExecution call() throws Exception { + step.execute(stepExecution); + return stepExecution; + } + }); + } + + @Override + public void afterPropertiesSet() throws Exception { + Assert.notNull(propertyContext, "A BatchPropertyContext is required"); + Assert.isTrue(mapper != null || threads > 0, "Either a mapper implementation or the number of partitions/threads is required"); + + if(partitionDataQueue == null) { + partitionDataQueue = new LinkedBlockingQueue(); + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java new file mode 100644 index 000000000..ac49fe4dd --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java @@ -0,0 +1,79 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +import java.io.Serializable; +import java.util.Queue; + +import javax.batch.api.partition.PartitionAnalyzer; +import javax.batch.api.partition.PartitionCollector; + +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; + +/** + * Adapter class used to wrap a {@link PartitionCollector} so that it can be consumed + * as a {@link ChunkListener}. A thread safe {@link Queue} is required along with the + * {@link PartitionCollector}. The {@link Queue} is where the result of the call to + * the PartitionCollector will be placed. + * + * @author Michael Minella + * @since 3.0 + */ +public class PartitionCollectorAdapter implements ChunkListener, InitializingBean { + + private PartitionCollector collector; + private Queue partitionQueue; + + /** + * @param queue destination for results of each {@link PartitionCollector#collectPartitionData()} call. + */ + public void setPartitionQueue(Queue queue) { + this.partitionQueue = queue; + } + + /** + * @param collector Provides partition specific information back to the {@link PartitionAnalyzer} as needed. + */ + public void setPartitionCollector(PartitionCollector collector) { + this.collector = collector; + } + + @Override + public void beforeChunk(ChunkContext context) { + } + + @Override + public void afterChunk(ChunkContext context) { + try { + partitionQueue.add(collector.collectPartitionData()); + } catch (Exception e) { + throw new PartitionException(e); + } + } + + @Override + public void afterChunkError(ChunkContext context) { + } + + @Override + public void afterPropertiesSet() throws Exception { + Assert.notNull(collector, "A PartitionCollector instance is required"); + Assert.notNull(partitionQueue, "A thread safe Queue instance is required"); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java new file mode 100644 index 000000000..58f9993e7 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java @@ -0,0 +1,36 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +@SuppressWarnings("serial") +public class PartitionException extends RuntimeException { + + public PartitionException() { + super(); + } + + public PartitionException(String message) { + super(message); + } + + public PartitionException(Throwable t) { + super(t); + } + + public PartitionException(String message, Throwable t) { + super(message, t); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/BatchletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/BatchletStep.java new file mode 100644 index 000000000..35842893b --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/BatchletStep.java @@ -0,0 +1,50 @@ +/* + * Copyright 2013 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.core.jsr.step; + +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.scope.context.StepSynchronizationManager; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.util.Assert; + +/** + * Special sub class of the {@link TaskletStep} for use with JSR-352 jobs. This + * implementation addresses the registration of a {@link BatchPropertyContext} for + * resolution of late binding parameters. + * + * @author Michael Minella + * @since 3.0 + */ +public class BatchletStep extends TaskletStep { + + private BatchPropertyContext propertyContext; + + /** + * @param name name of the step + * @param propertyContext {@link BatchPropertyContext} used to resolve batch properties. + */ + public BatchletStep(String name, BatchPropertyContext propertyContext) { + super(name); + Assert.notNull(propertyContext); + this.propertyContext = propertyContext; + } + + @Override + protected void doExecutionRegistration(StepExecution stepExecution) { + StepSynchronizationManager.register(stepExecution, propertyContext); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java new file mode 100644 index 000000000..4c1cd1e32 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java @@ -0,0 +1,84 @@ +/* + * Copyright 2013 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.core.jsr.step; + +import javax.batch.api.partition.PartitionReducer; +import javax.batch.api.partition.PartitionReducer.PartitionStatus; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.PartitionHandler; +import org.springframework.batch.core.partition.StepExecutionSplitter; +import org.springframework.batch.item.ExecutionContext; + +/** + * An extension of the {@link PartitionStep} that provides additional semantics + * required by JSR-352. Specifically, this implementation adds the required + * lifecycle calls to the {@link PartitionReducer} if it is used. + * + * @author Michael Minella + * @since 3.0 + */ +public class PartitionStep extends org.springframework.batch.core.partition.support.PartitionStep { + + private PartitionReducer reducer; + + public void setPartitionReducer(PartitionReducer reducer) { + this.reducer = reducer; + } + + /** + * Delegate execution to the {@link PartitionHandler} provided. The + * {@link StepExecution} passed in here becomes the parent or master + * execution for the partition, summarizing the status on exit of the + * logical grouping of work carried out by the {@link PartitionHandler}. The + * individual step executions and their input parameters (through + * {@link ExecutionContext}) for the partition elements are provided by the + * {@link StepExecutionSplitter}. + * + * @param stepExecution the master step execution for the partition + * + * @see Step#execute(StepExecution) + */ + @Override + protected void doExecute(StepExecution stepExecution) throws Exception { + + if(reducer != null) { + reducer.beginPartitionedStep(); + } + + // Wait for task completion and then aggregate the results + getPartitionHandler().handle(getStepExecutionSplitter(), stepExecution); + stepExecution.upgradeStatus(BatchStatus.COMPLETED); + + // If anything failed or had a problem we need to crap out + if (stepExecution.getStatus().isUnsuccessful()) { + if (reducer != null) { + reducer.rollbackPartitionedStep(); + reducer.beforePartitionedStepCompletion(); + reducer.afterPartitionedStepCompletion(PartitionStatus.ROLLBACK); + } + throw new JobExecutionException("Partition handler returned an unsuccessful step"); + } + + if (reducer != null) { + reducer.beforePartitionedStepCompletion(); + reducer.afterPartitionedStepCompletion(PartitionStatus.COMMIT); + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrBatchletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrBatchletStepBuilder.java new file mode 100644 index 000000000..7c28908d3 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrBatchletStepBuilder.java @@ -0,0 +1,102 @@ +/* + * Copyright 2013 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.core.jsr.step.builder; + +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.step.BatchletStep; +import org.springframework.batch.core.step.builder.StepBuilderException; +import org.springframework.batch.core.step.builder.StepBuilderHelper; +import org.springframework.batch.core.step.builder.TaskletStepBuilder; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; + +/** + * Extension of the {@link TaskletStepBuilder} that uses a {@link BatchletStep} instead + * of a {@link TaskletStep}. + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrBatchletStepBuilder extends TaskletStepBuilder { + + private BatchPropertyContext batchPropertyContext; + + /** + * @param context used to resolve lazy binded properties + */ + public void setBatchPropertyContext(BatchPropertyContext context) { + this.batchPropertyContext = context; + } + + public JsrBatchletStepBuilder(StepBuilderHelper parent) { + super(parent); + } + + /** + * Build the step from the components collected by the fluent setters. Delegates first to {@link #enhance(Step)} and + * then to {@link #createTasklet()} in subclasses to create the actual tasklet. + * + * @return a tasklet step fully configured and read to execute + */ + @Override + public TaskletStep build() { + + registerStepListenerAsChunkListener(); + + BatchletStep step = new BatchletStep(getName(), batchPropertyContext); + + super.enhance(step); + + step.setChunkListeners(chunkListeners.toArray(new ChunkListener[0])); + + if (getTransactionAttribute() != null) { + step.setTransactionAttribute(getTransactionAttribute()); + } + + if (getStepOperations() == null) { + + stepOperations(new RepeatTemplate()); + + if (getTaskExecutor() != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(getTaskExecutor()); + repeatTemplate.setThrottleLimit(getThrottleLimit()); + stepOperations(repeatTemplate); + } + + ((RepeatTemplate) getStepOperations()).setExceptionHandler(getExceptionHandler()); + + } + step.setStepOperations(getStepOperations()); + step.setTasklet(createTasklet()); + + step.setStreams(getStreams().toArray(new ItemStream[0])); + + try { + step.afterPropertiesSet(); + } + catch (Exception e) { + throw new StepBuilderException(e); + } + + return step; + + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java index e0c64960f..24058fa0f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java @@ -16,17 +16,26 @@ package org.springframework.batch.core.jsr.step.builder; import java.util.ArrayList; - import java.util.List; + +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.Step; import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.step.BatchletStep; import org.springframework.batch.core.jsr.step.item.JsrChunkProvider; import org.springframework.batch.core.jsr.step.item.JsrFaultTolerantChunkProcessor; import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder; import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.core.step.builder.StepBuilderException; import org.springframework.batch.core.step.item.ChunkOrientedTasklet; import org.springframework.batch.core.step.item.ChunkProcessor; import org.springframework.batch.core.step.item.ChunkProvider; import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; /** * A step builder that extends the {@link FaultTolerantStepBuilder} to create JSR-352 @@ -41,6 +50,12 @@ import org.springframework.batch.core.step.skip.SkipPolicy; */ public class JsrFaultTolerantStepBuilder extends FaultTolerantStepBuilder { + private BatchPropertyContext batchPropertyContext; + + public void setBatchPropertyContext(BatchPropertyContext batchPropertyContext) { + this.batchPropertyContext = batchPropertyContext; + } + public JsrFaultTolerantStepBuilder(StepBuilder parent) { super(parent); } @@ -50,6 +65,60 @@ public class JsrFaultTolerantStepBuilder extends FaultTolerantStepBuilder< return this; } + + /** + * Build the step from the components collected by the fluent setters. Delegates first to {@link #enhance(Step)} and + * then to {@link #createTasklet()} in subclasses to create the actual tasklet. + * + * @return a tasklet step fully configured and read to execute + */ + @Override + public TaskletStep build() { + registerStepListenerAsSkipListener(); + registerAsStreamsAndListeners(getReader(), getProcessor(), getWriter()); + + registerStepListenerAsChunkListener(); + + BatchletStep step = new BatchletStep(getName(), batchPropertyContext); + + super.enhance(step); + + step.setChunkListeners(chunkListeners.toArray(new ChunkListener[0])); + + if (getTransactionAttribute() != null) { + step.setTransactionAttribute(getTransactionAttribute()); + } + + if (getStepOperations() == null) { + + stepOperations(new RepeatTemplate()); + + if (getTaskExecutor() != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(getTaskExecutor()); + repeatTemplate.setThrottleLimit(getThrottleLimit()); + stepOperations(repeatTemplate); + } + + ((RepeatTemplate) getStepOperations()).setExceptionHandler(getExceptionHandler()); + + } + step.setStepOperations(getStepOperations()); + step.setTasklet(createTasklet()); + + step.setStreams(getStreams().toArray(new ItemStream[0])); + + try { + step.afterPropertiesSet(); + } + catch (Exception e) { + throw new StepBuilderException(e); + } + + return step; + + } + @Override protected ChunkProvider createChunkProvider() { return new JsrChunkProvider(); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrPartitionStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrPartitionStepBuilder.java new file mode 100644 index 000000000..902220e30 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrPartitionStepBuilder.java @@ -0,0 +1,129 @@ +/* + * Copyright 2013 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.core.jsr.step.builder; + +import javax.batch.api.partition.PartitionReducer; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.jsr.step.PartitionStep; +import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter; +import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler; +import org.springframework.batch.core.step.builder.PartitionStepBuilder; +import org.springframework.batch.core.step.builder.StepBuilderException; +import org.springframework.batch.core.step.builder.StepBuilderHelper; +import org.springframework.core.task.SyncTaskExecutor; + +/** + * An extension of the {@link PartitionStepBuilder} that uses {@link PartitionStep} + * so that JSR-352 specific semantics are honored. + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrPartitionStepBuilder extends PartitionStepBuilder { + + private PartitionReducer reducer; + + /** + * @param parent parent step builder for basic step properties + */ + public JsrPartitionStepBuilder(StepBuilderHelper parent) { + super(parent); + } + + /** + * @param reducer used to provide a single callback at the beginning and end + * of a partitioned step. + * + * @return this + */ + public JsrPartitionStepBuilder reducer(PartitionReducer reducer) { + this.reducer = reducer; + return this; + } + + @Override + public JsrPartitionStepBuilder step(Step step) { + super.step(step); + return this; + } + + @Override + public Step build() { + PartitionStep step = new PartitionStep(); + step.setName(getName()); + super.enhance(step); + + if (getPartitionHandler() != null) { + step.setPartitionHandler(getPartitionHandler()); + } + else { + TaskExecutorPartitionHandler partitionHandler = new TaskExecutorPartitionHandler(); + partitionHandler.setStep(getStep()); + if (getTaskExecutor() == null) { + taskExecutor(new SyncTaskExecutor()); + } + partitionHandler.setGridSize(getGridSize()); + partitionHandler.setTaskExecutor(getTaskExecutor()); + step.setPartitionHandler(partitionHandler); + } + + if (getSplitter() != null) { + step.setStepExecutionSplitter(getSplitter()); + } + else { + + boolean allowStartIfComplete = isAllowStartIfComplete(); + String name = getStepName(); + if (getStep() != null) { + try { + allowStartIfComplete = getStep().isAllowStartIfComplete(); + name = getStep().getName(); + } + catch (Exception e) { + logger.info("Ignored exception from step asking for name and allowStartIfComplete flag. " + + "Using default from enclosing PartitionStep (" + name + "," + allowStartIfComplete + ")."); + } + } + SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(); + splitter.setPartitioner(getPartitioner()); + splitter.setJobRepository(getJobRepository()); + splitter.setAllowStartIfComplete(allowStartIfComplete); + splitter.setStepName(name); + splitter(splitter); + step.setStepExecutionSplitter(splitter); + + } + + if (getAggregator() != null) { + step.setStepExecutionAggregator(getAggregator()); + } + + if(reducer != null) { + step.setPartitionReducer(reducer); + } + + try { + step.afterPropertiesSet(); + } + catch (Exception e) { + throw new StepBuilderException(e); + } + + return step; + + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java index 5781a6d17..3c45896ff 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java @@ -17,18 +17,26 @@ package org.springframework.batch.core.jsr.step.builder; import java.util.ArrayList; +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.Step; import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.step.BatchletStep; import org.springframework.batch.core.jsr.step.item.JsrChunkProcessor; import org.springframework.batch.core.jsr.step.item.JsrChunkProvider; import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder; import org.springframework.batch.core.step.builder.SimpleStepBuilder; import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.core.step.builder.StepBuilderException; import org.springframework.batch.core.step.item.ChunkOrientedTasklet; import org.springframework.batch.core.step.item.ChunkProcessor; import org.springframework.batch.core.step.item.ChunkProvider; import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; import org.springframework.util.Assert; /** @@ -43,10 +51,73 @@ import org.springframework.util.Assert; */ public class JsrSimpleStepBuilder extends SimpleStepBuilder { + private BatchPropertyContext batchPropertyContext; + public JsrSimpleStepBuilder(StepBuilder parent) { super(parent); } + public JsrPartitionStepBuilder partitioner(Step step) { + return new JsrPartitionStepBuilder(this).step(step); + } + + public void setBatchPropertyContext(BatchPropertyContext batchPropertyContext) { + this.batchPropertyContext = batchPropertyContext; + } + + /** + * Build the step from the components collected by the fluent setters. Delegates first to {@link #enhance(Step)} and + * then to {@link #createTasklet()} in subclasses to create the actual tasklet. + * + * @return a tasklet step fully configured and read to execute + */ + @Override + public TaskletStep build() { + registerStepListenerAsItemListener(); + registerAsStreamsAndListeners(getReader(), getProcessor(), getWriter()); + registerStepListenerAsChunkListener(); + + BatchletStep step = new BatchletStep(getName(), batchPropertyContext); + + super.enhance(step); + + step.setChunkListeners(chunkListeners.toArray(new ChunkListener[0])); + + if (getTransactionAttribute() != null) { + step.setTransactionAttribute(getTransactionAttribute()); + } + + if (getStepOperations() == null) { + + stepOperations(new RepeatTemplate()); + + if (getTaskExecutor() != null) { + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setTaskExecutor(getTaskExecutor()); + repeatTemplate.setThrottleLimit(getThrottleLimit()); + stepOperations(repeatTemplate); + } + + ((RepeatTemplate) getStepOperations()).setExceptionHandler(getExceptionHandler()); + + } + step.setStepOperations(getStepOperations()); + step.setTasklet(createTasklet()); + + ItemStream[] streams = getStreams().toArray(new ItemStream[0]); + step.setStreams(streams); + + try { + step.afterPropertiesSet(); + } + catch (Exception e) { + throw new StepBuilderException(e); + } + + return step; + + } + @Override @SuppressWarnings({"unchecked", "rawtypes"}) protected Tasklet createTasklet() { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/JsrStepExecutionSplitter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/JsrStepExecutionSplitter.java new file mode 100644 index 000000000..1c756bed2 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/JsrStepExecutionSplitter.java @@ -0,0 +1,87 @@ +/* + * Copyright 2013 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.core.partition; + +import java.util.Comparator; +import java.util.Set; +import java.util.TreeSet; + +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobExecutionException; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.launch.JsrJobOperator; +import org.springframework.batch.core.repository.JobRepository; + +/** + * Provides JSR-352 specific behavior for the splitting of {@link StepExecution}s. + * + * @author Michael Minella + * @since 3.0 + */ +public class JsrStepExecutionSplitter implements StepExecutionSplitter { + + private String stepName; + private JobRepository jobRepository; + + public JsrStepExecutionSplitter(String stepName, JobRepository jobRepository) { + this.stepName = stepName; + this.jobRepository = jobRepository; + } + + @Override + public String getStepName() { + return this.stepName; + } + + /** + * Returns the same number of {@link StepExecution}s as the gridSize specifies. Each + * of the child StepExecutions will not be available via the {@link JsrJobOperator} per + * JSR-352. + * + * @see https://java.net/projects/jbatch/lists/public/archive/2013-10/message/10 + */ + @Override + public Set split(StepExecution stepExecution, int gridSize) + throws JobExecutionException { + Set executions = new TreeSet(new Comparator() { + + @Override + public int compare(StepExecution arg0, StepExecution arg1) { + String r1 = ""; + String r2 = ""; + if (arg0 != null) { + r1 = arg0.getStepName(); + } + if (arg1 != null) { + r2 = arg1.getStepName(); + } + + return r1.compareTo(r2); + } + }); + JobExecution jobExecution = stepExecution.getJobExecution(); + + for(int i = 0; i < gridSize; i++) { + String stepName = this.stepName + ":partition" + i; + StepExecution curStepExecution = jobExecution.createStepExecution(stepName); + executions.add(curStepExecution); + } + + jobRepository.addAll(executions); + + return executions; + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java index 6eed9e759..fc29e4576 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java @@ -110,7 +110,13 @@ public class PartitionStep extends AbstractStep { if (stepExecution.getStatus().isUnsuccessful()) { throw new JobExecutionException("Partition handler returned an unsuccessful step"); } - } + protected StepExecutionSplitter getStepExecutionSplitter() { + return stepExecutionSplitter; + } + + protected PartitionHandler getPartitionHandler() { + return partitionHandler; + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContext.java index 1884604ae..d820c66cc 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContext.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContext.java @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2013 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. @@ -21,15 +21,16 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Properties; import java.util.Set; -import java.util.Map.Entry; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameter; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.batch.core.scope.StepScope; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor; @@ -42,9 +43,10 @@ import org.springframework.util.Assert; * convenience methods for accessing commonly used properties like the * {@link ExecutionContext} associated with the step or its enclosing job * execution. - * + * * @author Dave Syer - * + * @author Michael Minella + * */ public class StepContext extends SynchronizedAttributeAccessor { @@ -52,10 +54,12 @@ public class StepContext extends SynchronizedAttributeAccessor { private Map> callbacks = new HashMap>(); + private BatchPropertyContext propertyContext = null; + /** * Create a new instance of {@link StepContext} for this * {@link StepExecution}. - * + * * @param stepExecution a step execution */ public StepContext(StepExecution stepExecution) { @@ -64,11 +68,18 @@ public class StepContext extends SynchronizedAttributeAccessor { this.stepExecution = stepExecution; } + public StepContext(StepExecution stepExecution, BatchPropertyContext propertyContext) { + super(); + Assert.notNull(stepExecution, "A StepContext must have a non-null StepExecution"); + this.stepExecution = stepExecution; + this.propertyContext = propertyContext; + } + /** * Convenient accessor for current step name identifier. Usually this is the * same as the bean name of the step that is executing (but might not be * e.g. in a partition). - * + * * @return the step name identifier of the current {@link StepExecution} */ public String getStepName() { @@ -77,7 +88,7 @@ public class StepContext extends SynchronizedAttributeAccessor { /** * Convenient accessor for current job name identifier. - * + * * @return the job name identifier of the enclosing {@link JobInstance} * associated with the current {@link StepExecution} */ @@ -91,7 +102,7 @@ public class StepContext extends SynchronizedAttributeAccessor { /** * Convenient accessor for System properties to make it easy to access them * from placeholder expressions. - * + * * @return the current System properties */ public Properties getSystemProperties() { @@ -131,9 +142,21 @@ public class StepContext extends SynchronizedAttributeAccessor { return Collections.unmodifiableMap(result); } + @SuppressWarnings({"rawtypes", "unchecked"}) + public Map getPartitionPlan() { + Map partitionPlanProperties = new HashMap(); + + if(propertyContext != null) { + Map partitionProperties = propertyContext.getStepProperties(getStepName()); + partitionPlanProperties = partitionProperties; + } + + return Collections.unmodifiableMap(partitionPlanProperties); + } + /** * Allow clients to register callbacks for clean up on close. - * + * * @param name the callback id (unique attribute key in this context) * @param callback a callback to execute on close */ @@ -157,7 +180,7 @@ public class StepContext extends SynchronizedAttributeAccessor { /** * Override base class behaviour to ensure destruction callbacks are * unregistered as well as the default behaviour. - * + * * @see SynchronizedAttributeAccessor#removeAttribute(String) */ @Override @@ -212,7 +235,7 @@ public class StepContext extends SynchronizedAttributeAccessor { /** * The current {@link StepExecution} that is active in this context. - * + * * @return the current {@link StepExecution} */ public StepExecution getStepExecution() { @@ -232,15 +255,17 @@ public class StepContext extends SynchronizedAttributeAccessor { * Extend the base class method to include the step execution itself as a * key (i.e. two contexts are only equal if their step executions are the * same). - * + * * @see SynchronizedAttributeAccessor#equals(Object) */ @Override public boolean equals(Object other) { - if (!(other instanceof StepContext)) + if (!(other instanceof StepContext)) { return false; - if (other == this) + } + if (other == this) { return true; + } StepContext context = (StepContext) other; if (context.stepExecution == stepExecution) { return true; @@ -251,7 +276,7 @@ public class StepContext extends SynchronizedAttributeAccessor { /** * Overrides the default behaviour to provide a hash code based only on the * step execution. - * + * * @see SynchronizedAttributeAccessor#hashCode() */ @Override diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java index 50c556a89..71a6c509f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java @@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; /** * Central convenience class for framework use in managing the step scope @@ -29,9 +30,10 @@ import org.springframework.batch.core.StepExecution; * it is the responsibility of every {@link Step} implementation to ensure that * a {@link StepContext} is available on every thread that might be involved in * a step execution, including worker threads from a pool. - * + * * @author Dave Syer - * + * @author Michael Minella + * */ public class StepSynchronizationManager { @@ -64,7 +66,7 @@ public class StepSynchronizationManager { /** * Getter for the current context if there is one, otherwise returns null. - * + * * @return the current {@link StepContext} or null if there is none (if one * has not been registered for this thread). */ @@ -81,7 +83,7 @@ public class StepSynchronizationManager { * Register a context with the current thread - always put a matching * {@link #close()} call in a finally block to ensure that the correct * context is available in the enclosing block. - * + * * @param stepExecution the step context to register * @return a new {@link StepContext} or the current one if it has the same * {@link StepExecution} @@ -103,6 +105,32 @@ public class StepSynchronizationManager { return context; } + /** + * Register a context with the current thread - always put a matching + * {@link #close()} call in a finally block to ensure that the correct + * context is available in the enclosing block. + * + * @param stepExecution the step context to register + * @return a new {@link StepContext} or the current one if it has the same + * {@link StepExecution} + */ + public static StepContext register(StepExecution stepExecution, BatchPropertyContext propertyContext) { + if (stepExecution == null) { + return null; + } + getCurrent().push(stepExecution); + StepContext context; + synchronized (contexts) { + context = contexts.get(stepExecution); + if (context == null) { + context = new StepContext(stepExecution, propertyContext); + contexts.put(stepExecution, context); + } + } + increment(); + return context; + } + /** * Method for de-registering the current context - should always and only be * used by in conjunction with a matching {@link #register(StepExecution)} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index 91bf92520..211eebdd4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -26,6 +26,7 @@ import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.UnexpectedJobExecutionException; +import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.launch.NoSuchJobException; import org.springframework.batch.core.launch.support.ExitCodeMapper; import org.springframework.batch.core.listener.CompositeStepExecutionListener; @@ -45,6 +46,7 @@ import org.springframework.util.ClassUtils; * @author Dave Syer * @author Ben Hale * @author Robert Kasanicky + * @author Michael Minella */ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAware { @@ -185,7 +187,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw // Start with a default value that will be trumped by anything ExitStatus exitStatus = ExitStatus.EXECUTING; - StepSynchronizationManager.register(stepExecution); + doExecutionRegistration(stepExecution); try { getCompositeListener().beforeStep(stepExecution); @@ -268,12 +270,28 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw stepExecution.addFailureException(e); } - StepSynchronizationManager.release(); + doExecutionRelease(); logger.debug("Step execution complete: " + stepExecution.getSummary()); } } + /** + * Releases the most recent {@link StepExecution} + */ + protected void doExecutionRelease() { + StepSynchronizationManager.release(); + } + + /** + * Registers the {@link StepExecution} for property resolution via {@link StepScope} + * + * @param stepExecution + */ + protected void doExecutionRegistration(StepExecution stepExecution) { + StepSynchronizationManager.register(stepExecution); + } + /** * Determine the step status based on the exception. */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java index b605d5519..332213a2f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java @@ -36,15 +36,16 @@ import org.springframework.transaction.interceptor.TransactionAttribute; /** * Base class for step builders that want to build a {@link TaskletStep}. Handles common concerns across all tasklet * step variants, which are mostly to do with the type of tasklet they carry. - * + * * @author Dave Syer - * + * @author Michael Minella + * * @since 2.2 - * + * * @param the type of builder represented */ public abstract class AbstractTaskletStepBuilder> extends - StepBuilderHelper> { +StepBuilderHelper> { protected Set chunkListeners = new LinkedHashSet(); @@ -69,11 +70,11 @@ public abstract class AbstractTaskletStepBuilder getStreams() { + return this.streams; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index 16b709b77..1a4a322b3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -25,6 +25,7 @@ import java.util.Map; import java.util.Set; import javax.batch.operations.BatchRuntimeException; + import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.SkipListener; @@ -148,7 +149,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { } @SuppressWarnings("unchecked") - private void registerStepListenerAsSkipListener() { + protected void registerStepListenerAsSkipListener() { for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){ if (stepExecutionListener instanceof SkipListener){ listener((SkipListener)stepExecutionListener); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java index 99640ef79..4ed72102f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java @@ -29,9 +29,9 @@ import org.springframework.core.task.TaskExecutor; /** * Step builder for {@link PartitionStep} instances. A partition step executes the same step (possibly remotely) * multiple times with different input parameters (in the form of execution context). Useful for parallelization. - * + * * @author Dave Syer - * + * * @since 2.2 */ public class PartitionStepBuilder extends StepBuilderHelper { @@ -56,7 +56,7 @@ public class PartitionStepBuilder extends StepBuilderHelper parent) { @@ -66,7 +66,7 @@ public class PartitionStepBuilder extends StepBuilderHelper extends AbstractTaskletStepBuilder extends AbstractTaskletStepBuilder itemReader, + protected void registerAsStreamsAndListeners(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter) { for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) { if (itemHandler instanceof ItemStream) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContextTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContextTests.java index 741a68228..9b18aad86 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContextTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContextTests.java @@ -15,13 +15,15 @@ */ package org.springframework.batch.core.jsr.configuration.support; +import static org.junit.Assert.assertEquals; + import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Properties; + import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.assertEquals; /** *

@@ -35,6 +37,7 @@ public class BatchPropertyContextTests { private List stepProperties = new ArrayList(); private List artifactProperties = new ArrayList(); private List stepArtifactProperties = new ArrayList(); + private List partitionProperties = new ArrayList(); @Before public void setUp() { @@ -69,6 +72,26 @@ public class BatchPropertyContextTests { batchPropertyContextEntry.setStepName("step1"); this.stepArtifactProperties.add(batchPropertyContextEntry); + + Properties partitionProperties = new Properties(); + partitionProperties.setProperty("writerProperty1", "writerProperty1valuePartition0"); + partitionProperties.setProperty("writerProperty2", "writerProperty2valuePartition0"); + + BatchPropertyContext.BatchPropertyContextEntry partitionBatchPropertyContextEntry = + batchPropertyContext.new BatchPropertyContextEntry("writer", partitionProperties, BatchArtifact.BatchArtifactType.STEP_ARTIFACT); + partitionBatchPropertyContextEntry.setStepName("step2:partition0"); + + this.partitionProperties.add(partitionBatchPropertyContextEntry); + + Properties partitionStepProperties = new Properties(); + partitionStepProperties.setProperty("writerProperty1Step", "writerProperty1"); + partitionStepProperties.setProperty("writerProperty2Step", "writerProperty2"); + + BatchPropertyContext.BatchPropertyContextEntry partitionStepBatchPropertyContextEntry = + batchPropertyContext.new BatchPropertyContextEntry("writer", partitionStepProperties, BatchArtifact.BatchArtifactType.STEP_ARTIFACT); + partitionStepBatchPropertyContextEntry.setStepName("step2"); + + this.partitionProperties.add(partitionStepBatchPropertyContextEntry); } @Test @@ -179,4 +202,23 @@ public class BatchPropertyContextTests { assertEquals("jobProperty1value", job.getProperty("jobProperty1")); assertEquals("jobProperty2value", job.getProperty("jobProperty2")); } + + @Test + public void testPartitionProperties() { + BatchPropertyContext batchPropertyContext = new BatchPropertyContext(); + batchPropertyContext.setJobPropertiesContextEntry(jobProperties); + batchPropertyContext.setArtifactPropertiesContextEntry(artifactProperties); + batchPropertyContext.setStepPropertiesContextEntry(stepProperties); + batchPropertyContext.setStepArtifactPropertiesContextEntry(stepArtifactProperties); + batchPropertyContext.setStepArtifactPropertiesContextEntry(partitionProperties); + + Properties artifactProperties = batchPropertyContext.getStepArtifactProperties("step2:partition0", "writer"); + assertEquals(6, artifactProperties.size()); + assertEquals("writerProperty1", artifactProperties.getProperty("writerProperty1Step")); + assertEquals("writerProperty2", artifactProperties.getProperty("writerProperty2Step")); + assertEquals("writerProperty1valuePartition0", artifactProperties.getProperty("writerProperty1")); + assertEquals("writerProperty2valuePartition0", artifactProperties.getProperty("writerProperty2")); + assertEquals("step2PropertyValue1", artifactProperties.getProperty("step2PropertyName1")); + assertEquals("step2PropertyValue2", artifactProperties.getProperty("step2PropertyName2")); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java new file mode 100644 index 000000000..a8fc87ea3 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java @@ -0,0 +1,365 @@ +/* + * Copyright 2013 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.core.jsr.configuration.xml; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.Vector; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import javax.batch.api.BatchProperty; +import javax.batch.api.Batchlet; +import javax.batch.api.chunk.AbstractItemReader; +import javax.batch.api.chunk.AbstractItemWriter; +import javax.batch.api.partition.PartitionPlan; +import javax.batch.api.partition.PartitionPlanImpl; +import javax.batch.operations.JobOperator; +import javax.batch.runtime.BatchRuntime; +import javax.batch.runtime.BatchStatus; +import javax.batch.runtime.JobExecution; +import javax.batch.runtime.context.JobContext; +import javax.inject.Inject; + +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.springframework.util.Assert; + +public class PartitionParserTests { + + private static JobOperator operator; + private Pattern caPattern = Pattern.compile("ca"); + private Pattern asPattern = Pattern.compile("AS"); + + @BeforeClass + public static void beforeClass() { + operator = BatchRuntime.getJobOperator(); + } + + @Before + public void before() { + MyBatchlet.processed = 0; + MyBatchlet.threadNames = Collections.synchronizedSet(new HashSet()); + MyBatchlet.artifactNames = Collections.synchronizedSet(new HashSet()); + PartitionCollector.artifactNames = Collections.synchronizedSet(new HashSet()); + } + + @Test + public void testBatchletNoProperties() { + JobExecution execution = operator.getJobExecution(operator.start("partitionParserTestsBatchlet", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertEquals(10, MyBatchlet.processed); + assertEquals(10, MyBatchlet.threadNames.size()); + } + + @Test + public void testChunkNoProperties() { + JobExecution execution = operator.getJobExecution(operator.start("partitionParserTestsChunk", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertEquals(30, ItemReader.processedItems.size()); + assertEquals(10, ItemReader.threadNames.size()); + assertEquals(30, ItemWriter.processedItems.size()); + assertEquals(10, ItemWriter.threadNames.size()); + } + + @Test + public void testFullPartitionConfiguration() { + JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserTests", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertTrue(execution.getExitStatus().startsWith("BPS_")); + assertTrue(execution.getExitStatus().endsWith("BPSC_APSC")); + assertEquals(3, countMatches(execution.getExitStatus(), caPattern)); + assertEquals(3, countMatches(execution.getExitStatus(), asPattern)); + assertEquals(3, MyBatchlet.processed); + assertEquals(3, MyBatchlet.threadNames.size()); + } + + @Test + public void testFullPartitionConfigurationWithProperties() { + JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithPropertiesTests", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertTrue(execution.getExitStatus().startsWith("BPS_")); + assertTrue(execution.getExitStatus().endsWith("BPSC_APSC")); + assertEquals(3, countMatches(execution.getExitStatus(), caPattern)); + assertEquals(3, countMatches(execution.getExitStatus(), asPattern)); + assertEquals(3, MyBatchlet.processed); + assertEquals(3, MyBatchlet.threadNames.size()); + assertEquals(MyBatchlet.artifactNames.iterator().next(), "batchlet"); + assertEquals(PartitionMapper.name, "mapper"); + assertEquals(PartitionAnalyzer.name, "analyzer"); + assertEquals(PartitionReducer.name, "reducer"); + assertEquals(PartitionCollector.artifactNames.size(), 1); + assertTrue(PartitionCollector.artifactNames.contains("collector")); + } + + @Test + public void testFullPartitionConfigurationWithMapperSuppliedProperties() { + JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithMapperPropertiesTests", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertTrue(execution.getExitStatus().startsWith("BPS_")); + assertTrue(execution.getExitStatus().endsWith("BPSC_APSC")); + assertEquals(3, countMatches(execution.getExitStatus(), caPattern)); + assertEquals(3, countMatches(execution.getExitStatus(), asPattern)); + assertEquals(3, MyBatchlet.processed); + assertEquals(3, MyBatchlet.threadNames.size()); + + assertEquals(MyBatchlet.artifactNames.size(), 3); + assertTrue(MyBatchlet.artifactNames.contains("batchlet0")); + assertTrue(MyBatchlet.artifactNames.contains("batchlet1")); + assertTrue(MyBatchlet.artifactNames.contains("batchlet2")); + assertEquals(PartitionCollector.artifactNames.size(), 3); + assertTrue(PartitionCollector.artifactNames.contains("collector0")); + assertTrue(PartitionCollector.artifactNames.contains("collector1")); + assertTrue(PartitionCollector.artifactNames.contains("collector2")); + + assertEquals(PartitionMapper.name, "mapper"); + assertEquals(PartitionAnalyzer.name, "analyzer"); + assertEquals(PartitionReducer.name, "reducer"); + } + + @Test + public void testFullPartitionConfigurationWithHardcodedProperties() { + JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithHardcodedPropertiesTests", new Properties())); + + assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus()); + assertTrue(execution.getExitStatus().startsWith("BPS_")); + assertTrue(execution.getExitStatus().endsWith("BPSC_APSC")); + assertEquals(3, countMatches(execution.getExitStatus(), caPattern)); + assertEquals(3, countMatches(execution.getExitStatus(), asPattern)); + assertEquals(3, MyBatchlet.processed); + assertEquals(3, MyBatchlet.threadNames.size()); + + assertEquals(MyBatchlet.artifactNames.size(), 3); + assertTrue(MyBatchlet.artifactNames.contains("batchlet0")); + assertTrue(MyBatchlet.artifactNames.contains("batchlet1")); + assertTrue(MyBatchlet.artifactNames.contains("batchlet2")); + assertEquals(PartitionCollector.artifactNames.size(), 3); + assertTrue(PartitionCollector.artifactNames.contains("collector0")); + assertTrue(PartitionCollector.artifactNames.contains("collector1")); + assertTrue(PartitionCollector.artifactNames.contains("collector2")); + + assertEquals(PartitionMapper.name, "mapper"); + assertEquals(PartitionAnalyzer.name, "analyzer"); + assertEquals(PartitionReducer.name, "reducer"); + } + + private int countMatches(String string, Pattern pattern) { + Matcher matcher = pattern.matcher(string); + + int count = 0; + while(matcher.find()) { + count++; + } + + return count; + } + + public static class PartitionReducer implements javax.batch.api.partition.PartitionReducer { + + public static String name; + + @Inject + @BatchProperty + String artifactName; + + @Inject + protected JobContext jobContext; + + @Override + public void beginPartitionedStep() throws Exception { + name = artifactName; + jobContext.setExitStatus("BPS_"); + } + + @Override + public void beforePartitionedStepCompletion() throws Exception { + jobContext.setExitStatus(jobContext.getExitStatus() + "BPSC_"); + } + + @Override + public void rollbackPartitionedStep() throws Exception { + jobContext.setExitStatus(jobContext.getExitStatus() + "RPS"); + } + + @Override + public void afterPartitionedStepCompletion(PartitionStatus status) + throws Exception { + jobContext.setExitStatus(jobContext.getExitStatus() + "APSC"); + } + } + + public static class PartitionAnalyzer implements javax.batch.api.partition.PartitionAnalyzer { + + public static String name; + + @Inject + @BatchProperty + String artifactName; + + @Inject + protected JobContext jobContext; + + @Override + public void analyzeCollectorData(Serializable data) throws Exception { + name = artifactName; + + Assert.isTrue(data.equals("c")); + jobContext.setExitStatus(jobContext.getExitStatus() + data + "a"); + } + + @Override + public void analyzeStatus(BatchStatus batchStatus, String exitStatus) + throws Exception { + Assert.isTrue(batchStatus.equals(BatchStatus.COMPLETED)); + jobContext.setExitStatus(jobContext.getExitStatus() + "AS"); + } + } + + public static class PartitionCollector implements javax.batch.api.partition.PartitionCollector { + + protected static Set artifactNames = Collections.synchronizedSet(new HashSet()); + + @Inject + @BatchProperty + String artifactName; + + @Override + public Serializable collectPartitionData() throws Exception { + artifactNames.add(artifactName); + return "c"; + } + } + + public static class PropertyPartitionMapper implements javax.batch.api.partition.PartitionMapper { + + @Override + public PartitionPlan mapPartitions() throws Exception { + Properties[] props = new Properties[3]; + + for(int i = 0; i < props.length; i++) { + props[i] = new Properties(); + props[i].put("collectorName", "collector" + i); + props[i].put("batchletName", "batchlet" + i); + } + + PartitionPlan plan = new PartitionPlanImpl(); + plan.setPartitions(3); + plan.setThreads(3); + plan.setPartitionProperties(props); + + return plan; + } + } + + public static class PartitionMapper implements javax.batch.api.partition.PartitionMapper { + + public static String name; + + @Inject + @BatchProperty + public String artifactName; + + @Override + public PartitionPlan mapPartitions() throws Exception { + name = artifactName; + + PartitionPlan plan = new PartitionPlanImpl(); + plan.setPartitions(3); + plan.setThreads(3); + + return plan; + } + } + + public static class MyBatchlet implements Batchlet { + + protected static int processed = 0; + protected static Set threadNames = Collections.synchronizedSet(new HashSet()); + protected static Set artifactNames = Collections.synchronizedSet(new HashSet()); + + @Inject + @BatchProperty + String artifactName; + + @Override + public String process() throws Exception { + artifactNames.add(artifactName); + threadNames.add(Thread.currentThread().getName()); + processed++; + + return null; + } + + @Override + public void stop() throws Exception { + } + } + + public static class ItemReader extends AbstractItemReader { + + private List items; + protected static Vector processedItems = new Vector(); + protected static Set threadNames = Collections.synchronizedSet(new HashSet()); + + @Override + public void open(Serializable checkpoint) throws Exception { + items = new ArrayList(); + items.add(1); + items.add(2); + items.add(3); + } + + @Override + public Object readItem() throws Exception { + threadNames.add(Thread.currentThread().getName()); + if(items.size() > 0) { + Integer curItem = items.remove(0); + processedItems.add(curItem); + return curItem; + } else { + return null; + } + } + } + + public static class ItemWriter extends AbstractItemWriter { + + protected static Vector processedItems = new Vector(); + protected static Set threadNames = Collections.synchronizedSet(new HashSet()); + + @Override + public void writeItems(List items) throws Exception { + threadNames.add(Thread.currentThread().getName()); + for (Object object : items) { + processedItems.add((Integer) object); + } + } + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java index 0e867b3ea..b98a9ebc0 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java @@ -1,3 +1,18 @@ +/* + * Copyright 2013 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.core.jsr.launch; import static org.junit.Assert.assertEquals; @@ -341,6 +356,27 @@ public class JsrJobOperatorTests { jsrJobOperator.getStepExecutions(5l); } + @Test + public void testGetStepExecutionsPartitionedStepScenario() { + JobExecution jobExecution = new JobExecution(5l); + List stepExecutions = new ArrayList(); + stepExecutions.add(new StepExecution("step1", jobExecution, 1l)); + stepExecutions.add(new StepExecution("step2", jobExecution, 2l)); + stepExecutions.add(new StepExecution("step2:partition0", jobExecution, 2l)); + stepExecutions.add(new StepExecution("step2:partition1", jobExecution, 2l)); + stepExecutions.add(new StepExecution("step2:partition2", jobExecution, 2l)); + jobExecution.addStepExecutions(stepExecutions); + + when(jobExplorer.getJobExecution(5l)).thenReturn(jobExecution); + when(jobExplorer.getStepExecution(5l, 1l)).thenReturn(new StepExecution("step1", jobExecution, 1l)); + when(jobExplorer.getStepExecution(5l, 2l)).thenReturn(new StepExecution("step2", jobExecution, 2l)); + + List results = jsrJobOperator.getStepExecutions(5l); + + assertEquals("step1", results.get(0).getStepName()); + assertEquals("step2", results.get(1).getStepName()); + } + @Test public void testGetStepExecutionsNoStepExecutions() { JobExecution jobExecution = new JobExecution(5l); @@ -422,7 +458,7 @@ public class JsrJobOperatorTests { JsrJobOperator jobOperator = (JsrJobOperator) jsrJobOperator; JobExecution jobExecution = new JobExecution(1L, - new JobParametersBuilder().addString("prevKey1", "prevVal1").toJobParameters()); + new JobParametersBuilder().addString("prevKey1", "prevVal1").toJobParameters()); Properties userProperties = new Properties(); userProperties.put("userKey1", "userVal1"); @@ -440,9 +476,9 @@ public class JsrJobOperatorTests { JobExecution jobExecution = new JobExecution(1L, new JobParametersBuilder() - .addString("prevKey1", "prevVal1") - .addString("overrideTest", "jobExecution") - .toJobParameters()); + .addString("prevKey1", "prevVal1") + .addString("overrideTest", "jobExecution") + .toJobParameters()); Properties userProperties = new Properties(); userProperties.put("userKey1", "userVal1"); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java new file mode 100644 index 000000000..32294dfb4 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java @@ -0,0 +1,218 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.io.Serializable; +import java.util.Collection; +import java.util.Properties; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +import javax.batch.api.partition.PartitionAnalyzer; +import javax.batch.api.partition.PartitionMapper; +import javax.batch.api.partition.PartitionPlan; +import javax.batch.api.partition.PartitionPlanImpl; +import javax.batch.runtime.BatchStatus; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.partition.JsrStepExecutionSplitter; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.JobRepositorySupport; +import org.springframework.batch.core.step.StepSupport; + +public class JsrPartitionHandlerTests { + + private JsrPartitionHandler handler; + private JobRepository repository = new JobRepositorySupport(); + private StepExecution stepExecution = new StepExecution("step", new JobExecution(1L)); + private int count; + private BatchPropertyContext propertyContext; + private JsrStepExecutionSplitter stepSplitter; + + @Before + public void setUp() throws Exception { + stepSplitter = new JsrStepExecutionSplitter("step1", repository); + Analyzer.collectorData = ""; + Analyzer.status = ""; + count = 0; + handler = new JsrPartitionHandler(); + handler.setStep(new StepSupport() { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException { + count++; + stepExecution.setStatus(org.springframework.batch.core.BatchStatus.COMPLETED); + stepExecution.setExitStatus(new ExitStatus("done")); + } + }); + propertyContext = new BatchPropertyContext(); + handler.setPropertyContext(propertyContext); + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + } + + @Test + public void testAfterPropertiesSet() throws Exception { + handler = new JsrPartitionHandler(); + + try { + handler.afterPropertiesSet(); + fail("PropertyContext was not checked for"); + } catch(IllegalArgumentException iae) { + assertEquals("A BatchPropertyContext is required", iae.getMessage()); + } + + handler.setPropertyContext(new BatchPropertyContext()); + + try { + handler.afterPropertiesSet(); + fail("Threads or mapper was not checked for"); + } catch(IllegalArgumentException iae) { + assertEquals("Either a mapper implementation or the number of partitions/threads is required", iae.getMessage()); + } + + handler.setThreads(3); + handler.afterPropertiesSet(); + } + + @Test + public void testHardcodedNumberOfPartitions() throws Exception { + handler.setThreads(3); + handler.setPartitions(3); + handler.afterPropertiesSet(); + + Collection executions = handler.handle(stepSplitter, stepExecution); + + assertEquals(3, executions.size()); + assertEquals(3, count); + } + + @Test + public void testMapperProvidesPartitions() throws Exception { + handler.setPartitionMapper(new PartitionMapper() { + + @Override + public PartitionPlan mapPartitions() throws Exception { + PartitionPlan plan = new PartitionPlanImpl(); + plan.setPartitions(3); + plan.setThreads(0); + return plan; + } + }); + + handler.afterPropertiesSet(); + + Collection executions = handler.handle(new JsrStepExecutionSplitter("step1", repository), stepExecution); + + assertEquals(3, executions.size()); + assertEquals(3, count); + } + + @Test + public void testMapperProvidesPartitionsAndThreads() throws Exception { + handler.setPartitionMapper(new PartitionMapper() { + + @Override + public PartitionPlan mapPartitions() throws Exception { + PartitionPlan plan = new PartitionPlanImpl(); + plan.setPartitions(3); + plan.setThreads(1); + return plan; + } + }); + + handler.afterPropertiesSet(); + + Collection executions = handler.handle(new JsrStepExecutionSplitter("step1", repository), stepExecution); + + assertEquals(3, executions.size()); + assertEquals(3, count); + } + + @Test + public void testMapperWithProperties() throws Exception { + handler.setPartitionMapper(new PartitionMapper() { + + @Override + public PartitionPlan mapPartitions() throws Exception { + PartitionPlan plan = new PartitionPlanImpl(); + Properties [] props = new Properties[2]; + props[0] = new Properties(); + props[0].put("key1", "value1"); + props[1] = new Properties(); + props[1].put("key1", "value2"); + plan.setPartitionProperties(props); + plan.setPartitions(3); + plan.setThreads(1); + return plan; + } + }); + + handler.afterPropertiesSet(); + + Collection executions = handler.handle(new JsrStepExecutionSplitter("step1", repository), stepExecution); + + assertEquals(3, executions.size()); + assertEquals(3, count); + assertEquals("value1", propertyContext.getStepProperties("step1:partition0").get("key1")); + assertEquals("value2", propertyContext.getStepProperties("step1:partition1").get("key1")); + } + + @Test + public void testAnalyzer() throws Exception { + Queue queue = new ConcurrentLinkedQueue(); + queue.add("foo"); + queue.add("bar"); + + handler.setPartitionDataQueue(queue); + handler.setThreads(2); + handler.setPartitions(2); + handler.setPartitionAnalyzer(new Analyzer()); + handler.afterPropertiesSet(); + + Collection executions = handler.handle(new JsrStepExecutionSplitter("step1", repository), stepExecution); + + assertEquals(2, executions.size()); + assertEquals(2, count); + assertEquals("foobar", Analyzer.collectorData); + assertEquals("COMPLETEDdone", Analyzer.status); + } + + public static class Analyzer implements PartitionAnalyzer { + + public static String collectorData; + public static String status; + + @Override + public void analyzeCollectorData(Serializable data) throws Exception { + collectorData = collectorData + data; + } + + @Override + public void analyzeStatus(BatchStatus batchStatus, String exitStatus) + throws Exception { + status = batchStatus + exitStatus; + } + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrStepExecutionSplitterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrStepExecutionSplitterTests.java new file mode 100644 index 000000000..8cde55318 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrStepExecutionSplitterTests.java @@ -0,0 +1,54 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +import static org.junit.Assert.assertEquals; + +import java.util.Iterator; +import java.util.Set; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.partition.JsrStepExecutionSplitter; +import org.springframework.batch.core.step.JobRepositorySupport; + +public class JsrStepExecutionSplitterTests { + + private JsrStepExecutionSplitter splitter; + + @Before + public void setUp() throws Exception { + splitter = new JsrStepExecutionSplitter("step1", new JobRepositorySupport()); + } + + @Test + public void test() throws Exception { + Set executions = splitter.split(new StepExecution("step1", new JobExecution(5l)), 3); + + assertEquals(3, executions.size()); + + Iterator stepExecutions = executions.iterator(); + + int count = 0; + while(stepExecutions.hasNext()) { + StepExecution curExecution = stepExecutions.next(); + assertEquals("step1:partition" + count, curExecution.getStepName()); + count++; + } + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java new file mode 100644 index 000000000..4ddcb27b0 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2013 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.core.jsr.partition; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.io.Serializable; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +import javax.batch.api.partition.PartitionCollector; + +import org.junit.Before; +import org.junit.Test; + +public class PartitionCollectorAdapterTests { + + private PartitionCollectorAdapter adapter; + + @Before + public void setUp() throws Exception { + adapter = new PartitionCollectorAdapter(); + } + + @Test + public void testPropertiesSeet() throws Exception { + try { + adapter.afterPropertiesSet(); + fail("Did not check for a PartitionCollector instance"); + } catch (IllegalArgumentException iae) { + assertEquals(iae.getMessage(), "A PartitionCollector instance is required"); + } + + adapter.setPartitionCollector(new PartitionCollector() { + + @Override + public Serializable collectPartitionData() throws Exception { + return null; + } + }); + + try { + adapter.afterPropertiesSet(); + fail("Did not check for a queue"); + } catch (IllegalArgumentException iae) { + assertEquals(iae.getMessage(), "A thread safe Queue instance is required"); + } + + adapter.setPartitionQueue(new ConcurrentLinkedQueue()); + + adapter.afterPropertiesSet(); + } + + @Test + public void testAfterChunkSuccessful() throws Exception { + adapter.setPartitionCollector(new PartitionCollector() { + + private int count = 0; + + @Override + public Serializable collectPartitionData() throws Exception { + return String.valueOf(count++); + } + }); + + Queue dataQueue = new ConcurrentLinkedQueue(); + adapter.setPartitionQueue(dataQueue); + adapter.afterPropertiesSet(); + + adapter.afterChunk(null); + adapter.afterChunk(null); + adapter.afterChunk(null); + + assertEquals(3, dataQueue.size()); + assertEquals("0", dataQueue.remove()); + assertEquals("1", dataQueue.remove()); + assertEquals("2", dataQueue.remove()); + } + + @Test(expected=PartitionException.class) + public void testAfterChunkException() { + // Throws an NPE due to a null collector + adapter.afterChunk(null); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java index bafdf1d77..91a97de9c 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java @@ -31,6 +31,7 @@ import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.batch.core.jsr.step.builder.JsrSimpleStepBuilder; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; @@ -68,6 +69,7 @@ public class JsrChunkProcessorTests { readListener = new CountingListener(); builder = new JsrSimpleStepBuilder(new StepBuilder("step1")); + builder.setBatchPropertyContext(new BatchPropertyContext()); repository = new MapJobRepositoryFactoryBean().getJobRepository(); builder.repository(repository); builder.transactionManager(new ResourcelessTransactionManager()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java index 62b10cc83..814115a3d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java @@ -32,6 +32,7 @@ import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.batch.core.jsr.step.builder.JsrFaultTolerantStepBuilder; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; @@ -69,6 +70,7 @@ public class JsrFaultTolerantChunkProcessorTests { readListener = new CountingListener(); builder = new JsrFaultTolerantStepBuilder(new StepBuilder("step1")); + builder.setBatchPropertyContext(new BatchPropertyContext()); repository = new MapJobRepositoryFactoryBean().getJobRepository(); builder.repository(repository); builder.transactionManager(new ResourcelessTransactionManager()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextTests.java index 85c78b3b1..9f615e3a0 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextTests.java @@ -23,6 +23,8 @@ import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.Properties; import org.junit.Test; import org.springframework.batch.core.JobExecution; @@ -30,6 +32,9 @@ import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchArtifact.BatchArtifactType; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext.BatchPropertyContextEntry; import org.springframework.batch.item.ExecutionContext; /** @@ -44,6 +49,8 @@ public class StepContextTests { private StepContext context = new StepContext(stepExecution); + private BatchPropertyContext propertyContext = new BatchPropertyContext(); + @Test public void testGetStepExecution() { context = new StepContext(stepExecution); @@ -61,6 +68,22 @@ public class StepContextTests { } } + @Test + public void testGetPartitionPlan() { + Properties partitionPropertyValues = new Properties(); + partitionPropertyValues.put("key1", "value1"); + List entries = new ArrayList(); + BatchPropertyContextEntry entry = propertyContext.new BatchPropertyContextEntry(stepExecution.getStepName(), partitionPropertyValues, BatchArtifactType.STEP); + entries.add(entry); + + propertyContext.setStepPropertiesContextEntry(entries); + + context = new StepContext(stepExecution, propertyContext); + + Map plan = context.getPartitionPlan(); + assertEquals("value1", plan.get("key1")); + } + @Test public void testEqualsSelf() { assertEquals(context, context); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepSynchronizationManagerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepSynchronizationManagerTests.java index 188c40e3b..1a4836373 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepSynchronizationManagerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepSynchronizationManagerTests.java @@ -1,3 +1,18 @@ +/* + * Copyright 2013 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.core.scope.context; import static org.junit.Assert.assertEquals; @@ -19,11 +34,13 @@ import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext; import org.springframework.util.ReflectionUtils; public class StepSynchronizationManagerTests { private StepExecution stepExecution = new StepExecution("step", new JobExecution(0L)); + private BatchPropertyContext propertyContext = new BatchPropertyContext(); @Before @After @@ -40,6 +57,16 @@ public class StepSynchronizationManagerTests { assertNotNull(StepSynchronizationManager.getContext()); } + @Test + public void testGetContextWithBatchProperties() { + StepContext context = StepSynchronizationManager.getContext(); + assertNull(context); + StepSynchronizationManager.register(stepExecution, propertyContext); + context = StepSynchronizationManager.getContext(); + assertNotNull(context); + assertEquals(stepExecution, context.getStepExecution()); + } + @Test public void testClose() throws Exception { final List list = new ArrayList(); diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserTests.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserTests.xml new file mode 100644 index 000000000..56c1eefa3 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserTests.xml @@ -0,0 +1,12 @@ + + + + + + + + + + + + diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithHardcodedPropertiesTests.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithHardcodedPropertiesTests.xml new file mode 100644 index 000000000..09d5d605e --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithHardcodedPropertiesTests.xml @@ -0,0 +1,41 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithMapperPropertiesTests.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithMapperPropertiesTests.xml new file mode 100644 index 000000000..6045c7099 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithMapperPropertiesTests.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithPropertiesTests.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithPropertiesTests.xml new file mode 100644 index 000000000..00136cee0 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithPropertiesTests.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsBatchlet.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsBatchlet.xml new file mode 100644 index 000000000..3a549ce72 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsBatchlet.xml @@ -0,0 +1,9 @@ + + + + + + + + + diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsChunk.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsChunk.xml new file mode 100644 index 000000000..edfc55e15 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsChunk.xml @@ -0,0 +1,12 @@ + + + + + + + + + + + +