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 bc4b687f3..72d962449 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 @@ -309,7 +309,7 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa private Step createFaultTolerantStep() { - FaultTolerantStepBuilder builder = new FaultTolerantStepBuilder(new StepBuilder(name)); + FaultTolerantStepBuilder builder = getFaultTolerantStepBuilder(this.name); if (commitInterval != null) { builder.chunk(commitInterval); @@ -382,6 +382,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa } + protected FaultTolerantStepBuilder getFaultTolerantStepBuilder(String stepName) { + return new FaultTolerantStepBuilder(new StepBuilder(stepName)); + } + private void registerItemListeners(SimpleStepBuilder builder) { for (ItemReadListener listener : readListeners) { builder.listener(listener); @@ -396,7 +400,7 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa @SuppressWarnings("unchecked") private Step createSimpleStep() { - SimpleStepBuilder builder = new SimpleStepBuilder(new StepBuilder(name)); + SimpleStepBuilder builder = getSimpleStepBuilder(this.name); if(timeout != null && commitInterval != null) { CompositeCompletionPolicy completionPolicy = new CompositeCompletionPolicy(); @@ -420,6 +424,11 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanNameAwa return builder.build(); } + @SuppressWarnings("unchecked") + protected SimpleStepBuilder getSimpleStepBuilder(String stepName) { + return new SimpleStepBuilder(new StepBuilder(stepName)); + } + private TaskletStep createTaskletStep() { TaskletStepBuilder builder = new StepBuilder(name).tasklet(tasklet); enhanceTaskletStepBuilder(builder); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java index 20ca37308..0417821a2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/StepContextFactoryBean.java @@ -34,9 +34,9 @@ public class StepContextFactoryBean implements FactoryBean { @Autowired private BatchPropertyContext batchPropertyContext; - /* (non-Javadoc) - * @see org.springframework.beans.factory.FactoryBean#getObject() - */ + /* (non-Javadoc) + * @see org.springframework.beans.factory.FactoryBean#getObject() + */ @Override public StepContext getObject() throws Exception { org.springframework.batch.core.StepExecution stepExecution = StepSynchronizationManager.getContext().getStepExecution(); @@ -45,9 +45,9 @@ public class StepContextFactoryBean implements FactoryBean { return new StepContext(stepExecution, properties); } - /* (non-Javadoc) - * @see org.springframework.beans.factory.FactoryBean#getObjectType() - */ + /* (non-Javadoc) + * @see org.springframework.beans.factory.FactoryBean#getObjectType() + */ @Override public Class getObjectType() { return StepContext.class; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java index d85a465a4..f511bff80 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java @@ -16,6 +16,7 @@ package org.springframework.batch.core.jsr.configuration.xml; import org.springframework.batch.core.configuration.xml.CoreNamespaceUtils; +import org.springframework.batch.core.jsr.StepContextFactoryBean; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -28,7 +29,7 @@ import org.w3c.dom.Element; /** * Parses a <job /> tag as defined in JSR-352. Current state parses into * the standard Spring Batch artifacts. - * + * * @author Michael Minella * @author Chris Schaefer * @since 3.0 @@ -61,6 +62,13 @@ public class JobParser extends AbstractSingleBeanDefinitionParser { BeanDefinition flowDef = new FlowParser(jobName).parse(element, parserContext); builder.addPropertyValue("flow", flowDef); + AbstractBeanDefinition stepContextBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(StepContextFactoryBean.class) + .getBeanDefinition(); + + stepContextBeanDefinition.setScope("step"); + + parserContext.getRegistry().registerBeanDefinition("stepContextFactory", stepContextBeanDefinition); + new ListnerParser(JobListenerFactoryBean.class, "jobExecutionListeners").parseListeners(element, parserContext, builder); new PropertyParser("job-" + jobName, parserContext).parseProperties(element); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java index 7838115d3..da2b71210 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepFactoryBean.java @@ -16,6 +16,7 @@ package org.springframework.batch.core.jsr.configuration.xml; import javax.batch.api.Batchlet; +import javax.batch.api.chunk.CheckpointAlgorithm; import javax.batch.api.chunk.ItemProcessor; import javax.batch.api.chunk.ItemReader; import javax.batch.api.chunk.ItemWriter; @@ -23,23 +24,37 @@ import javax.batch.api.chunk.ItemWriter; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.xml.StepParserStepFactoryBean; import org.springframework.batch.core.jsr.step.batchlet.BatchletAdapter; +import org.springframework.batch.core.jsr.step.builder.JsrFaultTolerantStepBuilder; +import org.springframework.batch.core.jsr.step.builder.JsrSimpleStepBuilder; +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.tasklet.Tasklet; 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; /** * This {@link FactoryBean} is used by the JSR-352 namespace parser to create * {@link Step} objects. It stores all of the properties that are * configurable on the <step/>. - * + * * @author Michael Minella * @since 3.0 */ @SuppressWarnings({"rawtypes", "unchecked"}) public class StepFactoryBean extends StepParserStepFactoryBean { + /** + * Wraps a {@link Batchlet} in a {@link BatchletAdapter} if required for consumption + * by the rest of the framework. + * + * @param tasklet {@link Tasklet} or {@link Batchlet} implementation + * @throws IllegalArgumentException if tasklet does not implement either Tasklet or Batchlet + */ public void setTasklet(Object tasklet) { if(tasklet instanceof Tasklet) { super.setTasklet((Tasklet) tasklet); @@ -51,6 +66,13 @@ public class StepFactoryBean extends StepParserStepFactoryBean { } } + /** + * Wraps a {@link ItemReader} in a {@link ItemReaderAdapter} if required for consumption + * by the rest of the framework. + * + * @param itemReader {@link ItemReader} or {@link org.springframework.batch.item.ItemReader} implementation + * @throws IllegalArgumentException if itemReader does not implement either version of ItemReader + */ public void setItemReader(Object itemReader) { if(itemReader instanceof org.springframework.batch.item.ItemReader) { super.setItemReader((org.springframework.batch.item.ItemReader) itemReader); @@ -62,6 +84,13 @@ public class StepFactoryBean extends StepParserStepFactoryBean { } } + /** + * Wraps a {@link ItemProcessor} in a {@link ItemProcessorAdapter} if required for consumption + * by the rest of the framework. + * + * @param itemProcessor {@link ItemProcessor} or {@link org.springframework.batch.item.ItemProcessor} implementation + * @throws IllegalArgumentException if itemProcessor does not implement either version of ItemProcessor + */ public void setItemProcessor(Object itemProcessor) { if(itemProcessor instanceof org.springframework.batch.item.ItemProcessor) { super.setItemProcessor((org.springframework.batch.item.ItemProcessor) itemProcessor); @@ -73,6 +102,13 @@ public class StepFactoryBean extends StepParserStepFactoryBean { } } + /** + * Wraps a {@link ItemWriter} in a {@link ItemWriterAdapter} if required for consumption + * by the rest of the framework. + * + * @param itemWriter {@link ItemWriter} or {@link org.springframework.batch.item.ItemWriter} implementation + * @throws IllegalArgumentException if itemWriter does not implement either version of ItemWriter + */ public void setItemWriter(Object itemWriter) { if(itemWriter instanceof org.springframework.batch.item.ItemWriter) { super.setItemWriter((org.springframework.batch.item.ItemWriter) itemWriter); @@ -83,4 +119,32 @@ public class StepFactoryBean extends StepParserStepFactoryBean { "org.springframework.batch.item.ItemWriter or javax.batch.api.chunk.ItemWriter"); } } + + /** + * Wraps a {@link CheckpointAlgorithm} in a {@link CheckpointAlgorithmAdapter} if required for consumption + * by the rest of the framework. + * + * @param chunkCompletionPolicy {@link CompletionPolicy} or {@link CheckpointAlgorithm} implementation + * @throws IllegalArgumentException if chunkCompletionPolicy does not implement either CompletionPolicy or CheckpointAlgorithm + */ + public void setChunkCompletionPolicy(Object chunkCompletionPolicy) { + if(chunkCompletionPolicy instanceof CompletionPolicy) { + super.setChunkCompletionPolicy((CompletionPolicy) chunkCompletionPolicy); + } else if(chunkCompletionPolicy instanceof CheckpointAlgorithm) { + super.setChunkCompletionPolicy(new CheckpointAlgorithmAdapter((CheckpointAlgorithm) chunkCompletionPolicy)); + } else { + throw new IllegalArgumentException("The definition of a chunk completion policy must implement either " + + "org.springframework.batch.repeat.CompletionPolicy or javax.batch.api.chunk.CheckpointAlgorithm"); + } + } + + @Override + protected FaultTolerantStepBuilder getFaultTolerantStepBuilder(String stepName) { + return new JsrFaultTolerantStepBuilder(new StepBuilder(stepName)); + } + + @Override + protected SimpleStepBuilder getSimpleStepBuilder(String stepName) { + return new JsrSimpleStepBuilder(new StepBuilder(stepName)); + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java index 169ef8f24..3d8f404c8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/StepParser.java @@ -17,12 +17,7 @@ package org.springframework.batch.core.jsr.configuration.xml; import java.util.Collection; -import org.w3c.dom.Element; -import org.w3c.dom.Node; -import org.w3c.dom.NodeList; - import org.springframework.batch.core.job.flow.support.state.StepState; -import org.springframework.batch.core.jsr.StepContextFactoryBean; import org.springframework.batch.core.listener.StepListenerFactoryBean; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; @@ -31,10 +26,13 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.w3c.dom.NodeList; /** * Parser for the <step /> element defined by JSR-352. - * + * * @author Michael Minella * @author Glenn Renfro * @author Chris Schaefer @@ -91,13 +89,6 @@ public class StepParser extends AbstractSingleBeanDefinitionParser { } } - AbstractBeanDefinition stepContextBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(StepContextFactoryBean.class) - .getBeanDefinition(); - - stepContextBeanDefinition.setScope("step"); - - parserContext.getRegistry().registerBeanDefinition(stepName + "stepContext", stepContextBeanDefinition); - return FlowParser.getNextElements(parserContext, stepName, stateBuilder.getBeanDefinition(), element); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java new file mode 100644 index 000000000..641dee111 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrFaultTolerantStepBuilder.java @@ -0,0 +1,79 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.core.jsr.step.builder; + +import java.util.ArrayList; + +import org.springframework.batch.core.StepListener; +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.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; + +/** + * A step builder that extends the {@link FaultTolerantStepBuilder} to create JSR-352 + * specific {@link ChunkProvider} and {@link ChunkProcessor} supporting both the chunking + * pattern defined by the spec as well as skip/retry logic. + * + * @author Michael Minella + * + * @param The input type for the step + * @param The output type for the step + */ +public class JsrFaultTolerantStepBuilder extends FaultTolerantStepBuilder { + + public JsrFaultTolerantStepBuilder(StepBuilder parent) { + super(parent); + } + + @Override + public FaultTolerantStepBuilder faultTolerant() { + return this; + } + + @Override + protected ChunkProvider createChunkProvider() { + return new JsrChunkProvider(); + } + + /** + * Provides a JSR-352 specific implementation of a {@link ChunkProcessor} for use + * within the {@link ChunkOrientedTasklet} + * + * @return a JSR-352 implementation of the {@link ChunkProcessor} + * @see {@link JsrFaultTolerantChunkProcessor} + */ + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + protected ChunkProcessor createChunkProcessor() { + SkipPolicy skipPolicy = createSkipPolicy(); + skipPolicy = getFatalExceptionAwareProxy(skipPolicy); + JsrFaultTolerantChunkProcessor chunkProcessor = new JsrFaultTolerantChunkProcessor(getReader(), getProcessor(), + getWriter(), createChunkOperations(), createRetryOperations()); + chunkProcessor.setSkipPolicy(skipPolicy); + chunkProcessor.setRollbackClassifier(getRollbackClassifier()); + detectStreamInReader(); + chunkProcessor.setChunkMonitor(getChunkMonitor()); + ArrayList listeners = new ArrayList(getItemListeners()); + chunkProcessor.setListeners(listeners); + + return chunkProcessor; + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java new file mode 100644 index 000000000..5781a6d17 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/builder/JsrSimpleStepBuilder.java @@ -0,0 +1,70 @@ +/* + * 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 java.util.ArrayList; + +import org.springframework.batch.core.StepListener; +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.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.repeat.RepeatOperations; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.util.Assert; + +/** + * A step builder that extends the {@link FaultTolerantStepBuilder} to create JSR-352 + * specific {@link ChunkProvider} and {@link ChunkProcessor} supporting the chunking + * pattern defined by the spec. + * + * @author Michael Minella + * + * @param The input type for the step + * @param The output type for the step + */ +public class JsrSimpleStepBuilder extends SimpleStepBuilder { + + public JsrSimpleStepBuilder(StepBuilder parent) { + super(parent); + } + + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + protected Tasklet createTasklet() { + Assert.state(getReader() != null, "ItemReader must be provided"); + Assert.state(getProcessor() != null || getWriter() != null, "ItemWriter or ItemProcessor must be provided"); + RepeatOperations repeatOperations = createRepeatOperations(); + ChunkProvider chunkProvider = new JsrChunkProvider(); + JsrChunkProcessor chunkProcessor = new JsrChunkProcessor(getReader(), getProcessor(), getWriter(), repeatOperations); + chunkProcessor.setListeners(new ArrayList(getItemListeners())); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + tasklet.setBuffering(!isReaderTransactionalQueue()); + return tasklet; + } + + private RepeatOperations createRepeatOperations() { + RepeatTemplate repeatOperations = new RepeatTemplate(); + repeatOperations.setCompletionPolicy(getChunkCompletionPolicy()); + repeatOperations.setExceptionHandler(getExceptionHandler()); + return repeatOperations; + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessor.java new file mode 100644 index 000000000..138872ad1 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessor.java @@ -0,0 +1,256 @@ +/* + * 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.item; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.listener.MulticasterBatchListener; +import org.springframework.batch.core.step.item.Chunk; +import org.springframework.batch.core.step.item.ChunkProcessor; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.RepeatStatus; + +/** + * {@link ChunkProcessor} implementation that implements JSR-352's chunking pattern + * (read and process in a loop until the chunk is complete then write). This + * implementation is responsible for all three phases of chunk based processing + * (reading, processing and writing). + * + * @author Michael Minella + * + * @param The input type for the step + * @param The output type for the step + */ +public class JsrChunkProcessor implements ChunkProcessor { + + private final Log logger = LogFactory.getLog(getClass()); + private ItemReader itemReader; + @SuppressWarnings("rawtypes") + private final MulticasterBatchListener listener = new MulticasterBatchListener(); + private RepeatOperations repeatTemplate; + private ItemProcessor itemProcessor; + private ItemWriter itemWriter; + + public JsrChunkProcessor() { + this(null, null, null, null); + } + + public JsrChunkProcessor(ItemReader reader, ItemProcessor processor, ItemWriter writer, RepeatOperations repeatTemplate) { + this.itemReader = reader; + this.itemProcessor = processor; + this.itemWriter = writer; + this.repeatTemplate = repeatTemplate; + } + + @SuppressWarnings("rawtypes") + protected MulticasterBatchListener getListener() { + return listener; + } + + /** + * Loops through reading (via {@link #provide(StepContribution, Chunk)} and + * processing (via {@link #transform(StepContribution, Object)}) until the chunk + * is complete. Once the chunk is complete, the results are written (via + * {@link #persist(StepContribution, Chunk)}. + * + * @see ChunkProcessor#process(StepContribution, Chunk) + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + */ + @Override + public void process(final StepContribution contribution, final Chunk chunk) + throws Exception { + + final AtomicInteger filterCount = new AtomicInteger(0); + final Chunk output = new Chunk(); + + repeatTemplate.iterate(new RepeatCallback() { + + @Override + public RepeatStatus doInIteration(RepeatContext context) throws Exception { + I item = provide(contribution, chunk); + + if(item != null) { + contribution.incrementReadCount(); + } else { + return RepeatStatus.FINISHED; + } + + O processedItem = transform(contribution, item); + + if(item != null && processedItem == null) { + filterCount.incrementAndGet(); + } else { + output.add(processedItem); + } + + return RepeatStatus.CONTINUABLE; + } + }); + + contribution.incrementFilterCount(filterCount.get()); + if(output.size() > 0) { + persist(contribution, output); + } + } + + /** + * Register some {@link StepListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage. + * + * @param listeners list of listeners to be used within this step + */ + public void setListeners(List listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a process. + * + * @param listener a {@link StepListener} + */ + public void registerListener(StepListener listener) { + this.listener.register(listener); + } + + /** + * Responsible for the reading portion of the chunking loop. In this implementation, delegates + * to {@link #doProvide(StepContribution, Chunk)} + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @return an item + * @throws Exception + */ + protected I provide(final StepContribution contribution, final Chunk chunk) throws Exception { + return doProvide(contribution, chunk); + } + + /** + * Implements reading as well as any related listener calls required. + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @return an item + * @throws Exception + */ + @SuppressWarnings("unchecked") + protected final I doProvide(final StepContribution contribution, final Chunk chunk) throws Exception { + try { + listener.beforeRead(); + I item = itemReader.read(); + if(item != null) { + listener.afterRead(item); + } else { + chunk.setEnd(); + } + + return item; + } + catch (Exception e) { + logger.debug(e.getMessage() + " : " + e.getClass().getName()); + listener.onReadError(e); + throw e; + } + } + + /** + * Responsible for the processing portion of the chunking loop. In this implementation, delegates to the + * {@link #doTransform(Object)} if a processor is available (returns the item unmodified if it is not) + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @return a processed item if a processor is present (the unmodified item if it is not) + * @throws Exception + */ + protected O transform(final StepContribution contribution, final I item) throws Exception { + if (itemProcessor == null) { + @SuppressWarnings("unchecked") + O result = (O) item; + return result; + } + + return doTransform(item); + } + + /** + * Implements processing and all related listener calls. + * + * @param item the item to be processed + * @return the processed item + * @throws Exception + */ + @SuppressWarnings("unchecked") + protected final O doTransform(I item) throws Exception { + try { + listener.beforeProcess(item); + O result = itemProcessor.process(item); + listener.afterProcess(item, result); + return result; + } + catch (Exception e) { + listener.onProcessError(item, e); + throw e; + } + } + + /** + * Responsible for the writing portion of the chunking loop. In this implementation, delegates to the + * {{@link #doPersist(StepContribution, Chunk)}. + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @throws Exception + */ + protected void persist(final StepContribution contribution, final Chunk chunk) throws Exception { + doPersist(contribution, chunk); + + contribution.incrementWriteCount(chunk.getItems().size()); + } + + /** + * Implements writing and all related listener calls + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @throws Exception + */ + @SuppressWarnings("unchecked") + protected final void doPersist(final StepContribution contribution, final Chunk chunk) throws Exception { + try { + List items = chunk.getItems(); + listener.beforeWrite(items); + itemWriter.write(items); + listener.afterWrite(items); + } + catch (Exception e) { + listener.onWriteError(e, chunk.getItems()); + throw e; + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProvider.java new file mode 100644 index 000000000..01e8dfd14 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrChunkProvider.java @@ -0,0 +1,47 @@ +/* + * 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.item; + +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.item.Chunk; +import org.springframework.batch.core.step.item.ChunkProvider; + +/** + * A no-op {@link ChunkProvider}. The JSR-352 chunking model does not cache the + * input as the regular Spring Batch implementations do so this component is not + * needed within a chunking loop. + * + * @author Michael Minella + * + * @param The type of input for the step + */ +public class JsrChunkProvider implements ChunkProvider { + + /* (non-Javadoc) + * @see org.springframework.batch.core.step.item.ChunkProvider#provide(org.springframework.batch.core.StepContribution) + */ + @Override + public Chunk provide(StepContribution contribution) throws Exception { + return new Chunk(); + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.step.item.ChunkProvider#postProcess(org.springframework.batch.core.StepContribution, org.springframework.batch.core.step.item.Chunk) + */ + @Override + public void postProcess(StepContribution contribution, Chunk chunk) { + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessor.java new file mode 100644 index 000000000..35b283d4c --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessor.java @@ -0,0 +1,342 @@ +/* + * 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.item; + +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.step.item.BatchRetryTemplate; +import org.springframework.batch.core.step.item.Chunk; +import org.springframework.batch.core.step.item.ChunkMonitor; +import org.springframework.batch.core.step.item.ForceRollbackForWriteSkipException; +import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; +import org.springframework.batch.core.step.skip.NonSkippableProcessException; +import org.springframework.batch.core.step.skip.SkipException; +import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.core.step.skip.SkipPolicyFailedException; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.classify.BinaryExceptionClassifier; +import org.springframework.classify.Classifier; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryException; +import org.springframework.util.Assert; + +/** + * Extension of the {@link JsrChunkProcessor} that adds skip and retry functionality. + * + * @author Michael Minella + * + * @param input type for the step + * @param output type for the step + */ +public class JsrFaultTolerantChunkProcessor extends JsrChunkProcessor { + + protected final Log logger = LogFactory.getLog(getClass()); + private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(); + private Classifier rollbackClassifier = new BinaryExceptionClassifier(true); + private final BatchRetryTemplate batchRetryTemplate; + private ChunkMonitor chunkMonitor = new ChunkMonitor(); + private boolean hasProcessor = false; + + public JsrFaultTolerantChunkProcessor() { + this(null, null, null, null, null); + } + + public JsrFaultTolerantChunkProcessor(ItemReader reader, ItemProcessor processor, ItemWriter writer, RepeatOperations repeatTemplate, BatchRetryTemplate batchRetryTemplate) { + super(reader, processor, writer, repeatTemplate); + hasProcessor = processor != null; + this.batchRetryTemplate = batchRetryTemplate; + } + + /** + * @param skipPolicy a {@link SkipPolicy} + */ + public void setSkipPolicy(SkipPolicy skipPolicy) { + Assert.notNull(skipPolicy, "A skip policy is required"); + + this.skipPolicy = skipPolicy; + } + + /** + * @param rollbackClassifier a {@link Classifier} + */ + public void setRollbackClassifier(Classifier rollbackClassifier) { + Assert.notNull(rollbackClassifier, "A rollbackClassifier is required"); + + this.rollbackClassifier = rollbackClassifier; + } + + /** + * @param chunkMonitor a {@link ChunkMonitor} + */ + public void setChunkMonitor(ChunkMonitor chunkMonitor) { + Assert.notNull(chunkMonitor, "A chunkMonitor is required"); + + this.chunkMonitor = chunkMonitor; + } + + /** + * Register some {@link StepListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage. + * + * @param listeners + */ + @Override + public void setListeners(List listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a process. + * + * @param listener a {@link StepListener} + */ + @Override + public void registerListener(StepListener listener) { + getListener().register(listener); + } + + /** + * Adds retry and skip logic to the reading phase of the chunk loop. + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @return I an item + * @throws Exception + */ + @Override + protected I provide(final StepContribution contribution, final Chunk chunk) throws Exception { + RetryCallback retryCallback = new RetryCallback() { + + @Override + public I doWithRetry(RetryContext arg0) throws Exception { + while (true) { + try { + return doProvide(contribution, chunk); + } + catch (Exception e) { + if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) { + + // increment skip count and try again + contribution.incrementReadSkipCount(); + chunk.skip(e); + + logger.debug("Skipping failed input", e); + } + else { + if (rollbackClassifier.classify(e)) { + throw e; + } + logger.debug("No-rollback for non-skippable exception (ignored)", e); + } + } + } + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + @Override + public I recover(RetryContext context) throws Exception { + Throwable e = context.getLastThrowable(); + if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) { + contribution.incrementReadSkipCount(); + logger.debug("Skipping after failed process", e); + return null; + } + else { + if (rollbackClassifier.classify(e)) { + // Default is to rollback unless the classifier + // allows us to continue + throw new RetryException("Non-skippable exception in recoverer while reading", e); + } + return null; + } + } + + }; + + return batchRetryTemplate.execute(retryCallback, recoveryCallback); + } + + /** + * Convenience method for calling process skip policy. + * + * @param policy the skip policy + * @param e the cause of the skip + * @param skipCount the current skip count + */ + private boolean shouldSkip(SkipPolicy policy, Throwable e, int skipCount) { + try { + return policy.shouldSkip(e, skipCount); + } + catch (SkipException ex) { + throw ex; + } + catch (RuntimeException ex) { + throw new SkipPolicyFailedException("Fatal exception in SkipPolicy.", ex, e); + } + } + + /** + * Adds retry and skip logic to the process phase of the chunk loop. + * + * @param contribution a {@link StepContribution} + * @param item an item to be processed + * @return O an item that has been processed if a processor is available + * @throws Exception + */ + @Override + @SuppressWarnings("unchecked") + protected O transform(final StepContribution contribution, final I item) throws Exception { + if (!hasProcessor) { + O result = (O) item; + return result; + } + + RetryCallback retryCallback = new RetryCallback() { + + @Override + public O doWithRetry(RetryContext context) throws Exception { + try { + return doTransform(item); + } + catch (Exception e) { + if (rollbackClassifier.classify(e)) { + // Default is to rollback unless the classifier + // allows us to continue + throw e; + } + else if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) { + // If we are not re-throwing then we should check if + // this is skippable + contribution.incrementProcessSkipCount(); + logger.debug("Skipping after failed process with no rollback", e); + // If not re-throwing then the listener will not be + // called in next chunk. + getListener().onSkipInProcess(item, e); + } + else { + // If it's not skippable that's an error in + // configuration - it doesn't make sense to not roll + // back if we are also not allowed to skip + throw new NonSkippableProcessException( + "Non-skippable exception in processor. Make sure any exceptions that do not cause a rollback are skippable.", + e); + } + } + return null; + } + + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + @Override + public O recover(RetryContext context) throws Exception { + Throwable e = context.getLastThrowable(); + if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) { + contribution.incrementProcessSkipCount(); + logger.debug("Skipping after failed process", e); + return null; + } + else { + if (rollbackClassifier.classify(e)) { + // Default is to rollback unless the classifier + // allows us to continue + throw new RetryException("Non-skippable exception in recoverer while processing", e); + } + return null; + } + } + + }; + + return batchRetryTemplate.execute(retryCallback, recoveryCallback); + } + + /** + * Adds retry and skip logic to the write phase of the chunk loop. + * + * @param contribution a {@link StepContribution} + * @param chunk a {@link Chunk} + * @throws Exception + */ + @Override + protected void persist(final StepContribution contribution, final Chunk chunk) throws Exception { + + RetryCallback retryCallback = new RetryCallback() { + @Override + public Object doWithRetry(RetryContext context) throws Exception { + + chunkMonitor.setChunkSize(chunk.size()); + try { + doPersist(contribution, chunk); + } + catch (Exception e) { + if (rollbackClassifier.classify(e)) { + throw e; + } + /* + * If the exception is marked as no-rollback, we need to + * override that, otherwise there's no way to write the + * rest of the chunk or to honour the skip listener + * contract. + */ + throw new ForceRollbackForWriteSkipException( + "Force rollback on skippable exception so that skipped item can be located.", e); + } + contribution.incrementWriteCount(chunk.size()); + return null; + + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + @Override + public O recover(RetryContext context) throws Exception { + Throwable e = context.getLastThrowable(); + if (shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + logger.debug("Skipping after failed write", e); + return null; + } + else { + if (rollbackClassifier.classify(e)) { + // Default is to rollback unless the classifier + // allows us to continue + throw new RetryException("Non-skippable exception in recoverer while write", e); + } + return null; + } + } + + }; + + batchRetryTemplate.execute(retryCallback, recoveryCallback); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index 423079821..501f10e8f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -35,6 +35,8 @@ import org.springframework.batch.core.step.FatalStepExecutionException; import org.springframework.batch.core.step.item.BatchRetryTemplate; import org.springframework.batch.core.step.item.ChunkMonitor; 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.item.FaultTolerantChunkProcessor; import org.springframework.batch.core.step.item.FaultTolerantChunkProvider; import org.springframework.batch.core.step.item.ForceRollbackForWriteSkipException; @@ -166,8 +168,8 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { Assert.state(getProcessor() != null || getWriter() != null, "ItemWriter or ItemProcessor must be provided"); addSpecialExceptions(); registerSkipListeners(); - FaultTolerantChunkProvider chunkProvider = createChunkProvider(); - FaultTolerantChunkProcessor chunkProcessor = createChunkProcessor(); + ChunkProvider chunkProvider = createChunkProvider(); + ChunkProcessor chunkProcessor = createChunkProcessor(); ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); tasklet.setBuffering(!isReaderTransactionalQueue()); return tasklet; @@ -382,7 +384,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { return this; } - private FaultTolerantChunkProvider createChunkProvider() { + protected ChunkProvider createChunkProvider() { SkipPolicy readSkipPolicy = createSkipPolicy(); readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy); @@ -399,7 +401,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { } - private FaultTolerantChunkProcessor createChunkProcessor() { + protected ChunkProcessor createChunkProcessor() { BatchRetryTemplate batchRetryTemplate = createRetryOperations(); @@ -435,7 +437,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class); } - private void detectStreamInReader() { + protected void detectStreamInReader() { if (streamIsReader) { if (!concurrent()) { chunkMonitor.setItemReader(getReader()); @@ -470,7 +472,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { * * @return an exception classifier: maps to true if an exception should cause rollback */ - private Classifier getRollbackClassifier() { + protected Classifier getRollbackClassifier() { Classifier classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false); @@ -534,7 +536,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { /** * @return fully configured retry template for item processing phase. */ - private BatchRetryTemplate createRetryOperations() { + protected BatchRetryTemplate createRetryOperations() { RetryPolicy retryPolicy = this.retryPolicy; SimpleRetryPolicy simpleRetryPolicy = null; @@ -582,6 +584,10 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { } + protected ChunkMonitor getChunkMonitor() { + return this.chunkMonitor; + } + /** * Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never retries explicitly non-retryable * exceptions. @@ -610,7 +616,7 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { * @param skipPolicy an existing skip policy * @return a skip policy that will not skip fatal exceptions */ - private SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) { + protected SkipPolicy getFatalExceptionAwareProxy(SkipPolicy skipPolicy) { NeverSkipItemSkipPolicy neverSkipPolicy = new NeverSkipItemSkipPolicy(); Map, SkipPolicy> map = new HashMap, SkipPolicy>(); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java index 697f86612..df1ad1002 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java @@ -311,7 +311,7 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder 0), "You must specify either a chunkCompletionPolicy or a commitInterval but not both."); Assert.state(chunkSize >= 0, "The commitInterval must be positive or zero (for default value)."); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java index 04493a01f..c7e177b07 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/JobPropertyTests.java @@ -15,19 +15,24 @@ */ package org.springframework.batch.core.jsr.configuration.xml; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + import java.io.Serializable; import java.util.List; -import java.util.Properties; + import javax.batch.api.BatchProperty; import javax.batch.api.Batchlet; import javax.batch.api.chunk.ItemProcessor; import javax.batch.api.chunk.ItemReader; import javax.batch.api.chunk.ItemWriter; import javax.inject.Inject; + import junit.framework.Assert; + import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.aop.framework.Advised; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; @@ -35,9 +40,7 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.batch.core.job.flow.JobExecutionDecider; -import org.springframework.batch.core.jsr.StepContext; import org.springframework.batch.core.launch.JobLauncher; -import org.springframework.batch.core.scope.context.StepSynchronizationManager; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; @@ -45,9 +48,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; /** *

@@ -59,500 +59,483 @@ import static org.junit.Assert.assertNull; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) public class JobPropertyTests { - @Autowired - private TestItemReader testItemReader; - - @Autowired - private Job job; - - @Autowired - private JobLauncher jobLauncher; - - @Autowired - private TestItemProcessor testItemProcessor; - - @Autowired - private TestItemWriter testItemWriter; - - @Autowired - private TestCheckpointAlgorithm testCheckpointAlgorithm; - - @Autowired - private TestDecider testDecider; - - @Autowired - private TestStepListener testStepListener; - - @Autowired - private TestBatchlet testBatchlet; - - @Autowired - private ApplicationContext applicationContext; - - @Test - public void testJobLevelPropertiesInItemReader() throws Exception { - assertEquals("jobPropertyValue1", testItemReader.getJobPropertyName1()); - assertEquals("jobPropertyValue2", testItemReader.getJobPropertyName2()); - } - - @Test - public void testStepContextProperties() throws Exception { - JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); - - Properties step1Properties = new Properties(); - Properties step2Properties = new Properties(); - - for (StepExecution stepExecution : jobExecution.getStepExecutions()) { - try { - StepSynchronizationManager.register(stepExecution); - String contextBeanName = stepExecution.getStepName() + "stepContext"; - - // fix me? StepSynchronizationManager.context returns org.springframework.batch.core.scope.context.StepContext - StepContext stepContext = (StepContext) ((Advised)applicationContext.getBean(contextBeanName)).getTargetSource().getTarget(); - - if(contextBeanName.startsWith("step1")) { - step1Properties.putAll(stepContext.getProperties()); - } else { - step2Properties.putAll(stepContext.getProperties()); - } - } finally { - StepSynchronizationManager.close(); - } - } - - assertEquals(4, step1Properties.size()); - assertEquals("step1PropertyValue1", step1Properties.getProperty("step1PropertyName1")); - assertEquals("step1PropertyValue2", step1Properties.getProperty("step1PropertyName2")); - assertEquals("jobPropertyValue1", step1Properties.getProperty("jobPropertyName1")); - assertEquals("jobPropertyValue2", step1Properties.getProperty("jobPropertyName2")); - - assertEquals(4, step2Properties.size()); - assertEquals("step2PropertyValue1", step2Properties.getProperty("step2PropertyName1")); - assertEquals("step2PropertyValue2", step2Properties.getProperty("step2PropertyName2")); - assertEquals("jobPropertyValue1", step2Properties.getProperty("jobPropertyName1")); - assertEquals("jobPropertyValue2", step2Properties.getProperty("jobPropertyName2")); - - assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus()); - } - - @Test - public void testItemReaderProperties() throws Exception { - assertEquals("readerPropertyValue1", testItemReader.getReaderPropertyName1()); - assertEquals("readerPropertyValue2", testItemReader.getReaderPropertyName2()); - assertEquals("annotationNamedReaderPropertyValue", testItemReader.getAnnotationNamedProperty()); - assertNull(testItemReader.getNotDefinedProperty()); - assertNull(testItemReader.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testItemProcessorProperties() throws Exception { - Assert.assertEquals("processorPropertyValue1", testItemProcessor.getProcessorPropertyName1()); - Assert.assertEquals("processorPropertyValue2", testItemProcessor.getProcessorPropertyName2()); - assertEquals("annotationNamedProcessorPropertyValue", testItemProcessor.getAnnotationNamedProperty()); - assertNull(testItemProcessor.getNotDefinedProperty()); - assertNull(testItemProcessor.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testItemWriterProperties() throws Exception { - Assert.assertEquals("writerPropertyValue1", testItemWriter.getWriterPropertyName1()); - Assert.assertEquals("writerPropertyValue2", testItemWriter.getWriterPropertyName2()); - assertEquals("annotationNamedWriterPropertyValue", testItemWriter.getAnnotationNamedProperty()); - assertNull(testItemWriter.getNotDefinedProperty()); - assertNull(testItemWriter.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testCheckpointAlgorithmProperties() throws Exception { - Assert.assertEquals("algorithmPropertyValue1", testCheckpointAlgorithm.getAlgorithmPropertyName1()); - Assert.assertEquals("algorithmPropertyValue2", testCheckpointAlgorithm.getAlgorithmPropertyName2()); - assertEquals("annotationNamedAlgorithmPropertyValue", testCheckpointAlgorithm.getAnnotationNamedProperty()); - assertNull(testCheckpointAlgorithm.getNotDefinedProperty()); - assertNull(testCheckpointAlgorithm.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testDeciderProperties() throws Exception { - Assert.assertEquals("deciderPropertyValue1", testDecider.getDeciderPropertyName1()); - Assert.assertEquals("deciderPropertyValue2", testDecider.getDeciderPropertyName2()); - assertEquals("annotationNamedDeciderPropertyValue", testDecider.getAnnotationNamedProperty()); - assertNull(testDecider.getNotDefinedProperty()); - assertNull(testDecider.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testStepListenerProperties() throws Exception { - Assert.assertEquals("stepListenerPropertyValue1", testStepListener.getStepListenerPropertyName1()); - Assert.assertEquals("stepListenerPropertyValue2", testStepListener.getStepListenerPropertyName2()); - assertEquals("annotationNamedStepListenerPropertyValue", testStepListener.getAnnotationNamedProperty()); - assertNull(testStepListener.getNotDefinedProperty()); - assertNull(testStepListener.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testBatchletProperties() throws Exception { - Assert.assertEquals("batchletPropertyValue1", testBatchlet.getBatchletPropertyName1()); - Assert.assertEquals("batchletPropertyValue2", testBatchlet.getBatchletPropertyName2()); - assertEquals("annotationNamedBatchletPropertyValue", testBatchlet.getAnnotationNamedProperty()); - assertNull(testBatchlet.getNotDefinedProperty()); - assertNull(testBatchlet.getNotDefinedAnnotationNamedProperty()); - } - - @Test - public void testFieldWithInjectAnnotationOnlyInjects() throws Exception { - assertNotNull(testItemReader.getInjectAnnotatedOnlyField()); - assertEquals("Chris", testItemReader.getInjectAnnotatedOnlyField().getName()); - } - - @Test - public void testFieldWithBatchPropertyAnnotationOnlyNoInjection() throws Exception { - assertNull(testItemReader.getBatchAnnotatedOnlyField()); - } - - public static final class TestItemReader implements ItemReader { - private int cnt; - - @Inject @BatchProperty String readerPropertyName1; - @Inject @BatchProperty String readerPropertyName2; - @Inject @BatchProperty(name = "annotationNamedReaderPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - @Inject @BatchProperty String jobPropertyName1; - @Inject @BatchProperty String jobPropertyName2; - @Inject InjectTestObj injectAnnotatedOnlyField; - @BatchProperty String batchAnnotatedOnlyField; - - @Override - public void open(Serializable serializable) throws Exception { - } - - @Override - public void close() throws Exception { - } - - @Override - public Object readItem() throws Exception { - if (cnt == 0) { - cnt++; - return "blah"; - } - - return null; - } - - @Override - public Serializable checkpointInfo() throws Exception { - return null; - } - - String getReaderPropertyName1() { - return readerPropertyName1; - } - - String getReaderPropertyName2() { - return readerPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - - String getJobPropertyName1() { - return jobPropertyName1; - } - - String getJobPropertyName2() { - return jobPropertyName2; - } - - InjectTestObj getInjectAnnotatedOnlyField() { - return injectAnnotatedOnlyField; - } - - String getBatchAnnotatedOnlyField() { - return batchAnnotatedOnlyField; - } - } - - public static final class TestItemProcessor implements ItemProcessor { - @Inject @BatchProperty String processorPropertyName1; - @Inject @BatchProperty String processorPropertyName2; - @Inject @BatchProperty(name = "annotationNamedProcessorPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public Object processItem(Object o) throws Exception { - return o; - } - - String getProcessorPropertyName1() { - return processorPropertyName1; - } - - String getProcessorPropertyName2() { - return processorPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static final class TestItemWriter implements ItemWriter { - @Inject @BatchProperty String writerPropertyName1; - @Inject @BatchProperty String writerPropertyName2; - @Inject @BatchProperty(name = "annotationNamedWriterPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public void open(Serializable serializable) throws Exception { - } - - @Override - public void close() throws Exception { - } - - @Override - public void writeItems(List objects) throws Exception { - System.out.println(objects); - } - - @Override - public Serializable checkpointInfo() throws Exception { - return null; - } - - String getWriterPropertyName1() { - return writerPropertyName1; - } - - String getWriterPropertyName2() { - return writerPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static final class TestCheckpointAlgorithm implements CompletionPolicy { - @Inject @BatchProperty String algorithmPropertyName1; - @Inject @BatchProperty String algorithmPropertyName2; - @Inject @BatchProperty(name = "annotationNamedAlgorithmPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public boolean isComplete(RepeatContext context, RepeatStatus result) { - return true; - } - - @Override - public boolean isComplete(RepeatContext context) { - return true; - } - - @Override - public RepeatContext start(RepeatContext parent) { - return parent; - } - - @Override - public void update(RepeatContext context) { - } - - String getAlgorithmPropertyName1() { - return algorithmPropertyName1; - } - - String getAlgorithmPropertyName2() { - return algorithmPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static class TestDecider implements JobExecutionDecider { - @Inject @BatchProperty String deciderPropertyName1; - @Inject @BatchProperty String deciderPropertyName2; - @Inject @BatchProperty(name = "annotationNamedDeciderPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public FlowExecutionStatus decide(JobExecution jobExecution, - StepExecution stepExecution) { - return new FlowExecutionStatus("step2"); - } - - String getDeciderPropertyName1() { - return deciderPropertyName1; - } - - String getDeciderPropertyName2() { - return deciderPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static class TestStepListener implements javax.batch.api.chunk.listener.ItemReadListener, - javax.batch.api.chunk.listener.ItemProcessListener, javax.batch.api.chunk.listener.ItemWriteListener { - @Inject @BatchProperty String stepListenerPropertyName1; - @Inject @BatchProperty String stepListenerPropertyName2; - @Inject @BatchProperty(name = "annotationNamedStepListenerPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public void beforeProcess(Object o) throws Exception { - } - - @Override - public void afterProcess(Object o, Object o2) throws Exception { - } - - @Override - public void onProcessError(Object o, Exception e) throws Exception { - } - - @Override - public void beforeRead() throws Exception { - } - - @Override - public void afterRead(Object o) throws Exception { - } - - @Override - public void onReadError(Exception e) throws Exception { - } - - @Override - public void beforeWrite(List objects) throws Exception { - } - - @Override - public void afterWrite(List objects) throws Exception { - } - - @Override - public void onWriteError(List objects, Exception e) throws Exception { - } - - String getStepListenerPropertyName1() { - return stepListenerPropertyName1; - } - - String getStepListenerPropertyName2() { - return stepListenerPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static class TestBatchlet implements Batchlet { - @Inject @BatchProperty String batchletPropertyName1; - @Inject @BatchProperty String batchletPropertyName2; - @Inject @BatchProperty(name = "annotationNamedBatchletPropertyName") String annotationNamedProperty; - @Inject @BatchProperty String notDefinedProperty; - @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; - - @Override - public String process() throws Exception { - return null; - } - - @Override - public void stop() throws Exception { - } - - String getBatchletPropertyName1() { - return batchletPropertyName1; - } - - String getBatchletPropertyName2() { - return batchletPropertyName2; - } - - String getAnnotationNamedProperty() { - return annotationNamedProperty; - } - - String getNotDefinedProperty() { - return notDefinedProperty; - } - - String getNotDefinedAnnotationNamedProperty() { - return notDefinedAnnotationNamedProperty; - } - } - - public static class InjectTestObj { - private String name; - - public InjectTestObj(String name) { - this.name = name; - } - - public String getName() { - return name; - } - } + @Autowired + private TestItemReader testItemReader; + + @Autowired + private Job job; + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private TestItemProcessor testItemProcessor; + + @Autowired + private TestItemWriter testItemWriter; + + @Autowired + private TestCheckpointAlgorithm testCheckpointAlgorithm; + + @Autowired + private TestDecider testDecider; + + @Autowired + private TestStepListener testStepListener; + + @Autowired + private TestBatchlet testBatchlet; + + @Autowired + private ApplicationContext applicationContext; + + @Test + public void testJobLevelPropertiesInItemReader() throws Exception { + assertEquals("jobPropertyValue1", testItemReader.getJobPropertyName1()); + assertEquals("jobPropertyValue2", testItemReader.getJobPropertyName2()); + } + + @Test + public void testStepContextProperties() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus()); + } + + @Test + public void testItemReaderProperties() throws Exception { + assertEquals("readerPropertyValue1", testItemReader.getReaderPropertyName1()); + assertEquals("readerPropertyValue2", testItemReader.getReaderPropertyName2()); + assertEquals("annotationNamedReaderPropertyValue", testItemReader.getAnnotationNamedProperty()); + assertNull(testItemReader.getNotDefinedProperty()); + assertNull(testItemReader.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testItemProcessorProperties() throws Exception { + Assert.assertEquals("processorPropertyValue1", testItemProcessor.getProcessorPropertyName1()); + Assert.assertEquals("processorPropertyValue2", testItemProcessor.getProcessorPropertyName2()); + assertEquals("annotationNamedProcessorPropertyValue", testItemProcessor.getAnnotationNamedProperty()); + assertNull(testItemProcessor.getNotDefinedProperty()); + assertNull(testItemProcessor.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testItemWriterProperties() throws Exception { + Assert.assertEquals("writerPropertyValue1", testItemWriter.getWriterPropertyName1()); + Assert.assertEquals("writerPropertyValue2", testItemWriter.getWriterPropertyName2()); + assertEquals("annotationNamedWriterPropertyValue", testItemWriter.getAnnotationNamedProperty()); + assertNull(testItemWriter.getNotDefinedProperty()); + assertNull(testItemWriter.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testCheckpointAlgorithmProperties() throws Exception { + Assert.assertEquals("algorithmPropertyValue1", testCheckpointAlgorithm.getAlgorithmPropertyName1()); + Assert.assertEquals("algorithmPropertyValue2", testCheckpointAlgorithm.getAlgorithmPropertyName2()); + assertEquals("annotationNamedAlgorithmPropertyValue", testCheckpointAlgorithm.getAnnotationNamedProperty()); + assertNull(testCheckpointAlgorithm.getNotDefinedProperty()); + assertNull(testCheckpointAlgorithm.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testDeciderProperties() throws Exception { + Assert.assertEquals("deciderPropertyValue1", testDecider.getDeciderPropertyName1()); + Assert.assertEquals("deciderPropertyValue2", testDecider.getDeciderPropertyName2()); + assertEquals("annotationNamedDeciderPropertyValue", testDecider.getAnnotationNamedProperty()); + assertNull(testDecider.getNotDefinedProperty()); + assertNull(testDecider.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testStepListenerProperties() throws Exception { + Assert.assertEquals("stepListenerPropertyValue1", testStepListener.getStepListenerPropertyName1()); + Assert.assertEquals("stepListenerPropertyValue2", testStepListener.getStepListenerPropertyName2()); + assertEquals("annotationNamedStepListenerPropertyValue", testStepListener.getAnnotationNamedProperty()); + assertNull(testStepListener.getNotDefinedProperty()); + assertNull(testStepListener.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testBatchletProperties() throws Exception { + Assert.assertEquals("batchletPropertyValue1", testBatchlet.getBatchletPropertyName1()); + Assert.assertEquals("batchletPropertyValue2", testBatchlet.getBatchletPropertyName2()); + assertEquals("annotationNamedBatchletPropertyValue", testBatchlet.getAnnotationNamedProperty()); + assertNull(testBatchlet.getNotDefinedProperty()); + assertNull(testBatchlet.getNotDefinedAnnotationNamedProperty()); + } + + @Test + public void testFieldWithInjectAnnotationOnlyInjects() throws Exception { + assertNotNull(testItemReader.getInjectAnnotatedOnlyField()); + assertEquals("Chris", testItemReader.getInjectAnnotatedOnlyField().getName()); + } + + @Test + public void testFieldWithBatchPropertyAnnotationOnlyNoInjection() throws Exception { + assertNull(testItemReader.getBatchAnnotatedOnlyField()); + } + + public static final class TestItemReader implements ItemReader { + private int cnt; + + @Inject @BatchProperty String readerPropertyName1; + @Inject @BatchProperty String readerPropertyName2; + @Inject @BatchProperty(name = "annotationNamedReaderPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + @Inject @BatchProperty String jobPropertyName1; + @Inject @BatchProperty String jobPropertyName2; + @Inject InjectTestObj injectAnnotatedOnlyField; + @BatchProperty String batchAnnotatedOnlyField; + @Inject javax.batch.runtime.context.StepContext stepContext; + + @Override + public void open(Serializable serializable) throws Exception { + org.springframework.util.Assert.notNull(stepContext); + org.springframework.util.Assert.isNull(stepContext.getProperties().get("step2PropertyName1")); + org.springframework.util.Assert.isNull(stepContext.getProperties().get("step2PropertyName2")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("step1PropertyName1").equals("step1PropertyValue1")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("step1PropertyName2").equals("step1PropertyValue2")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("jobPropertyName1").equals("jobPropertyValue1")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("jobPropertyName2").equals("jobPropertyValue2")); + } + + @Override + public void close() throws Exception { + } + + @Override + public Object readItem() throws Exception { + if (cnt == 0) { + cnt++; + return "blah"; + } + + return null; + } + + @Override + public Serializable checkpointInfo() throws Exception { + return null; + } + + String getReaderPropertyName1() { + return readerPropertyName1; + } + + String getReaderPropertyName2() { + return readerPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + + String getJobPropertyName1() { + return jobPropertyName1; + } + + String getJobPropertyName2() { + return jobPropertyName2; + } + + InjectTestObj getInjectAnnotatedOnlyField() { + return injectAnnotatedOnlyField; + } + + String getBatchAnnotatedOnlyField() { + return batchAnnotatedOnlyField; + } + } + + public static final class TestItemProcessor implements ItemProcessor { + @Inject @BatchProperty String processorPropertyName1; + @Inject @BatchProperty String processorPropertyName2; + @Inject @BatchProperty(name = "annotationNamedProcessorPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + + @Override + public Object processItem(Object o) throws Exception { + return o; + } + + String getProcessorPropertyName1() { + return processorPropertyName1; + } + + String getProcessorPropertyName2() { + return processorPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static final class TestItemWriter implements ItemWriter { + @Inject @BatchProperty String writerPropertyName1; + @Inject @BatchProperty String writerPropertyName2; + @Inject @BatchProperty(name = "annotationNamedWriterPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + + @Override + public void open(Serializable serializable) throws Exception { + } + + @Override + public void close() throws Exception { + } + + @Override + public void writeItems(List objects) throws Exception { + System.out.println(objects); + } + + @Override + public Serializable checkpointInfo() throws Exception { + return null; + } + + String getWriterPropertyName1() { + return writerPropertyName1; + } + + String getWriterPropertyName2() { + return writerPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static final class TestCheckpointAlgorithm implements CompletionPolicy { + @Inject @BatchProperty String algorithmPropertyName1; + @Inject @BatchProperty String algorithmPropertyName2; + @Inject @BatchProperty(name = "annotationNamedAlgorithmPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + + @Override + public boolean isComplete(RepeatContext context, RepeatStatus result) { + return true; + } + + @Override + public boolean isComplete(RepeatContext context) { + return true; + } + + @Override + public RepeatContext start(RepeatContext parent) { + return parent; + } + + @Override + public void update(RepeatContext context) { + } + + String getAlgorithmPropertyName1() { + return algorithmPropertyName1; + } + + String getAlgorithmPropertyName2() { + return algorithmPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static class TestDecider implements JobExecutionDecider { + @Inject @BatchProperty String deciderPropertyName1; + @Inject @BatchProperty String deciderPropertyName2; + @Inject @BatchProperty(name = "annotationNamedDeciderPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + + @Override + public FlowExecutionStatus decide(JobExecution jobExecution, + StepExecution stepExecution) { + return new FlowExecutionStatus("step2"); + } + + String getDeciderPropertyName1() { + return deciderPropertyName1; + } + + String getDeciderPropertyName2() { + return deciderPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static class TestStepListener implements javax.batch.api.chunk.listener.ItemReadListener, + javax.batch.api.chunk.listener.ItemProcessListener, javax.batch.api.chunk.listener.ItemWriteListener { + @Inject @BatchProperty String stepListenerPropertyName1; + @Inject @BatchProperty String stepListenerPropertyName2; + @Inject @BatchProperty(name = "annotationNamedStepListenerPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + + @Override + public void beforeProcess(Object o) throws Exception { + } + + @Override + public void afterProcess(Object o, Object o2) throws Exception { + } + + @Override + public void onProcessError(Object o, Exception e) throws Exception { + } + + @Override + public void beforeRead() throws Exception { + } + + @Override + public void afterRead(Object o) throws Exception { + } + + @Override + public void onReadError(Exception e) throws Exception { + } + + @Override + public void beforeWrite(List objects) throws Exception { + } + + @Override + public void afterWrite(List objects) throws Exception { + } + + @Override + public void onWriteError(List objects, Exception e) throws Exception { + } + + String getStepListenerPropertyName1() { + return stepListenerPropertyName1; + } + + String getStepListenerPropertyName2() { + return stepListenerPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static class TestBatchlet implements Batchlet { + @Inject @BatchProperty String batchletPropertyName1; + @Inject @BatchProperty String batchletPropertyName2; + @Inject @BatchProperty(name = "annotationNamedBatchletPropertyName") String annotationNamedProperty; + @Inject @BatchProperty String notDefinedProperty; + @Inject @BatchProperty(name = "notDefinedAnnotationNamedProperty") String notDefinedAnnotationNamedProperty; + @Inject javax.batch.runtime.context.StepContext stepContext; + + @Override + public String process() throws Exception { + org.springframework.util.Assert.notNull(stepContext); + org.springframework.util.Assert.isNull(stepContext.getProperties().get("step1PropertyName1")); + org.springframework.util.Assert.isNull(stepContext.getProperties().get("step1PropertyName2")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("step2PropertyName1").equals("step2PropertyValue1")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("step2PropertyName2").equals("step2PropertyValue2")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("jobPropertyName1").equals("jobPropertyValue1")); + org.springframework.util.Assert.isTrue(stepContext.getProperties().get("jobPropertyName2").equals("jobPropertyValue2")); + + return null; + } + + @Override + public void stop() throws Exception { + } + + String getBatchletPropertyName1() { + return batchletPropertyName1; + } + + String getBatchletPropertyName2() { + return batchletPropertyName2; + } + + String getAnnotationNamedProperty() { + return annotationNamedProperty; + } + + String getNotDefinedProperty() { + return notDefinedProperty; + } + + String getNotDefinedAnnotationNamedProperty() { + return notDefinedAnnotationNamedProperty; + } + } + + public static class InjectTestObj { + private String name; + + public InjectTestObj(String name) { + this.name = name; + } + + public String getName() { + return name; + } + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java new file mode 100644 index 000000000..bafdf1d77 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProcessorTests.java @@ -0,0 +1,418 @@ +/* + * 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.item; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.JobExecution; +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.step.builder.JsrSimpleStepBuilder; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + +public class JsrChunkProcessorTests { + + private FailingListItemReader reader; + private FailingCountingItemProcessor processor; + private StoringItemWriter writer; + private CountingListener readListener; + private JsrSimpleStepBuilder builder; + private JobRepository repository; + private StepExecution stepExecution; + + @Before + public void setUp() throws Exception { + + List items = new ArrayList(); + + for (int i = 0; i < 25; i++) { + items.add("item " + i); + } + + reader = new FailingListItemReader(items); + processor = new FailingCountingItemProcessor(); + writer = new StoringItemWriter(); + readListener = new CountingListener(); + + builder = new JsrSimpleStepBuilder(new StepBuilder("step1")); + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + builder.repository(repository); + builder.transactionManager(new ResourcelessTransactionManager()); + stepExecution = null; + } + + @Test + public void testNoInputNoListeners() throws Exception{ + reader = new FailingListItemReader(new ArrayList()); + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(0, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + } + + @Test + public void testSimpleScenarioNoListeners() throws Exception{ + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, writer.results.size()); + assertEquals(25, processor.count); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count++; + } + } + + @Test + public void testSimpleScenarioNoProcessor() throws Exception{ + Step step = builder.chunk(25).reader(reader).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(0, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + assertEquals(0, processor.count); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count++; + } + } + + @Test + public void testProcessorFilteringNoListeners() throws Exception{ + processor.filter = true; + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count += 2; + } + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(13, stepExecution.getWriteCount()); + assertEquals(12, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, processor.count); + } + + @Test + public void testReadError() throws Exception{ + reader.failCount = 10; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(9, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(9, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(1, stepExecution.getFailureExceptions().size()); + assertEquals("expected at read index 10", stepExecution.getFailureExceptions().get(0).getMessage()); + assertEquals(9, readListener.afterProcess); + assertEquals(9, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(9, readListener.beforeProcess); + assertEquals(10, readListener.beforeRead); + assertEquals(0, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(1, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testProcessError() throws Exception{ + processor.failCount = 10; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(10, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(10, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals("expected at process index 10", stepExecution.getFailureExceptions().get(0).getMessage()); + assertEquals(9, readListener.afterProcess); + assertEquals(10, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(10, readListener.beforeProcess); + assertEquals(10, readListener.beforeRead); + assertEquals(0, readListener.beforeWriteCount); + assertEquals(1, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testWriteError() throws Exception{ + writer.fail = true; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals("expected in write", stepExecution.getFailureExceptions().get(0).getMessage()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(25, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(1, readListener.onWriteError); + } + + @Test + public void testMultipleChunks() throws Exception{ + + Step step = builder.chunk(10).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(3, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(3, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + protected void runStep(Step step) + throws JobExecutionAlreadyRunningException, JobRestartException, + JobInstanceAlreadyCompleteException, JobInterruptedException { + JobExecution jobExecution = repository.createJobExecution("job1", new JobParameters()); + stepExecution = new StepExecution("step1", jobExecution); + repository.add(stepExecution); + + step.execute(stepExecution); + } + + public static class FailingListItemReader extends ListItemReader { + + protected int failCount = -1; + protected int count = 0; + + public FailingListItemReader(List list) { + super(list); + } + + @Override + public String read() { + count++; + + if(failCount == count) { + throw new RuntimeException("expected at read index " + failCount); + } else { + return super.read(); + } + } + } + + public static class FailingCountingItemProcessor implements ItemProcessor{ + protected int count = 0; + protected int failCount = -1; + protected boolean filter = false; + + @Override + public String process(String item) throws Exception { + count++; + + if(filter && count % 2 == 0) { + return null; + } else if(count == failCount){ + throw new RuntimeException("expected at process index " + failCount); + } else { + return item; + } + } + } + + public static class StoringItemWriter implements ItemWriter{ + + protected List results = new ArrayList(); + protected boolean fail = false; + + @Override + public void write(List items) throws Exception { + if(fail) { + throw new RuntimeException("expected in write"); + } + + results.addAll(items); + } + } + + public static class CountingListener implements ItemReadListener, ItemProcessListener, ItemWriteListener { + + protected int beforeWriteCount = 0; + protected int afterWrite = 0; + protected int onWriteError = 0; + protected int beforeProcess = 0; + protected int afterProcess = 0; + protected int onProcessError = 0; + protected int beforeRead = 0; + protected int afterRead = 0; + protected int onReadError = 0; + + @Override + public void beforeWrite(List items) { + beforeWriteCount++; + } + + @Override + public void afterWrite(List items) { + afterWrite++; + } + + @Override + public void onWriteError(Exception exception, + List items) { + onWriteError++; + } + + @Override + public void beforeProcess(String item) { + beforeProcess++; + } + + @Override + public void afterProcess(String item, String result) { + afterProcess++; + } + + @Override + public void onProcessError(String item, Exception e) { + onProcessError++; + } + + @Override + public void beforeRead() { + beforeRead++; + } + + @Override + public void afterRead(String item) { + afterRead++; + } + + @Override + public void onReadError(Exception ex) { + onReadError++; + } + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProviderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProviderTests.java new file mode 100644 index 000000000..edd8e0547 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrChunkProviderTests.java @@ -0,0 +1,40 @@ +/* + * 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.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.step.item.Chunk; + +public class JsrChunkProviderTests { + + private JsrChunkProvider provider; + + @Before + public void setUp() throws Exception { + provider = new JsrChunkProvider(); + } + + @Test + public void test() throws Exception { + Chunk chunk = provider.provide(null); + assertNotNull(chunk); + assertEquals(0, chunk.getItems().size()); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java new file mode 100644 index 000000000..62b10cc83 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/item/JsrFaultTolerantChunkProcessorTests.java @@ -0,0 +1,601 @@ +/* + * 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.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.JobExecution; +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.step.builder.JsrFaultTolerantStepBuilder; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; + +public class JsrFaultTolerantChunkProcessorTests { + + private FailingListItemReader reader; + private FailingCountingItemProcessor processor; + private StoringItemWriter writer; + private CountingListener readListener; + private JsrFaultTolerantStepBuilder builder; + private JobRepository repository; + private StepExecution stepExecution; + + @Before + public void setUp() throws Exception { + + List items = new ArrayList(); + + for (int i = 0; i < 25; i++) { + items.add("item " + i); + } + + reader = new FailingListItemReader(items); + processor = new FailingCountingItemProcessor(); + writer = new StoringItemWriter(); + readListener = new CountingListener(); + + builder = new JsrFaultTolerantStepBuilder(new StepBuilder("step1")); + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + builder.repository(repository); + builder.transactionManager(new ResourcelessTransactionManager()); + stepExecution = null; + } + + @Test + public void testNoInputNoListeners() throws Exception{ + reader = new FailingListItemReader(new ArrayList()); + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(0, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + } + + @Test + public void testSimpleScenarioNoListeners() throws Exception{ + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, writer.results.size()); + assertEquals(25, processor.count); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count++; + } + } + + @Test + public void testSimpleScenarioNoProcessor() throws Exception{ + Step step = builder.chunk(25).reader(reader).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(0, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + assertEquals(0, processor.count); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count++; + } + } + + @Test + public void testProcessorFilteringNoListeners() throws Exception{ + processor.filter = true; + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + int count = 0; + for (String curItem : writer.results) { + assertEquals("item " + count, curItem); + count += 2; + } + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(13, stepExecution.getWriteCount()); + assertEquals(12, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, processor.count); + } + + @Test + public void testSkipReadError() throws Exception{ + reader.failCount = 10; + + Step step = builder.faultTolerant().skip(RuntimeException.class).skipLimit(20).chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertNotNull(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(1, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, stepExecution.getFailureExceptions().size()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(27, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(1, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testRetryReadError() throws Exception{ + reader.failCount = 10; + + Step step = builder.faultTolerant().retry(RuntimeException.class).retryLimit(20).chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, stepExecution.getFailureExceptions().size()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(27, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(1, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testReadError() throws Exception{ + reader.failCount = 10; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertNotNull(stepExecution); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(9, processor.count); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(9, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(1, stepExecution.getFailureExceptions().size()); + assertEquals("expected at read index 10", stepExecution.getFailureExceptions().get(0).getCause().getMessage()); + assertEquals(9, readListener.afterProcess); + assertEquals(9, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(9, readListener.beforeProcess); + assertEquals(10, readListener.beforeRead); + assertEquals(0, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(1, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testProcessError() throws Exception{ + processor.failCount = 10; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(10, processor.count); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(10, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals("expected at process index 10", stepExecution.getFailureExceptions().get(0).getCause().getMessage()); + assertEquals(9, readListener.afterProcess); + assertEquals(10, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(10, readListener.beforeProcess); + assertEquals(10, readListener.beforeRead); + assertEquals(0, readListener.beforeWriteCount); + assertEquals(1, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testSkipProcessError() throws Exception{ + processor.failCount = 10; + + Step step = builder.faultTolerant().skip(RuntimeException.class).skipLimit(20).chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertNotNull(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(24, writer.results.size()); + assertEquals(1, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(24, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, stepExecution.getFailureExceptions().size()); + assertEquals(24, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(1, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testRetryProcessError() throws Exception{ + processor.failCount = 10; + + Step step = builder.faultTolerant().retry(RuntimeException.class).retryLimit(20).chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertNotNull(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(26, processor.count); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, stepExecution.getFailureExceptions().size()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(26, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(1, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + @Test + public void testWriteError() throws Exception{ + writer.fail = true; + + Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(25, processor.count); + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(0, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(0, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(25, readListener.beforeRead); + assertEquals(1, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(1, readListener.onWriteError); + } + + @Test + public void testRetryWriteError() throws Exception{ + writer.fail = true; + + Step step = builder.faultTolerant().retry(RuntimeException.class).retryLimit(25).chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(25, processor.count); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(1, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(2, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(1, readListener.onWriteError); + } + + @Test + public void testSkipWriteError() throws Exception{ + writer.fail = true; + + Step step = builder.faultTolerant().skip(RuntimeException.class).skipLimit(25).chunk(7).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, processor.count); + assertEquals(18, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(18, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(3, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(4, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(1, readListener.onWriteError); + } + + @Test + public void testMultipleChunks() throws Exception{ + + Step step = builder.chunk(10).reader(reader).processor(processor).writer(writer).listener((ItemReadListener) readListener).build(); + + runStep(step); + + assertEquals(25, processor.count); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(25, writer.results.size()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(25, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getSkipCount()); + assertEquals(25, stepExecution.getWriteCount()); + assertEquals(0, stepExecution.getFilterCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(25, readListener.afterProcess); + assertEquals(25, readListener.afterRead); + assertEquals(3, readListener.afterWrite); + assertEquals(25, readListener.beforeProcess); + assertEquals(26, readListener.beforeRead); + assertEquals(3, readListener.beforeWriteCount); + assertEquals(0, readListener.onProcessError); + assertEquals(0, readListener.onReadError); + assertEquals(0, readListener.onWriteError); + } + + protected void runStep(Step step) + throws JobExecutionAlreadyRunningException, JobRestartException, + JobInstanceAlreadyCompleteException, JobInterruptedException { + JobExecution jobExecution = repository.createJobExecution("job1", new JobParameters()); + stepExecution = new StepExecution("step1", jobExecution); + repository.add(stepExecution); + + step.execute(stepExecution); + } + + public static class FailingListItemReader extends ListItemReader { + + protected int failCount = -1; + protected int count = 0; + + public FailingListItemReader(List list) { + super(list); + } + + @Override + public String read() { + count++; + + if(failCount == count) { + throw new RuntimeException("expected at read index " + failCount); + } else { + return super.read(); + } + } + } + + public static class FailingCountingItemProcessor implements ItemProcessor{ + protected int count = 0; + protected int failCount = -1; + protected boolean filter = false; + + @Override + public String process(String item) throws Exception { + count++; + + if(filter && count % 2 == 0) { + return null; + } else if(count == failCount){ + throw new RuntimeException("expected at process index " + failCount); + } else { + return item; + } + } + } + + public static class StoringItemWriter implements ItemWriter{ + + protected List results = new ArrayList(); + protected boolean fail = false; + + @Override + public void write(List items) throws Exception { + if(fail) { + fail = false; + throw new RuntimeException("expected in write"); + } + + results.addAll(items); + } + } + + public static class CountingListener implements ItemReadListener, ItemProcessListener, ItemWriteListener { + + protected int beforeWriteCount = 0; + protected int afterWrite = 0; + protected int onWriteError = 0; + protected int beforeProcess = 0; + protected int afterProcess = 0; + protected int onProcessError = 0; + protected int beforeRead = 0; + protected int afterRead = 0; + protected int onReadError = 0; + + @Override + public void beforeWrite(List items) { + beforeWriteCount++; + } + + @Override + public void afterWrite(List items) { + afterWrite++; + } + + @Override + public void onWriteError(Exception exception, + List items) { + onWriteError++; + } + + @Override + public void beforeProcess(String item) { + beforeProcess++; + } + + @Override + public void afterProcess(String item, String result) { + afterProcess++; + } + + @Override + public void onProcessError(String item, Exception e) { + onProcessError++; + } + + @Override + public void beforeRead() { + beforeRead++; + } + + @Override + public void afterRead(String item) { + afterRead++; + } + + @Override + public void onReadError(Exception ex) { + onReadError++; + } + } +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/repeat/CheckpointAlgorithmAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/repeat/CheckpointAlgorithmAdapter.java index 4416ef7df..f476dac46 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/repeat/CheckpointAlgorithmAdapter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/repeat/CheckpointAlgorithmAdapter.java @@ -1,12 +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.jsr.repeat; import javax.batch.api.chunk.CheckpointAlgorithm; +import javax.batch.operations.BatchRuntimeException; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.util.Assert; +/** + * Wrapper for the {@link CheckpointAlgorithm} to be used via the rest + * of the framework. + * + * @author Michael Minella + * @see CheckpointAlgorithm + * @see CompletionPolicy + */ public class CheckpointAlgorithmAdapter implements CompletionPolicy { private CheckpointAlgorithm policy; @@ -17,39 +41,50 @@ public class CheckpointAlgorithmAdapter implements CompletionPolicy { this.policy = policy; } + /* (non-Javadoc) + * @see org.springframework.batch.repeat.CompletionPolicy#isComplete(org.springframework.batch.repeat.RepeatContext, org.springframework.batch.repeat.RepeatStatus) + */ @Override public boolean isComplete(RepeatContext context, RepeatStatus result) { try { return policy.isReadyToCheckpoint(); } catch (Exception e) { - //TODO: do something here + throw new BatchRuntimeException(e); } - - return false; } + /* (non-Javadoc) + * @see org.springframework.batch.repeat.CompletionPolicy#isComplete(org.springframework.batch.repeat.RepeatContext) + */ @Override public boolean isComplete(RepeatContext context) { try { return policy.isReadyToCheckpoint(); } catch (Exception e) { - //TODO: do something here + throw new BatchRuntimeException(e); } - - return false; } + /* (non-Javadoc) + * @see org.springframework.batch.repeat.CompletionPolicy#start(org.springframework.batch.repeat.RepeatContext) + */ @Override public RepeatContext start(RepeatContext parent) { try { policy.beginCheckpoint(); } catch (Exception e) { - //TODO: do something here + throw new BatchRuntimeException(e); } - return null; + return parent; } + /** + * If {@link CheckpointAlgorithm#isReadyToCheckpoint()} is true + * we will call {@link CheckpointAlgorithm#endCheckpoint()} + * + * @param context a {@link RepeatContext} + */ @Override public void update(RepeatContext context) { try { @@ -57,7 +92,7 @@ public class CheckpointAlgorithmAdapter implements CompletionPolicy { policy.endCheckpoint(); } } catch (Exception e) { - //TODO: do something here + throw new BatchRuntimeException(e); } } }