BATCH-2062: Added ChunkProcessor implementations to support JSR-352's chunking pattern
* Added the JsrChunkProvider implementation - a no-op ChunkProvider implementation to be used by the ChunkOrientedTasklet * Added the JsrChunkProcessor - A simple implementation of a ChunkProcessor that handles the JSR-352 chunking pattern (read and process loop with a single write per chunk). * Added the JsrFaultTolerantChunkProcessor - A ChunkProcessor that implements chunking the way JSR-352 requires as well as adding skip/retry functionality. * Added builders to support the above functionality.
This commit is contained in:
Michael Minella
@@ -309,7 +309,7 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
|
||||
|
||||
private Step createFaultTolerantStep() {
|
||||
|
||||
FaultTolerantStepBuilder<I, O> builder = new FaultTolerantStepBuilder<I, O>(new StepBuilder(name));
|
||||
FaultTolerantStepBuilder<I, O> builder = getFaultTolerantStepBuilder(this.name);
|
||||
|
||||
if (commitInterval != null) {
|
||||
builder.chunk(commitInterval);
|
||||
@@ -382,6 +382,10 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
|
||||
|
||||
}
|
||||
|
||||
protected FaultTolerantStepBuilder<I, O> getFaultTolerantStepBuilder(String stepName) {
|
||||
return new FaultTolerantStepBuilder<I, O>(new StepBuilder(stepName));
|
||||
}
|
||||
|
||||
private void registerItemListeners(SimpleStepBuilder<I, O> builder) {
|
||||
for (ItemReadListener<I> listener : readListeners) {
|
||||
builder.listener(listener);
|
||||
@@ -396,7 +400,7 @@ public class StepParserStepFactoryBean<I, O> 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<I, O> 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);
|
||||
|
||||
@@ -34,9 +34,9 @@ public class StepContextFactoryBean implements FactoryBean<StepContext> {
|
||||
@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<StepContext> {
|
||||
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;
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 <I> The input type for the step
|
||||
* @param <O> The output type for the step
|
||||
*/
|
||||
public class JsrFaultTolerantStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
|
||||
|
||||
public JsrFaultTolerantStepBuilder(StepBuilder parent) {
|
||||
super(parent);
|
||||
}
|
||||
|
||||
@Override
|
||||
public FaultTolerantStepBuilder<I, O> faultTolerant() {
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ChunkProvider<I> createChunkProvider() {
|
||||
return new JsrChunkProvider<I>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<I> 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<StepListener> listeners = new ArrayList<StepListener>(getItemListeners());
|
||||
chunkProcessor.setListeners(listeners);
|
||||
|
||||
return chunkProcessor;
|
||||
}
|
||||
}
|
||||
@@ -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 <I> The input type for the step
|
||||
* @param <O> The output type for the step
|
||||
*/
|
||||
public class JsrSimpleStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
|
||||
|
||||
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<I> chunkProvider = new JsrChunkProvider<I>();
|
||||
JsrChunkProcessor<I, O> chunkProcessor = new JsrChunkProcessor(getReader(), getProcessor(), getWriter(), repeatOperations);
|
||||
chunkProcessor.setListeners(new ArrayList<StepListener>(getItemListeners()));
|
||||
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
|
||||
tasklet.setBuffering(!isReaderTransactionalQueue());
|
||||
return tasklet;
|
||||
}
|
||||
|
||||
private RepeatOperations createRepeatOperations() {
|
||||
RepeatTemplate repeatOperations = new RepeatTemplate();
|
||||
repeatOperations.setCompletionPolicy(getChunkCompletionPolicy());
|
||||
repeatOperations.setExceptionHandler(getExceptionHandler());
|
||||
return repeatOperations;
|
||||
}
|
||||
}
|
||||
@@ -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 <I> The input type for the step
|
||||
* @param <O> The output type for the step
|
||||
*/
|
||||
public class JsrChunkProcessor<I,O> implements ChunkProcessor<I> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
private ItemReader<? extends I> itemReader;
|
||||
@SuppressWarnings("rawtypes")
|
||||
private final MulticasterBatchListener listener = new MulticasterBatchListener();
|
||||
private RepeatOperations repeatTemplate;
|
||||
private ItemProcessor<? super I, ? extends O> itemProcessor;
|
||||
private ItemWriter<? super O> itemWriter;
|
||||
|
||||
public JsrChunkProcessor() {
|
||||
this(null, null, null, null);
|
||||
}
|
||||
|
||||
public JsrChunkProcessor(ItemReader<I> reader, ItemProcessor<I,O> processor, ItemWriter<O> 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<I> chunk)
|
||||
throws Exception {
|
||||
|
||||
final AtomicInteger filterCount = new AtomicInteger(0);
|
||||
final Chunk<O> output = new Chunk<O>();
|
||||
|
||||
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<? extends StepListener> 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<I> 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<I> 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<O> 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<O> chunk) throws Exception {
|
||||
try {
|
||||
List<O> items = chunk.getItems();
|
||||
listener.beforeWrite(items);
|
||||
itemWriter.write(items);
|
||||
listener.afterWrite(items);
|
||||
}
|
||||
catch (Exception e) {
|
||||
listener.onWriteError(e, chunk.getItems());
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 <T> The type of input for the step
|
||||
*/
|
||||
public class JsrChunkProvider<T> implements ChunkProvider<T> {
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.batch.core.step.item.ChunkProvider#provide(org.springframework.batch.core.StepContribution)
|
||||
*/
|
||||
@Override
|
||||
public Chunk<T> provide(StepContribution contribution) throws Exception {
|
||||
return new Chunk<T>();
|
||||
}
|
||||
|
||||
/* (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<T> chunk) {
|
||||
}
|
||||
}
|
||||
@@ -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 <I> input type for the step
|
||||
* @param <O> output type for the step
|
||||
*/
|
||||
public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O> {
|
||||
|
||||
protected final Log logger = LogFactory.getLog(getClass());
|
||||
private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy();
|
||||
private Classifier<Throwable, Boolean> 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<I> reader, ItemProcessor<I,O> processor, ItemWriter<O> 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<Throwable, Boolean> 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<? extends StepListener> 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<I> chunk) throws Exception {
|
||||
RetryCallback<I> retryCallback = new RetryCallback<I>() {
|
||||
|
||||
@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<I> recoveryCallback = new RecoveryCallback<I>() {
|
||||
|
||||
@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<O> retryCallback = new RetryCallback<O>() {
|
||||
|
||||
@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<O> recoveryCallback = new RecoveryCallback<O>() {
|
||||
|
||||
@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<O> chunk) throws Exception {
|
||||
|
||||
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
|
||||
@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<Object> recoveryCallback = new RecoveryCallback<Object>() {
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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<I, O> extends SimpleStepBuilder<I, O> {
|
||||
Assert.state(getProcessor() != null || getWriter() != null, "ItemWriter or ItemProcessor must be provided");
|
||||
addSpecialExceptions();
|
||||
registerSkipListeners();
|
||||
FaultTolerantChunkProvider<I> chunkProvider = createChunkProvider();
|
||||
FaultTolerantChunkProcessor<I, O> chunkProcessor = createChunkProcessor();
|
||||
ChunkProvider<I> chunkProvider = createChunkProvider();
|
||||
ChunkProcessor<I> chunkProcessor = createChunkProcessor();
|
||||
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<I>(chunkProvider, chunkProcessor);
|
||||
tasklet.setBuffering(!isReaderTransactionalQueue());
|
||||
return tasklet;
|
||||
@@ -382,7 +384,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
|
||||
return this;
|
||||
}
|
||||
|
||||
private FaultTolerantChunkProvider<I> createChunkProvider() {
|
||||
protected ChunkProvider<I> createChunkProvider() {
|
||||
|
||||
SkipPolicy readSkipPolicy = createSkipPolicy();
|
||||
readSkipPolicy = getFatalExceptionAwareProxy(readSkipPolicy);
|
||||
@@ -399,7 +401,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
|
||||
|
||||
}
|
||||
|
||||
private FaultTolerantChunkProcessor<I, O> createChunkProcessor() {
|
||||
protected ChunkProcessor<I> createChunkProcessor() {
|
||||
|
||||
BatchRetryTemplate batchRetryTemplate = createRetryOperations();
|
||||
|
||||
@@ -435,7 +437,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
|
||||
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<I, O> extends SimpleStepBuilder<I, O> {
|
||||
*
|
||||
* @return an exception classifier: maps to true if an exception should cause rollback
|
||||
*/
|
||||
private Classifier<Throwable, Boolean> getRollbackClassifier() {
|
||||
protected Classifier<Throwable, Boolean> getRollbackClassifier() {
|
||||
|
||||
Classifier<Throwable, Boolean> classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false);
|
||||
|
||||
@@ -534,7 +536,7 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
|
||||
/**
|
||||
* @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<I, O> extends SimpleStepBuilder<I, O> {
|
||||
|
||||
}
|
||||
|
||||
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<I, O> extends SimpleStepBuilder<I, O> {
|
||||
* @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<Class<? extends Throwable>, SkipPolicy> map = new HashMap<Class<? extends Throwable>, SkipPolicy>();
|
||||
|
||||
@@ -311,7 +311,7 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
|
||||
/**
|
||||
* @return a {@link CompletionPolicy} consistent with the chunk size and injected policy (if present).
|
||||
*/
|
||||
private CompletionPolicy getChunkCompletionPolicy() {
|
||||
protected CompletionPolicy getChunkCompletionPolicy() {
|
||||
Assert.state(!(completionPolicy != null && chunkSize > 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).");
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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<String, String> builder;
|
||||
private JobRepository repository;
|
||||
private StepExecution stepExecution;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
|
||||
List<String> items = new ArrayList<String>();
|
||||
|
||||
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<String, String>(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<String>());
|
||||
Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String> {
|
||||
|
||||
protected int failCount = -1;
|
||||
protected int count = 0;
|
||||
|
||||
public FailingListItemReader(List<String> 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<String, String>{
|
||||
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<String>{
|
||||
|
||||
protected List<String> results = new ArrayList<String>();
|
||||
protected boolean fail = false;
|
||||
|
||||
@Override
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if(fail) {
|
||||
throw new RuntimeException("expected in write");
|
||||
}
|
||||
|
||||
results.addAll(items);
|
||||
}
|
||||
}
|
||||
|
||||
public static class CountingListener implements ItemReadListener<String>, ItemProcessListener<String, String>, ItemWriteListener<String> {
|
||||
|
||||
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<? extends String> items) {
|
||||
beforeWriteCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> items) {
|
||||
afterWrite++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onWriteError(Exception exception,
|
||||
List<? extends String> 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++;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String> provider;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
provider = new JsrChunkProvider<String>();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void test() throws Exception {
|
||||
Chunk<String> chunk = provider.provide(null);
|
||||
assertNotNull(chunk);
|
||||
assertEquals(0, chunk.getItems().size());
|
||||
}
|
||||
}
|
||||
@@ -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<String, String> builder;
|
||||
private JobRepository repository;
|
||||
private StepExecution stepExecution;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
|
||||
List<String> items = new ArrayList<String>();
|
||||
|
||||
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<String, String>(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<String>());
|
||||
Step step = builder.chunk(25).reader(reader).processor(processor).writer(writer).listener((ItemReadListener<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String>) 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<String> {
|
||||
|
||||
protected int failCount = -1;
|
||||
protected int count = 0;
|
||||
|
||||
public FailingListItemReader(List<String> 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<String, String>{
|
||||
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<String>{
|
||||
|
||||
protected List<String> results = new ArrayList<String>();
|
||||
protected boolean fail = false;
|
||||
|
||||
@Override
|
||||
public void write(List<? extends String> items) throws Exception {
|
||||
if(fail) {
|
||||
fail = false;
|
||||
throw new RuntimeException("expected in write");
|
||||
}
|
||||
|
||||
results.addAll(items);
|
||||
}
|
||||
}
|
||||
|
||||
public static class CountingListener implements ItemReadListener<String>, ItemProcessListener<String, String>, ItemWriteListener<String> {
|
||||
|
||||
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<? extends String> items) {
|
||||
beforeWriteCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterWrite(List<? extends String> items) {
|
||||
afterWrite++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onWriteError(Exception exception,
|
||||
List<? extends String> 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++;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user