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

OPEN - issue BATCH-789: Remove mark/reset from ItemReader

Browse Source 
Still some open issues with retry and skip, but nearly there...
This commit is contained in:
dsyer
2008-08-21 09:58:04 +00:00
parent 24168016af
commit 04db13bbc4
9 changed files with 323 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
View File

@@ -16,8 +16,8 @@
package org.springframework.batch.core;
/**
* Represents a contribution to a {@link StepExecution}, buffering changes
* until they can be applied at a chunk boundary.
* Represents a contribution to a {@link StepExecution}, buffering changes until
* they can be applied at a chunk boundary.
*
* @author Dave Syer
*
@@ -34,8 +34,6 @@ public class StepContribution {
private volatile int writeSkipCount;
private volatile int uncommitedReadSkipCount;
/**
* @param execution
*/
@@ -76,17 +74,16 @@ public class StepContribution {
/**
* @return the sum of skips accumulated in the parent {@link StepExecution}
* and this <code>StepContribution</code>, including uncommitted read
* skips.
* and this <code>StepContribution</code>.
*/
public int getStepSkipCount() {
return uncommitedReadSkipCount + readSkipCount + writeSkipCount + parentSkipCount;
return readSkipCount + writeSkipCount + parentSkipCount;
}
/**
* @return the number of skips collected in this
* <code>StepContribution</code> (not including skips accumulated in the
* parent {@link StepExecution}.
* parent {@link StepExecution}).
*/
public int getSkipCount() {
return readSkipCount + writeSkipCount;
@@ -95,10 +92,17 @@ public class StepContribution {
/**
* Increment the read skip count for this contribution
*/
public void incrementReadsSkipCount() {
public void incrementReadSkipCount() {
readSkipCount++;
}
/**
* Increment the read skip count for this contribution
*/
public void incrementReadSkipCount(int count) {
readSkipCount += count;
}
/**
* Increment the write skip count for this contribution
*/
@@ -106,13 +110,6 @@ public class StepContribution {
writeSkipCount++;
}
/**
* Increment the counter for temporary skipped reads
*/
public void incrementTemporaryReadSkipCount() {
uncommitedReadSkipCount++;
}
/**
* @return the read skip count
*/
@@ -127,22 +124,14 @@ public class StepContribution {
return writeSkipCount;
}
/**
* Add the temporary read skip count to read skip count and reset the
* temporary counter.
*/
public void combineSkipCounts() {
readSkipCount += uncommitedReadSkipCount;
uncommitedReadSkipCount = 0;
}
/*
* (non-Javadoc)
*
* @see java.lang.Object#toString()
*/
public String toString() {
return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", uncommitedReadSkips="
+ uncommitedReadSkipCount + ", readSkips=" + readSkipCount + "writeSkips=" + writeSkipCount + "]";
return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", readSkips=" + readSkipCount
+ ", writeSkips=" + writeSkipCount + "]";
}
}

18
spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java
View File

@@ -19,8 +19,6 @@ package org.springframework.batch.core;
import java.util.Date;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.util.Assert;
@@ -56,7 +54,7 @@ public class StepExecution extends Entity {
private volatile Date startTime = new Date(System.currentTimeMillis());
private volatile Date endTime = null;
private volatile Date lastUpdated = null;
private volatile ExecutionContext executionContext = new ExecutionContext();
@@ -311,8 +309,6 @@ public class StepExecution extends Entity {
public synchronized void apply(StepContribution contribution) {
itemCount += contribution.getItemCount();
commitCount += contribution.getCommitCount();
contribution.combineSkipCounts();
readSkipCount += contribution.getReadSkipCount();
writeSkipCount += contribution.getWriteSkipCount();
}
@@ -347,7 +343,7 @@ public class StepExecution extends Entity {
}
/**
* Increment the number of items skipped in the {@link ItemReader}
* Increment the number of items skipped on read
*
* @param count - the number of skips to increment by.
*/
@@ -356,7 +352,7 @@ public class StepExecution extends Entity {
}
/**
* Increment the number of items skipped in the {@link ItemWriter}
* Increment the number of items skipped on write
*
* @param count - the number of skips to increment by.
*/
@@ -378,21 +374,21 @@ public class StepExecution extends Entity {
}
/**
* @return the number of records skipped in the {@link ItemReader}
* @return the number of records skipped on read
*/
public int getReadSkipCount() {
return readSkipCount;
}
/**
* @return the number of records skipped in the {@link ItemWriter}
* @return the number of records skipped on write
*/
public int getWriteSkipCount() {
return writeSkipCount;
}
/**
* Set the number of records skipped in the {@link ItemReader}
* Set the number of records skipped on read
*
* @param readSkipCount
*/
@@ -401,7 +397,7 @@ public class StepExecution extends Entity {
}
/**
* Set the number of records skipped in the {@link ItemWriter}
* Set the number of records skipped on write
*
* @param writeSkipCount
*/

133
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java
View File

@@ -15,7 +15,10 @@
*/
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -29,7 +32,7 @@ import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* Simplest possible implementation of {@link StepHandler} with no skipping or
@@ -81,31 +84,74 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
*/
public ExitStatus handle(final StepContribution contribution) throws Exception {
ExitStatus result = repeatOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
boolean processed = false;
while (!processed) {
T item = read(contribution);
final List<ReadWrapper<T>> buffer = getItemBuffer();
ExitStatus result = ExitStatus.CONTINUABLE;
if (buffer.isEmpty()) {
result = repeatOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
ReadWrapper<T> item = read(contribution);
if (item == null) {
return ExitStatus.FINISHED;
}
// TODO: segregate read / write / filter count
contribution.incrementItemCount();
processed = write(item, contribution);
contribution.incrementReadSkipCount(item.getSkipCount());
buffer.add(item);
return ExitStatus.CONTINUABLE;
}
return ExitStatus.CONTINUABLE;
}
});
});
}
List<S> processed = new ArrayList<S>();
for (Iterator<ReadWrapper<T>> iterator = buffer.iterator(); iterator.hasNext();) {
ReadWrapper<T> item = iterator.next();
S output = null;
// TODO: segregate read / write / filter count
// (this is read count)
contribution.incrementItemCount();
// TODO: processor listener
output = itemProcessor.process(item.getItem());
// TODO: increment filter count if this is null
if (output != null) {
processed.add(output);
}
}
// TODO: use ItemWriter interface properly
// TODO: make sure exceptions get handled by the appropriate handler
for (S data : processed) {
write(data, contribution);
}
buffer.clear();
logger.info("Contribution: " + contribution);
return result;
}
private List<ReadWrapper<T>> getItemBuffer() {
if (!TransactionSynchronizationManager.hasResource(this)) {
TransactionSynchronizationManager.bindResource(this, new ArrayList<ReadWrapper<T>>());
}
@SuppressWarnings("unchecked")
List<ReadWrapper<T>> resource = (List<ReadWrapper<T>>) TransactionSynchronizationManager.getResource(this);
return resource;
}
/**
* @param contribution current context
* @return next item for writing
*/
protected T read(StepContribution contribution) throws Exception {
return doRead();
protected ReadWrapper<T> read(StepContribution contribution) throws Exception {
T item = doRead();
return item==null ? null : new ReadWrapper<T>(item);
}
/**
@@ -120,24 +166,18 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
*
* @param item the item to write
* @param contribution current context
* @return true if the item was written (as opposed to filtered)
*/
protected boolean write(T item, StepContribution contribution) throws Exception {
return doWrite(item);
protected void write(S item, StepContribution contribution) throws Exception {
doWrite(item);
}
/**
* @param item
* @throws Exception
*/
protected final boolean doWrite(T item) throws Exception {
S processed = itemProcessor.process(item);
if (processed != null) {
// TODO: increment filtered item count
itemWriter.write(Collections.singletonList(processed));
return true;
}
return false;
protected final void doWrite(S item) throws Exception {
// TODO: increment write count
itemWriter.write(Collections.singletonList(item));
}
/**
@@ -154,4 +194,47 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
itemReader.reset();
}
/**
* @author Dave Syer
*
*/
static protected class ReadWrapper<T> {
final private T item;
final private int skipCount;
/**
* @param item
*/
public ReadWrapper(T item) {
this(item, 0);
}
/**
* @param item
* @param skipCount
*/
public ReadWrapper(T item, int skipCount) {
this.item = item;
this.skipCount = skipCount;
}
/**
* @return the item we are wrapping
*/
public T getItem() {
return item;
}
/**
* Public getter for the skipCount.
* @return the skipCount
*/
public int getSkipCount() {
return skipCount;
}
}
}

23
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
View File

@@ -368,22 +368,25 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
* count
* @return next item for processing
*/
protected T read(StepContribution contribution) throws Exception {
protected ReadWrapper<T> read(StepContribution contribution) throws Exception {
int skipCount = 0;
while (true) {
try {
return doRead();
T item = doRead();
return item==null ? null : new ReadWrapper<T>(item, skipCount);
}
catch (Exception e) {
try {
if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementTemporaryReadSkipCount();
try {
skipCount++;
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
contribution.combineSkipCounts();
contribution.incrementReadSkipCount(skipCount);
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
logger.debug("Skipping failed input", e);
@@ -397,7 +400,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
catch (SkipLimitExceededException ex) {
// we are headed for a abnormal ending so bake in the
// skip count
contribution.combineSkipCounts();
contribution.incrementReadSkipCount(skipCount);
throw ex;
}
}
@@ -413,10 +416,12 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
* exhausted. The listener callback (on write failure) will happen in
* the next transaction automatically.<br/>
*/
protected boolean write(final T item, final StepContribution contribution) throws Exception {
@Override
protected void write(final S item, final StepContribution contribution) throws Exception {
RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
public Object doWithRetry(RetryContext context) throws Throwable {
return doWrite(item);
doWrite(item);
return null;
}
}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item);
retryCallback.setRecoveryCallback(new RecoveryCallback() {
@@ -431,10 +436,10 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
}
}
return true;
return null;
}
});
return (Boolean) retryOperations.execute(retryCallback);
retryOperations.execute(retryCallback);
}
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java
View File

@@ -71,8 +71,8 @@ public class ItemOrientedStepHandlerTests {
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
assertEquals(4, itemReader.count);
assertEquals("1234", itemWriter.values);
assertEquals(2, itemReader.count);
assertEquals("12", itemWriter.values);
}
/**

43
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -46,13 +46,9 @@ import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
/**
@@ -131,22 +127,6 @@ public class SimpleStepFactoryBeanTests {
@Test
public void testSimpleJobWithItemListeners() throws Exception {
final List<Throwable> throwables = new ArrayList<Throwable>();
RepeatTemplate chunkOperations = new RepeatTemplate();
// Always handle the exception to check it is the right one...
chunkOperations.setExceptionHandler(new ExceptionHandler() {
public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException {
throwables.add(throwable);
assertEquals("Error!", throwable.getMessage());
}
});
/*
* Each message fails once and the chunk (size=1) "rolls back"; then it
* is recovered ("skipped") on the second attempt (see retry policy
* definition above)...
*/
SimpleStepFactoryBean<String,String> factory = getStepFactory(new String[] { "foo", "bar", "spam" });
factory.setItemWriter(new ItemWriter<String>() {
@@ -164,19 +144,24 @@ public class SimpleStepFactoryBeanTests {
}
} });
factory.setChunkOperations(chunkOperations);
StepHandlerStep step = (StepHandlerStep) factory.getObject();
Step step = (Step) factory.getObject();
job.setSteps(Collections.singletonList((Step) step));
job.setSteps(Collections.singletonList(step));
JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
job.execute(jobExecution);
try {
job.execute(jobExecution);
fail("Expected RuntimeException");
} catch (RuntimeException e) {
// expected
assertEquals("Error!", e.getMessage());
}
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
assertEquals(0, written.size());
// provider should be exhausted
assertEquals(null, reader.read());
assertEquals(3, recovered.size());
// provider should be at second item
assertEquals("bar", reader.read());
assertEquals(1, recovered.size());
}
@Test
@@ -313,7 +298,7 @@ public class SimpleStepFactoryBeanTests {
private SimpleStepFactoryBean<String,String> getStepFactory(String... args) throws Exception {
SimpleStepFactoryBean<String,String> factory = new SimpleStepFactoryBean<String,String>();
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
List<String> items = new ArrayList<String>();
items.addAll(Arrays.asList(args));
reader = new ListItemReader<String>(items);

105
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
View File

@@ -1,27 +1,30 @@
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import junit.framework.TestCase;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.MarkFailedException;
@@ -38,7 +41,7 @@ import org.springframework.util.StringUtils;
/**
* Tests for {@link SkipLimitStepFactoryBean}.
*/
public class SkipLimitStepFactoryBeanTests extends TestCase {
public class SkipLimitStepFactoryBeanTests {
protected final Log logger = LogFactory.getLog(getClass());
@@ -58,7 +61,8 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
protected int count;
protected void setUp() throws Exception {
@Before
public void setUp() throws Exception {
factory.setBeanName("stepName");
factory.setJobRepository(new JobRepositorySupport());
factory.setTransactionManager(new ResourcelessTransactionManager());
@@ -75,8 +79,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@Test
public void testSkip() throws Exception {
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
@@ -103,13 +109,14 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
* Check skippable write exception does not cause rollback when included on
* transaction attributes as "no rollback for".
*/
@Test
public void testSkipWithoutRethrow() throws Exception {
factory.setTransactionAttribute(new DefaultTransactionAttribute() {
public boolean rollbackOn(Throwable ex) {
return !(ex instanceof SkippableRuntimeException);
};
});
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
@@ -129,6 +136,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
* Fatal exception should cause immediate termination regardless of other
* skip settings (note the fatal exception is also classified as skippable).
*/
@Test
public void testFatalException() throws Exception {
factory.setFatalExceptionClasses(new Class[] { FatalRuntimeException.class });
factory.setItemWriter(new SkipWriterStub() {
@@ -137,7 +145,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
}
});
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
try {
@@ -152,11 +160,12 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@Test
public void testSkipOverLimit() throws Exception {
factory.setSkipLimit(1);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -182,17 +191,17 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
@Test
public void testSkipOverLimitOnRead() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Arrays
.asList(StringUtils.commaDelimitedListToStringArray("2,3,5")));
factory.setSkipLimit(3);
factory.setItemReader(reader);
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -224,11 +233,11 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
@Test
public void testSkipListenerFailsOnRead() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Arrays
.asList(StringUtils.commaDelimitedListToStringArray("2,3,5")));
factory.setSkipLimit(3);
factory.setItemReader(reader);
@@ -240,7 +249,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
} });
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -261,11 +270,11 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
@Test
public void testSkipListenerFailsOnWrite() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Arrays
.asList(StringUtils.commaDelimitedListToStringArray("2,3,5")));
factory.setSkipLimit(3);
factory.setItemReader(reader);
@@ -277,7 +286,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
} });
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -289,8 +298,8 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
assertEquals("oops", e.getCause().getMessage());
}
assertEquals(1, stepExecution.getSkipCount());
assertEquals(0, stepExecution.getReadSkipCount());
assertEquals(3, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getReadSkipCount());
assertEquals(1, stepExecution.getWriteSkipCount());
}
@@ -298,16 +307,16 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
@Test
public void testSkipOnReadNotDoubleCounted() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Arrays
.asList(StringUtils.commaDelimitedListToStringArray("2,3,5")));
factory.setSkipLimit(4);
factory.setItemReader(reader);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = jobExecution.createStepExecution(step);
@@ -325,19 +334,19 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
@Test
public void testSkipOnWriteNotDoubleCounted() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6,7"), StringUtils
.commaDelimitedListToSet("2,3"));
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6,7"), Arrays
.asList(StringUtils.commaDelimitedListToStringArray("2,3")));
writer = new SkipWriterStub(StringUtils.commaDelimitedListToSet("4,5"));
writer = new SkipWriterStub(Arrays.asList(StringUtils.commaDelimitedListToStringArray("4,5")));
factory.setSkipLimit(4);
factory.setItemReader(reader);
factory.setItemWriter(writer);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = jobExecution.createStepExecution(step);
@@ -354,15 +363,14 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
}
@SuppressWarnings("unchecked")
@Test
public void testDefaultSkipPolicy() throws Exception {
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
factory.setSkipLimit(1);
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c" }));
ItemReader provider = new ListItemReader(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "a", "b", "c" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
if ("b".equals(item)) {
throw new RuntimeException("Read error - planned failure.");
@@ -371,7 +379,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
}
};
factory.setItemReader(provider);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
@@ -381,6 +389,8 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
assertEquals(4, count);
}
// TODO: test with transactional reader (e.g. list with tx proxy)
/**
* Simple item reader that supports skip functionality.
*/
@@ -394,8 +404,6 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
private int counter = -1;
private int marked = 0;
private final Collection<String> failures;
public SkipReaderStub() {
@@ -424,13 +432,9 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
}
public void mark() throws MarkFailedException {
logger.debug("Marked at count=" + counter);
marked = counter;
}
public void reset() throws ResetFailedException {
counter = marked;
logger.debug("Reset at count=" + counter);
}
}
@@ -442,6 +446,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
protected final Log logger = LogFactory.getLog(getClass());
// simulate transactional output
private List<Object> written = TransactionAwareProxyFactory.createTransactionalList();
private final Collection<String> failures;
@@ -458,12 +463,6 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
this.failures = failures;
}
public void clear() throws ClearFailedException {
}
public void flush() throws FlushFailedException {
}
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
if (failures.contains(item)) {

172
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
View File

@@ -15,15 +15,19 @@
*/
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import junit.framework.TestCase;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
@@ -47,23 +51,24 @@ import org.springframework.batch.retry.RetryException;
import org.springframework.batch.retry.policy.RetryCacheCapacityExceededException;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* @author Dave Syer
*
*/
public class StatefulRetryStepFactoryBeanTests extends TestCase {
public class StatefulRetryStepFactoryBeanTests {
protected final Log logger = LogFactory.getLog(getClass());
private SkipLimitStepFactoryBean<Object,Object> factory = new SkipLimitStepFactoryBean<Object,Object>();
private SkipLimitStepFactoryBean<String, String> factory = new SkipLimitStepFactoryBean<String, String>();
private List<Object> recovered = new ArrayList<Object>();
private List<Object> processed = new ArrayList<Object>();
private List<Object> provided = new ArrayList<Object>();
int count = 0;
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
@@ -71,17 +76,19 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
JobExecution jobExecution;
private ItemWriter<Object> processor = new ItemWriter<Object>() {
public void write(List<? extends Object> data) throws Exception {
private ItemWriter<String> processor = new ItemWriter<String>() {
public void write(List<? extends String> data) throws Exception {
processed.addAll(data);
}
};
/*
* (non-Javadoc)
*
* @see junit.framework.TestCase#setUp()
*/
protected void setUp() throws Exception {
@Before
public void setUp() throws Exception {
MapJobInstanceDao.clear();
MapJobExecutionDao.clear();
@@ -89,7 +96,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setBeanName("step");
factory.setItemReader(new ListItemReader<Object>(new ArrayList<Object>()));
factory.setItemReader(new ListItemReader<String>(new ArrayList<String>()));
factory.setItemWriter(processor);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
@@ -104,10 +111,12 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
}
@Test
public void testType() throws Exception {
assertEquals(Step.class, factory.getObjectType());
}
@Test
public void testDefaultValue() throws Exception {
assertTrue(factory.getObject() instanceof Step);
}
@@ -120,12 +129,13 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
*
* @throws Exception
*/
@Test
public void testSuccessfulRetryWithReadFailure() throws Exception {
List<Object> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c" }));
ItemReader<Object> provider = new ListItemReader<Object>(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "a", "b", "c" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
provided.add(item);
count++;
if (count == 2) {
throw new RuntimeException("Temporary error - retry for success.");
@@ -143,19 +153,24 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertEquals(0, stepExecution.getSkipCount());
// b is processed twice, plus a, plus c, plus the null at end
assertEquals(5, count);
assertEquals(3, stepExecution.getItemCount());
// [a, b, c, null]
assertEquals(4, provided.size());
// [a, c]
assertEquals(2, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(2, stepExecution.getItemCount());
assertEquals(0, stepExecution.getReadSkipCount());
}
@Test
public void testSkipAndRetry() throws Exception {
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
factory.setSkipLimit(2);
List<Object> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }));
ItemReader<Object> provider = new ListItemReader<Object>(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
count++;
if ("b".equals(item) || "d".equals(item)) {
throw new RuntimeException("Read error - planned but skippable.");
@@ -165,7 +180,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
};
factory.setItemReader(provider);
factory.setRetryLimit(10);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);
@@ -176,6 +191,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertEquals(4, stepExecution.getItemCount());
}
@Test
public void testSkipAndRetryWithWriteFailure() throws Exception {
factory.setSkippableExceptionClasses(new Class[] { RetryException.class });
@@ -186,22 +202,23 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
}
} });
factory.setSkipLimit(2);
List<Object> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }));
ItemReader<Object> provider = new ListItemReader<Object>(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
logger.debug("Read Called! Item: [" + item + "]");
provided.add(item);
count++;
return item;
}
};
ItemWriter<Object> itemWriter = new ItemWriter<Object>() {
public void write(List<? extends Object> item) throws Exception {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
logger.debug("Write Called! Item: [" + item + "]");
processed.addAll(item);
if (item.contains("b") || item.contains("d")) {
throw new RuntimeException("Read error - planned but skippable.");
throw new RuntimeException("Write error - planned but recoverable.");
}
}
};
@@ -217,33 +234,39 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
assertEquals(2, recovered.size());
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
// each item once, plus 5 failed retries each for b and d, plus the null
// terminator
assertEquals(17, count);
// [a, b, c, d, e, f, null]
assertEquals(7, provided.size());
// [a, b, b, b, b, b, c, d, d, d, d, d, e, f]
assertEquals(14, processed.size());
// [b, d]
assertEquals(2, recovered.size());
}
@Test
public void testRetryWithNoSkip() throws Exception {
factory.setRetryableExceptionClasses(new Class[] { Exception.class });
factory.setRetryLimit(4);
factory.setSkipLimit(0);
List<Object> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "b" }));
ItemReader<Object> provider = new ListItemReader<Object>(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "b" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
provided.add(item);
count++;
return item;
}
};
ItemWriter<Object> itemWriter = new ItemWriter<Object>() {
public void write(List<? extends Object> item) throws Exception {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
try {
@@ -255,25 +278,31 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
}
assertEquals(0, stepExecution.getSkipCount());
// b is processed 4 times plus the null at end
assertEquals(5, count);
// [b]
assertEquals(1, provided.size());
// [b, b, b, b]
assertEquals(4, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(0, stepExecution.getItemCount());
}
@Test
public void testRetryPolicy() throws Exception {
factory.setRetryPolicy(new SimpleRetryPolicy(4));
factory.setSkipLimit(0);
List<Object> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "b" }));
ItemReader<Object> provider = new ListItemReader<Object>(items) {
public Object read() {
Object item = super.read();
List<String> items = Arrays.asList(new String[] { "b" });
ItemReader<String> provider = new ListItemReader<String>(items) {
public String read() {
String item = super.read();
provided.add(item);
count++;
return item;
}
};
ItemWriter<Object> itemWriter = new ItemWriter<Object>() {
public void write(List<? extends Object> item) throws Exception {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
@@ -292,26 +321,39 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
}
assertEquals(0, stepExecution.getSkipCount());
// b is processed 4 times plus the null at end
assertEquals(5, count);
// [b]
assertEquals(1, provided.size());
// [b, b, b, b]
assertEquals(4, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(0, stepExecution.getItemCount());
}
@Test
public void testCacheLimitWithRetry() throws Exception {
factory.setRetryableExceptionClasses(new Class[] { Exception.class });
factory.setRetryLimit(2);
factory.setCommitInterval(3);
// sufficiently high so we never hit it
factory.setSkipLimit(10);
// set the cache limit lower than the number of unique un-recovered
// errors expected
factory.setCacheCapacity(2);
ItemReader<Object> provider = new AbstractItemReader<Object>() {
public Object read() {
Object item = new Object();
ItemReader<String> provider = new AbstractItemReader<String>() {
public String read() {
String item = ""+count;
provided.add(item);
count++;
if (count >= 10) {
// prevent infinite loop in worst case scenario
return null;
}
return item;
}
};
ItemWriter<Object> itemWriter = new ItemWriter<Object>() {
public void write(List<? extends Object> item) throws Exception {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
@@ -329,8 +371,14 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
// expected
}
assertEquals(0, stepExecution.getSkipCount());
// 2 processed and cached, 3rd barfed because cache was full
assertEquals(3, count);
assertEquals(1, stepExecution.getSkipCount());
// only one item processed but three (the commit interval) were provided
// [0, 1, 2]
assertEquals(3, provided.size());
// TODO: this is a bug: 0 was skipped but it came back in the buffer for the second try
// [0, 0, 1, 0, 0]
assertEquals(5, processed.size());
// []
assertEquals(0, recovered.size());
}
}

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java
View File

@@ -203,7 +203,6 @@ public class RecoveryCallbackRetryPolicy extends AbstractStatefulRetryPolicy {
public Object handleRetryExhausted(RetryContext context) throws ExhaustedRetryException {
// If there is no going back, then we can remove the history
retryContextCache.remove(key);
RepeatSynchronizationManager.setCompleteOnly();
if (recoverer != null) {
return recoverer.recover(context);
}