diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 489b4c4dc..8868443d0 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -293,7 +293,7 @@ public class StepParserStepFactoryBean 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 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); } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowExecutionException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowExecutionException.java index 896fcb01e..daf69d167 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowExecutionException.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowExecutionException.java @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ package org.springframework.batch.core.job.flow; * @author Dave Syer * */ +@SuppressWarnings("serial") public class FlowExecutionException extends Exception { /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrAutowiredAnnotationBeanPostProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrAutowiredAnnotationBeanPostProcessor.java index 27d6ba06e..ea55af716 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrAutowiredAnnotationBeanPostProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/support/JsrAutowiredAnnotationBeanPostProcessor.java @@ -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; /** *

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.

*/ 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); + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java index c99749003..376a45c6e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParser.java @@ -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) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java index 634ba437b..780fc17c9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java @@ -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> tasks, - final Set result) throws InterruptedException, - ExecutionException, Exception { + final Set result) throws Exception { for(int i = 0; i < tasks.size(); i++) { Future curTask = tasks.get(i); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java index ac49fe4dd..bfe102be4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapter.java @@ -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 partitionQueue; - /** - * @param queue destination for results of each {@link PartitionCollector#collectPartitionData()} call. - */ - public void setPartitionQueue(Queue queue) { - this.partitionQueue = queue; - } + public PartitionCollectorAdapter(Queue 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"); - } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java deleted file mode 100644 index 58f9993e7..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/PartitionException.java +++ /dev/null @@ -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); - } -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java index 4c1cd1e32..b1d2fc949 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java @@ -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); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java index 4ddcb27b0..1dfe1b6e5 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/partition/PartitionCollectorAdapterTests.java @@ -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()); - - adapter.afterPropertiesSet(); - } - @Test public void testAfterChunkSuccessful() throws Exception { - adapter.setPartitionCollector(new PartitionCollector() { + Queue dataQueue = new ConcurrentLinkedQueue(); + + adapter = new PartitionCollectorAdapter(dataQueue, new PartitionCollector() { private int count = 0; @@ -77,10 +43,6 @@ public class PartitionCollectorAdapterTests { } }); - Queue dataQueue = new ConcurrentLinkedQueue(); - 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); - } }