Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BATCH-2003: Implemented:

Browse Source 
* Basic partitioning support per JSR-352
* Parameter injection at the partition level
* PartitionCollector/PartitionAnalyzer functionality
* PartitionReducer functionality
This commit is contained in:
Michael Minella
2013-10-08 15:08:53 -05:00
parent 2eda8c2aeb
commit 102f614874
43 changed files with 2592 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
View File

@@ -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<I, O> implements FactoryBean, BeanNameAwa
private int gridSize = DEFAULT_GRID_SIZE;
private Queue<Serializable> partitionQueue;
//
// Tasklet Elements
//
@@ -238,13 +245,20 @@ public class StepParserStepFactoryBean<I, O> 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<Serializable> 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 <chunk/> element and a 'ref' attribute referencing a Tasklet.");
@@ -278,7 +292,10 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> 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<I, O> implements FactoryBean, BeanNameAwa
}
private Step createFaultTolerantStep() {
protected Step createFaultTolerantStep() {
FaultTolerantStepBuilder<I, O> builder = getFaultTolerantStepBuilder(this.name);
@@ -417,7 +434,7 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> 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<I, O> 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<I, O> 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<I, O> 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<I, O> implements FactoryBean, BeanNameAwa
this.name = name;
}
public String getName() {
return this.name;
}
// =========================================================
// Flow Attributes
// =========================================================
@@ -656,6 +686,13 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> 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<I, O> implements FactoryBean, BeanNameAwa
this.tasklet = tasklet;
}
protected Tasklet getTasklet() {
return this.tasklet;
}
/**
* @return transactionManager
*/
@@ -752,7 +800,6 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> skipListener = (SkipListener<I, O>) listener;
@@ -819,6 +866,13 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> implements FactoryBean, BeanNameAwa
this.hasChunkElement = hasChunkElement;
}
/**
* @return true if the defined step has a &lt;chunk&gt; element
*/
protected boolean hasChunkElement() {
return this.hasChunkElement;
}
/**
* @return true if the defined step has a &lt;tasklet&gt; element
*/
protected boolean hasTasklet() {
return this.tasklet != null;
}
/**
* @return true if the defined step has a &lt;partition&gt; element
*/
protected boolean hasPartitionElement() {
return this.partitionHandler != null;
}
}

3
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyBeanPostProcessor.java
View File

@@ -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);
}

47
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContext.java
View File

@@ -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<String, Properties> stepProperties = new HashMap<String, Properties>();
private Map<String, Properties> artifactProperties = new HashMap<String, Properties>();
@@ -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<String, Properties> 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<BatchPropertyContextEntry> 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);
}
}

2
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java
View File

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

155
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java Normal file
View File

@@ -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 &lt;partition&gt; 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<Element> 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);
}
}
}

32
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PropertyParser.java
View File

@@ -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<Element> 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 <properties> element may not appear more than once in a single <listener>.", element);
}
@@ -94,12 +90,32 @@ public class PropertyParser {
setJobProperties(properties);
}
public void parsePartitionProperties(Element element) {
Properties properties = new Properties();
List<Element> elements = new ArrayList<Element>();
elements.add(element);
parsePropertiesElement(elements, properties);
setJobProperties(properties);
}
private void parsePropertiesElement(List<Element> propertiesElements, Properties properties) {
List<Element> 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);

120
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java
View File

@@ -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 <chunk/> 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 <chunk/> 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;
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java
View File

@@ -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<BeanDefinition> 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);
}
}
}

4
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java
View File

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

267
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java Normal file
View File

@@ -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<Serializable> 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<Serializable> 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<StepExecution> handle(StepExecutionSplitter stepSplitter,
StepExecution stepExecution) throws Exception {
final List<Future<StepExecution>> tasks = new ArrayList<Future<StepExecution>>();
final Set<StepExecution> result = new HashSet<StepExecution>();
final ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
Set<StepExecution> 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<StepExecution> 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<Future<StepExecution>> tasks,
final Set<StepExecution> result) throws InterruptedException,
ExecutionException, Exception {
for(int i = 0; i < tasks.size(); i++) {
Future<StepExecution> 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<StepExecution> partitionStepExecutions, PartitionPlan plan) {
Properties[] partitionProperties = plan.getPartitionProperties();
if(partitionProperties != null) {
Iterator<StepExecution> 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<BatchPropertyContextEntry> entries = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
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<StepExecution> createTask(final Step step,
final StepExecution stepExecution) {
return new FutureTask<StepExecution>(new Callable<StepExecution>() {
@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<Serializable>();
}
}
}

79
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java Normal file
View File

@@ -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<Serializable> partitionQueue;
/**
* @param queue destination for results of each {@link PartitionCollector#collectPartitionData()} call.
*/
public void setPartitionQueue(Queue<Serializable> 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");
}
}

36
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java Normal file
View File

@@ -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);
}
}

50
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/BatchletStep.java Normal file
View File

@@ -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);
}
}

84
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java Normal file
View File

@@ -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);
}
}
}

102
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrBatchletStepBuilder.java Normal file
View File

@@ -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<? extends 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;
}
}

71
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java
View File

@@ -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<I, O> extends FaultTolerantStepBuilder<I, O> {
private BatchPropertyContext batchPropertyContext;
public void setBatchPropertyContext(BatchPropertyContext batchPropertyContext) {
this.batchPropertyContext = batchPropertyContext;
}
public JsrFaultTolerantStepBuilder(StepBuilder parent) {
super(parent);
}
@@ -50,6 +65,60 @@ public class JsrFaultTolerantStepBuilder<I, O> 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<I> createChunkProvider() {
return new JsrChunkProvider<I>();

129
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrPartitionStepBuilder.java Normal file
View File

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

71
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java
View File

@@ -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<I, O> extends SimpleStepBuilder<I, O> {
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() {

87
spring-batch-core/src/main/java/org/springframework/batch/core/partition/JsrStepExecutionSplitter.java Normal file
View File

@@ -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 <em>not</em> be available via the {@link JsrJobOperator} per
* JSR-352.
*
* @see <a href="https://java.net/projects/jbatch/lists/public/archive/2013-10/message/10">https://java.net/projects/jbatch/lists/public/archive/2013-10/message/10</a>
*/
@Override
public Set<StepExecution> split(StepExecution stepExecution, int gridSize)
throws JobExecutionException {
Set<StepExecution> executions = new TreeSet<StepExecution>(new Comparator<StepExecution>() {
@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;
}
}

8
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java
View File

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

55
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContext.java
View File

@@ -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<String, Set<Runnable>> callbacks = new HashMap<String, Set<Runnable>>();
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<String, Object> getPartitionPlan() {
Map<String, Object> partitionPlanProperties = new HashMap<String, Object>();
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

36
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepSynchronizationManager.java
View File

@@ -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)}

22
spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java
View File

@@ -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.
*/

53
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java
View File

@@ -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 <B> the type of builder represented
*/
public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBuilder<B>> extends
StepBuilderHelper<AbstractTaskletStepBuilder<B>> {
StepBuilderHelper<AbstractTaskletStepBuilder<B>> {
protected Set<ChunkListener> chunkListeners = new LinkedHashSet<ChunkListener>();
@@ -69,11 +70,11 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* 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
*/
public TaskletStep build() {
registerStepListenerAsChunkListener();
TaskletStep step = new TaskletStep(getName());
@@ -116,7 +117,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
}
private void registerStepListenerAsChunkListener() {
protected void registerStepListenerAsChunkListener() {
for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){
if (stepExecutionListener instanceof ChunkListener){
listener((ChunkListener)stepExecutionListener);
@@ -126,7 +127,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Register a chunk listener.
*
*
* @param listener the listener to register
* @return this for fluent chaining
*/
@@ -137,7 +138,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Register a stream for callbacks that manage restart data.
*
*
* @param stream the stream to register
* @return this for fluent chaining
*/
@@ -149,7 +150,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Provide a task executor to use when executing the tasklet. Default is to use a single-threaded (synchronous)
* executor.
*
*
* @param taskExecutor the task executor to register
* @return this for fluent chaining
*/
@@ -162,7 +163,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
* In the case of an asynchronous {@link #taskExecutor(TaskExecutor)} the number of concurrent tasklet executions
* can be throttled (beyond any throttling provided by a thread pool). The throttle limit should be less than the
* data source pool size used in the job repository for this step.
*
*
* @param throttleLimit maximium number of concurrent tasklet executions allowed
* @return this for fluent chaining
*/
@@ -173,7 +174,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Sets the exception handler to use in the case of tasklet failures. Default is to rethrow everything.
*
*
* @param exceptionHandler the exception handler
* @return this for fluent chaining
*/
@@ -185,7 +186,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Sets the repeat template used for iterating the tasklet execution. By default it will terminate only when the
* tasklet returns FINISHED (or null).
*
*
* @param repeatTemplate a repeat template with rules for iterating
* @return this for fluent chaining
*/
@@ -197,7 +198,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Sets the transaction attributes for the tasklet execution. Defaults to the default values for the transaction
* manager, but can be manipulated to provide longer timeouts for instance.
*
*
* @param transactionAttribute a transaction attribute set
* @return this for fluent chaining
*/
@@ -208,7 +209,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Convenience method for subclasses to access the step operations that were injected by user.
*
*
* @return the repeat operations used to iterate the tasklet executions
*/
protected RepeatOperations getStepOperations() {
@@ -217,7 +218,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Convenience method for subclasses to access the exception handler that was injected by user.
*
*
* @return the exception handler
*/
protected ExceptionHandler getExceptionHandler() {
@@ -226,7 +227,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
/**
* Convenience method for subclasses to determine if the step is concurrent.
*
*
* @return true if the tasklet is going to be run in multiple threads
*/
protected boolean concurrent() {
@@ -234,4 +235,20 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
return concurrent;
}
}
protected TaskExecutor getTaskExecutor() {
return taskExecutor;
}
protected int getThrottleLimit() {
return throttleLimit;
}
protected TransactionAttribute getTransactionAttribute() {
return transactionAttribute;
}
protected Set<ItemStream> getStreams() {
return this.streams;
}
}

3
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
View File

@@ -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<I, O> extends SimpleStepBuilder<I, O> {
}
@SuppressWarnings("unchecked")
private void registerStepListenerAsSkipListener() {
protected void registerStepListenerAsSkipListener() {
for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){
if (stepExecutionListener instanceof SkipListener){
listener((SkipListener<I,O>)stepExecutionListener);

54
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/PartitionStepBuilder.java
View File

@@ -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<PartitionStepBuilder> {
@@ -56,7 +56,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
/**
* Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used.
*
*
* @param parent a parent helper containing common step properties
*/
public PartitionStepBuilder(StepBuilderHelper<?> parent) {
@@ -66,7 +66,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
/**
* Add a partitioner which can be used to create a {@link StepExecutionSplitter}. Use either this or an explicit
* {@link #splitter(StepExecutionSplitter)} but not both.
*
*
* @param slaveStepName the name of the slave step (used to construct step execution names)
* @param partitioner a partitioner to use
* @return this for fluent chaining
@@ -81,7 +81,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
* Provide an actual step instance to execute in parallel. If an explicit
* {@link #partitionHandler(PartitionHandler)} is provided, the step is optional and is only used to extract
* configuration data (name and other basic properties of a step).
*
*
* @param step a step to execute in parallel
* @return this for fluent chaining
*/
@@ -94,7 +94,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
* Provide a task executor to use when constructing a {@link PartitionHandler} from the {@link #step(Step)}. Mainly
* used for running a step locally in parallel, but can be used to execute remotely if the step is remote. Not used
* if an explicit {@link #partitionHandler(PartitionHandler)} is provided.
*
*
* @param taskExecutor a task executor to use when executing steps in parallel
* @return this for fluent chaining
*/
@@ -107,9 +107,9 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
* Provide an explicit partition handler that will carry out the work of the partition step. The partition handler
* is the main SPI for adapting a partition step to a specific distributed computation environment. Optional if you
* only need local or remote processing through the Step interface.
*
*
* @see #step(Step) for setting up a default handler that works with a local or remote Step
*
*
* @param partitionHandler a partition handler
* @return this for fluent chaining
*/
@@ -122,7 +122,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
* A hint to the {@link #splitter(StepExecutionSplitter)} about how many step executions are required. If running
* locally or remotely through a {@link #taskExecutor(TaskExecutor)} determines precisely the number of step
* execution sin the first attempt at a partition step execution.
*
*
* @param gridSize the grid size
* @return this for fluent chaining
*/
@@ -134,7 +134,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
/**
* Provide an explicit {@link StepExecutionSplitter} instead of having one build from the
* {@link #partitioner(String, Partitioner)}. USeful if you need more control over the splitting.
*
*
* @param splitter a step execution splitter
* @return this for fluent chaining
*/
@@ -146,7 +146,7 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
/**
* Provide a step execution aggregator for aggregating partitioned step executions into a single result for the
* {@link PartitionStep} itself. Default is a simple implementation that works in most cases.
*
*
* @param aggregator a step execution aggregator
* @return this for fluent chaining
*/
@@ -156,7 +156,6 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
}
public Step build() {
PartitionStep step = new PartitionStep();
step.setName(getName());
super.enhance(step);
@@ -217,4 +216,35 @@ public class PartitionStepBuilder extends StepBuilderHelper<PartitionStepBuilder
}
protected TaskExecutor getTaskExecutor() {
return taskExecutor;
}
protected Partitioner getPartitioner() {
return partitioner;
}
protected Step getStep() {
return step;
}
protected PartitionHandler getPartitionHandler() {
return partitionHandler;
}
protected int getGridSize() {
return gridSize;
}
protected StepExecutionSplitter getSplitter() {
return splitter;
}
protected StepExecutionAggregator getAggregator() {
return aggregator;
}
protected String getStepName() {
return stepName;
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
View File

@@ -110,12 +110,13 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
*/
@Override
public TaskletStep build() {
registerStepListenerAsItemListener();
registerAsStreamsAndListeners(reader, processor, writer);
return super.build();
}
private void registerStepListenerAsItemListener() {
protected void registerStepListenerAsItemListener() {
for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){
checkAndAddItemListener(stepExecutionListener);
}
@@ -326,7 +327,7 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
return new SimpleCompletionPolicy(chunkSize);
}
private void registerAsStreamsAndListeners(ItemReader<? extends I> itemReader,
protected void registerAsStreamsAndListeners(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter) {
for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
if (itemHandler instanceof ItemStream) {

44
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/support/BatchPropertyContextTests.java
View File

@@ -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;
/**
* <p>
@@ -35,6 +37,7 @@ public class BatchPropertyContextTests {
private List<BatchPropertyContext.BatchPropertyContextEntry> stepProperties = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
private List<BatchPropertyContext.BatchPropertyContextEntry> artifactProperties = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
private List<BatchPropertyContext.BatchPropertyContextEntry> stepArtifactProperties = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
private List<BatchPropertyContext.BatchPropertyContextEntry> partitionProperties = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
@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"));
}
}

365
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java Normal file
View File

@@ -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<String>());
MyBatchlet.artifactNames = Collections.synchronizedSet(new HashSet<String>());
PartitionCollector.artifactNames = Collections.synchronizedSet(new HashSet<String>());
}
@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<String> artifactNames = Collections.synchronizedSet(new HashSet<String>());
@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<String> threadNames = Collections.synchronizedSet(new HashSet<String>());
protected static Set<String> artifactNames = Collections.synchronizedSet(new HashSet<String>());
@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<Integer> items;
protected static Vector<Integer> processedItems = new Vector<Integer>();
protected static Set<String> threadNames = Collections.synchronizedSet(new HashSet<String>());
@Override
public void open(Serializable checkpoint) throws Exception {
items = new ArrayList<Integer>();
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<Integer> processedItems = new Vector<Integer>();
protected static Set<String> threadNames = Collections.synchronizedSet(new HashSet<String>());
@Override
public void writeItems(List<Object> items) throws Exception {
threadNames.add(Thread.currentThread().getName());
for (Object object : items) {
processedItems.add((Integer) object);
}
}
}
}

44
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java
View File

@@ -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<StepExecution> stepExecutions = new ArrayList<StepExecution>();
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<javax.batch.runtime.StepExecution> 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");

218
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandlerTests.java Normal file
View File

@@ -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<StepExecution> 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<StepExecution> 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<StepExecution> 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<StepExecution> 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<Serializable> queue = new ConcurrentLinkedQueue<Serializable>();
queue.add("foo");
queue.add("bar");
handler.setPartitionDataQueue(queue);
handler.setThreads(2);
handler.setPartitions(2);
handler.setPartitionAnalyzer(new Analyzer());
handler.afterPropertiesSet();
Collection<StepExecution> 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;
}
}
}

54
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/JsrStepExecutionSplitterTests.java Normal file
View File

@@ -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<StepExecution> executions = splitter.split(new StepExecution("step1", new JobExecution(5l)), 3);
assertEquals(3, executions.size());
Iterator<StepExecution> stepExecutions = executions.iterator();
int count = 0;
while(stepExecutions.hasNext()) {
StepExecution curExecution = stepExecutions.next();
assertEquals("step1:partition" + count, curExecution.getStepName());
count++;
}
}
}

99
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java Normal file
View File

@@ -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<Serializable>());
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<Serializable> dataQueue = new ConcurrentLinkedQueue<Serializable>();
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);
}
}

2
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java
View File

@@ -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<String, String>(new StepBuilder("step1"));
builder.setBatchPropertyContext(new BatchPropertyContext());
repository = new MapJobRepositoryFactoryBean().getJobRepository();
builder.repository(repository);
builder.transactionManager(new ResourcelessTransactionManager());

2
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java
View File

@@ -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<String, String>(new StepBuilder("step1"));
builder.setBatchPropertyContext(new BatchPropertyContext());
repository = new MapJobRepositoryFactoryBean().getJobRepository();
builder.repository(repository);
builder.transactionManager(new ResourcelessTransactionManager());

23
spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextTests.java
View File

@@ -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<BatchPropertyContextEntry> entries = new ArrayList<BatchPropertyContext.BatchPropertyContextEntry>();
BatchPropertyContextEntry entry = propertyContext.new BatchPropertyContextEntry(stepExecution.getStepName(), partitionPropertyValues, BatchArtifactType.STEP);
entries.add(entry);
propertyContext.setStepPropertiesContextEntry(entries);
context = new StepContext(stepExecution, propertyContext);
Map<String, Object> plan = context.getPartitionPlan();
assertEquals("value1", plan.get("key1"));
}
@Test
public void testEqualsSelf() {
assertEquals(context, context);

27
spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepSynchronizationManagerTests.java
View File

@@ -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<String> list = new ArrayList<String>();

12
spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserTests.xml Normal file
View File

@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$MyBatchlet"/>
<partition>
<mapper ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionMapper"/>
<collector ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionCollector"/>
<analyzer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionAnalyzer"/>
<reducer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionReducer"/>
</partition>
</step>
</job>

41
spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithHardcodedPropertiesTests.xml Normal file
View File

@@ -0,0 +1,41 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$MyBatchlet">
<properties>
<property name="artifactName" value="#{partitionPlan['batchletName']}"/>
</properties>
</batchlet>
<partition>
<plan partitions="3">
<properties partition="0">
<property name="batchletName" value="batchlet0"/>
<property name="collectorName" value="collector0"/>
</properties>
<properties partition="1">
<property name="batchletName" value="batchlet1"/>
<property name="collectorName" value="collector1"/>
</properties>
<properties partition="2">
<property name="batchletName" value="batchlet2"/>
<property name="collectorName" value="collector2"/>
</properties>
</plan>
<collector ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionCollector">
<properties>
<property name="artifactName" value="#{partitionPlan['collectorName']}"/>
</properties>
</collector>
<analyzer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionAnalyzer">
<properties>
<property name="artifactName" value="analyzer"/>
</properties>
</analyzer>
<reducer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionReducer">
<properties>
<property name="artifactName" value="reducer"/>
</properties>
</reducer>
</partition>
</step>
</job>

28
spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithMapperPropertiesTests.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$MyBatchlet">
<properties>
<property name="artifactName" value="#{partitionPlan['batchletName']}"/>
</properties>
</batchlet>
<partition>
<mapper ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PropertyPartitionMapper"/>
<collector ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionCollector">
<properties>
<property name="artifactName" value="#{partitionPlan['collectorName']}"/>
</properties>
</collector>
<analyzer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionAnalyzer">
<properties>
<property name="artifactName" value="analyzer"/>
</properties>
</analyzer>
<reducer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionReducer">
<properties>
<property name="artifactName" value="reducer"/>
</properties>
</reducer>
</partition>
</step>
</job>

32
spring-batch-core/src/test/resources/META-INF/batch-jobs/fullPartitionParserWithPropertiesTests.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$MyBatchlet">
<properties>
<property name="artifactName" value="batchlet"/>
</properties>
</batchlet>
<partition>
<mapper ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionMapper">
<properties>
<property name="artifactName" value="mapper"/>
</properties>
</mapper>
<collector ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionCollector">
<properties>
<property name="artifactName" value="collector"/>
</properties>
</collector>
<analyzer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionAnalyzer">
<properties>
<property name="artifactName" value="analyzer"/>
</properties>
</analyzer>
<reducer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$PartitionReducer">
<properties>
<property name="artifactName" value="reducer"/>
</properties>
</reducer>
</partition>
</step>
</job>

9
spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsBatchlet.xml Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$MyBatchlet" />
<partition>
<plan partitions="10" />
</partition>
</step>
</job>

12
spring-batch-core/src/test/resources/META-INF/batch-jobs/partitionParserTestsChunk.xml Normal file
View File

@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="job_partitioned_1step" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<chunk item-count="3">
<reader ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$ItemReader"/>
<writer ref="org.springframework.batch.core.jsr.configuration.xml.PartitionParserTests$ItemWriter"/>
</chunk>
<partition>
<plan partitions="10" />
</partition>
</step>
</job>