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

BATCH-2003: Updated per code review comments

Browse Source
This commit is contained in:
Michael Minella
2013-11-19 10:13:32 -06:00
parent 102f614874
commit fb7582a54f
9 changed files with 80 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -293,7 +293,7 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
}
/**
* @param builder
* @param builder {@link StepBuilderHelper} representing the step to be enhanced
*/
protected void enhanceCommonStep(StepBuilderHelper<?> builder) {
if (allowStartIfComplete != null) {
@@ -867,11 +867,8 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
jsrRetryListeners.add(new RetryWriteListenerAdapter((RetryWriteListener) listener));
}
if(listener instanceof PartitionCollector) {
PartitionCollectorAdapter adapter = new PartitionCollectorAdapter();
adapter.setPartitionCollector((PartitionCollector) listener);
adapter.setPartitionQueue(partitionQueue);
PartitionCollectorAdapter adapter = new PartitionCollectorAdapter(partitionQueue, (PartitionCollector) listener);
chunkListeners.add(adapter);
stepExecutionListeners.add(adapter);
}
}
}

3
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowExecutionException.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.
@@ -19,6 +19,7 @@ package org.springframework.batch.core.job.flow;
* @author Dave Syer
*
*/
@SuppressWarnings("serial")
public class FlowExecutionException extends Exception {
/**

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

@@ -17,26 +17,28 @@ package org.springframework.batch.core.jsr.configuration.support;
import java.lang.annotation.Annotation;
import java.lang.reflect.AccessibleObject;
import javax.batch.api.BatchProperty;
import org.springframework.beans.factory.annotation.InjectionMetadata;
/**
* <p>This class overrides methods in the copied {@link SpringAutowiredAnnotationBeanPostProcessor} class
* to check for the {@link @BatchProperty} annotation before processing injection annotations. If the annotation
* to check for the {@link BatchProperty} annotation before processing injection annotations. If the annotation
* is found, further injection processing for the field is skipped.</p>
*/
public class JsrAutowiredAnnotationBeanPostProcessor extends SpringAutowiredAnnotationBeanPostProcessor {
@Override
protected InjectionMetadata findAutowiringMetadata(Class<?> clazz) {
return super.buildAutowiringMetadata(clazz);
}
@Override
protected InjectionMetadata findAutowiringMetadata(Class<?> clazz) {
return super.buildAutowiringMetadata(clazz);
}
@Override
protected Annotation findAutowiredAnnotation(AccessibleObject ao) {
if (ao.getAnnotation(BatchProperty.class) != null) {
return null;
}
@Override
protected Annotation findAutowiredAnnotation(AccessibleObject ao) {
if (ao.getAnnotation(BatchProperty.class) != null) {
return null;
}
return super.findAutowiredAnnotation(ao);
}
return super.findAutowiredAnnotation(ao);
}
}

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

@@ -75,32 +75,23 @@ public class PartitionParser {
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);
}
paserMapperElement(element, parserContext, properties);
parsePartitionPlan(element, parserContext, stepName, properties);
parseAnalyzerElement(element, parserContext, properties);
parseReducerElement(element, parserContext, factoryBeanProperties);
parseCollectorElement(element, parserContext, factoryBeanProperties,
properties);
Element analyzerElement = DomUtils.getChildElementByTagName(element, ANALYZER_ELEMENT);
String partitionHandlerBeanName = name + ".partitionHandler";
registry.registerBeanDefinition(partitionHandlerBeanName, partitionHandlerDefinition);
factoryBeanProperties.add("partitionHandler", new RuntimeBeanReference(partitionHandlerBeanName));
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);
}
}
private void parseCollectorElement(Element element,
ParserContext parserContext,
MutablePropertyValues factoryBeanProperties,
MutablePropertyValues properties) {
Element collectorElement = DomUtils.getChildElementByTagName(element, COLLECTOR_ELEMENT);
if(collectorElement != null) {
@@ -112,11 +103,40 @@ public class PartitionParser {
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 parseReducerElement(Element element,
ParserContext parserContext,
MutablePropertyValues factoryBeanProperties) {
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);
}
}
private void parseAnalyzerElement(Element element,
ParserContext parserContext, MutablePropertyValues 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);
}
}
private void paserMapperElement(Element element,
ParserContext parserContext, MutablePropertyValues properties) {
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);
}
}
private void registerCollectorAnalyzerQueue(ParserContext parserContext) {

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

@@ -25,7 +25,6 @@ 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;
@@ -189,8 +188,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
private void processFinishedPartitions(
final List<Future<StepExecution>> tasks,
final Set<StepExecution> result) throws InterruptedException,
ExecutionException, Exception {
final Set<StepExecution> result) throws Exception {
for(int i = 0; i < tasks.size(); i++) {
Future<StepExecution> curTask = tasks.get(i);

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

@@ -18,12 +18,11 @@ 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 javax.batch.operations.BatchRuntimeException;
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;
/**
@@ -35,22 +34,16 @@ import org.springframework.util.Assert;
* @author Michael Minella
* @since 3.0
*/
public class PartitionCollectorAdapter implements ChunkListener, InitializingBean {
public class PartitionCollectorAdapter implements ChunkListener {
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;
}
public PartitionCollectorAdapter(Queue<Serializable> queue, PartitionCollector collector) {
Assert.notNull(queue, "A thread safe Queue is required");
Assert.notNull(collector, "A PartitionCollector is required");
/**
* @param collector Provides partition specific information back to the {@link PartitionAnalyzer} as needed.
*/
public void setPartitionCollector(PartitionCollector collector) {
this.partitionQueue = queue;
this.collector = collector;
}
@@ -62,18 +55,12 @@ public class PartitionCollectorAdapter implements ChunkListener, InitializingBea
public void afterChunk(ChunkContext context) {
try {
partitionQueue.add(collector.collectPartitionData());
} catch (Exception e) {
throw new PartitionException(e);
} catch (Throwable e) {
throw new BatchRuntimeException("An error occured while collecting data from the PartionCollector", 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
View File

@@ -1,36 +0,0 @@
/*
* 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);
}
}

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

@@ -37,9 +37,11 @@ import org.springframework.batch.item.ExecutionContext;
public class PartitionStep extends org.springframework.batch.core.partition.support.PartitionStep {
private PartitionReducer reducer;
private boolean hasReducer = false;
public void setPartitionReducer(PartitionReducer reducer) {
this.reducer = reducer;
hasReducer = reducer != null;
}
/**
@@ -58,7 +60,7 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp
@Override
protected void doExecute(StepExecution stepExecution) throws Exception {
if(reducer != null) {
if(hasReducer) {
reducer.beginPartitionedStep();
}
@@ -66,9 +68,8 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp
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) {
if (hasReducer) {
reducer.rollbackPartitionedStep();
reducer.beforePartitionedStepCompletion();
reducer.afterPartitionedStepCompletion(PartitionStatus.ROLLBACK);
@@ -76,7 +77,7 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp
throw new JobExecutionException("Partition handler returned an unsuccessful step");
}
if (reducer != null) {
if (hasReducer) {
reducer.beforePartitionedStepCompletion();
reducer.afterPartitionedStepCompletion(PartitionStatus.COMMIT);
}

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

@@ -16,7 +16,6 @@
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;
@@ -24,50 +23,17 @@ 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() {
Queue<Serializable> dataQueue = new ConcurrentLinkedQueue<Serializable>();
adapter = new PartitionCollectorAdapter(dataQueue, new PartitionCollector() {
private int count = 0;
@@ -77,10 +43,6 @@ public class PartitionCollectorAdapterTests {
}
});
Queue<Serializable> dataQueue = new ConcurrentLinkedQueue<Serializable>();
adapter.setPartitionQueue(dataQueue);
adapter.afterPropertiesSet();
adapter.afterChunk(null);
adapter.afterChunk(null);
adapter.afterChunk(null);
@@ -90,10 +52,4 @@ public class PartitionCollectorAdapterTests {
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);
}
}