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

OPEN - issue BATCH-220: Chunk-oriented approach to processing

Browse Source 
Skip should be working now for reads and writes (not processing).
This commit is contained in:
dsyer
2008-08-27 11:08:47 +00:00
parent b55fc3b475
commit 32118407c7
11 changed files with 276 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -48,7 +48,7 @@ public class StepExecution extends Entity {
private volatile int rollbackCount = 0;
private volatile int readSkipCount = 0;
private volatile int writeSkipCount = 0;
private volatile Date startTime = new Date(System.currentTimeMillis());
@@ -234,38 +234,6 @@ public class StepExecution extends Entity {
return null;
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.container.common.domain.Entity#equals(java.lang.Object)
*/
public boolean equals(Object obj) {
Object jobExecutionId = getJobExecutionId();
if (jobExecutionId == null || !(obj instanceof StepExecution) || getId() == null) {
return super.equals(obj);
}
StepExecution other = (StepExecution) obj;
return stepName.equals(other.getStepName()) && (jobExecutionId.equals(other.getJobExecutionId()));
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.container.common.domain.Entity#hashCode()
*/
public int hashCode() {
Object jobExecutionId = getJobExecutionId();
return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91
* (jobExecutionId != null ? jobExecutionId.hashCode() : 0);
}
public String toString() {
return super.toString() + ", name=" + stepName + ", itemCount=" + itemCount + ", commitCount="
+ commitCount + ", rollbackCount=" + rollbackCount;
}
/**
* @param exitStatus
*/
@@ -401,8 +369,7 @@ public class StepExecution extends Entity {
}
/**
* Set the time when the StepExecution was last updated before
* persisting
* Set the time when the StepExecution was last updated before persisting
*
* @param lastUpdated
*/
@@ -410,5 +377,40 @@ public class StepExecution extends Entity {
this.lastUpdated = lastUpdated;
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.container.common.domain.Entity#equals(java.
* lang.Object)
*/
public boolean equals(Object obj) {
Object jobExecutionId = getJobExecutionId();
if (jobExecutionId == null || !(obj instanceof StepExecution) || getId() == null) {
return super.equals(obj);
}
StepExecution other = (StepExecution) obj;
return stepName.equals(other.getStepName()) && (jobExecutionId.equals(other.getJobExecutionId()));
}
/*
* (non-Javadoc)
*
* @see org.springframework.batch.container.common.domain.Entity#hashCode()
*/
public int hashCode() {
Object jobExecutionId = getJobExecutionId();
return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91
* (jobExecutionId != null ? jobExecutionId.hashCode() : 0);
}
public String toString() {
return super.toString()
+ String.format(", name=%s, itemCount=%d, readSkipCount=%d, writeSkipCount=%d"
+ ", commitCount=%d, rollbackCount=%d", stepName, itemCount, readSkipCount, writeSkipCount,
commitCount, rollbackCount);
}
}

2
spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java
View File

@@ -260,6 +260,7 @@ public class StepHandlerStep extends AbstractStep {
// Apply the contribution to the step
// even if unsuccessful
logger.debug("Applying contribution: " + contribution);
stepExecution.apply(contribution);
}
@@ -312,6 +313,7 @@ public class StepHandlerStep extends AbstractStep {
}
try {
logger.debug("Saving step execution after commit: " + stepExecution);
getJobRepository().update(stepExecution);
}
catch (Exception e) {

197
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java
View File

@@ -5,9 +5,13 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import org.springframework.util.Assert;
/**
* Encapsulation of a list of items to be processed and possibly a list of
* failed items to be skipped. To mark an item as skipped clients should iterate
* over the chunk using the {@link #iterator()} method, and if there is a
* failure call {@link ChunkIterator#remove(Exception)} on the iterator. The
* skipped items are then available through the chunk.
*
* @author Dave Syer
*
*/
@@ -15,13 +19,7 @@ class Chunk<W> implements Iterable<W> {
private List<W> items = new ArrayList<W>();
private int current = 0;
private int last = 0;
private Exception exception = null;
private boolean skipped = false;
private List<SkippedItem<W>> skips = new ArrayList<SkippedItem<W>>();
/**
* Add the item to the chunk.
@@ -29,7 +27,6 @@ class Chunk<W> implements Iterable<W> {
*/
public void add(W item) {
items.add(item);
last++;
}
/**
@@ -37,15 +34,20 @@ class Chunk<W> implements Iterable<W> {
*/
public void clear() {
items.clear();
last = 0;
current = 0;
}
/**
* @return a copy of the items to be processed as an unmodifiable list
*/
public List<W> getItems() {
return Collections.unmodifiableList(new ArrayList<W>(items.subList(current, last)));
return Collections.unmodifiableList(new ArrayList<W>(items));
}
/**
* @return a copy of the skips as an unmodifiable list
*/
public List<SkippedItem<W>> getSkips() {
return Collections.unmodifiableList(new ArrayList<SkippedItem<W>>(skips));
}
/**
@@ -59,92 +61,8 @@ class Chunk<W> implements Iterable<W> {
* Get an unmodifiable iterator for the underlying items.
* @see java.lang.Iterable#iterator()
*/
public Iterator<W> iterator() {
return getItems().iterator();
}
/**
* @return true if the chunk is ready for a retry attempt
*/
public boolean canRetry() {
return exception == null || canSkip();
}
/**
* Re-throw the last exception if there was one, and reset. Subsequent calls
* would do nothing until {@link #rethrow(Exception)} is called.
*
* @throws Exception if there is a last exception
*/
public void rethrow() throws Exception {
int size = items.size();
Exception throwable = exception;
if (exception != null && !skipped) {
if (isComplete()) {
// we tried all items and there was no exception
exception = null;
}
else {
// we tried some but not all elements with no exception
current = last;
}
}
if (skipped) {
skipped = false;
}
last = size; // reset end point of scan
if (current == size) {
// we scanned all the elements
current = 0;
exception = null;
}
if (throwable != null) {
throw throwable;
}
}
/**
* @return true if the current item slice includes all items
*/
public boolean isComplete() {
return current == 0 && last == items.size();
}
/**
* Get the skipped item and remove it from the backing list.
* @return the item that can be skipped
*/
public W getSkippedItem() {
Assert.state(canSkip(), "To remove a skipped item it has to be unique");
W item = items.remove(current);
if (last > items.size()) {
last = items.size();
}
skipped = true;
return item;
}
/**
* @param e an exception to register and re-throw
* @throws Exception the exception passed in
*/
public void rethrow(Exception e) throws Exception {
exception = e;
// narrow the search for the failed item
last = current + (last - current) / 2;
// ... unless it would lead to processing no data
if (last==current) {
last = current + 1;
}
throw e;
}
/**
* @return true if there is a single item waiting, so it can be identified
* and passed to listeners
*/
public boolean canSkip() {
return current == last - 1 && current < items.size();
public ChunkIterator iterator() {
return new ChunkIterator(items);
}
/*
@@ -154,7 +72,86 @@ class Chunk<W> implements Iterable<W> {
*/
@Override
public String toString() {
return String.format("items=%s, canSkip=%s, current=%d, last=%d", items, canSkip(), current, last);
return String.format("[items=%s, skips=%s]", items, skips);
}
/**
* Special iterator for a chunk providing the {@link #remove(Exception)}
* method for dynamically removing an item abd adding it to the skips.
*
* @author Dave Syer
*
*/
public class ChunkIterator implements Iterator<W> {
final private Iterator<W> iterator;
private W next;
public ChunkIterator(List<W> items) {
iterator = items.iterator();
}
public boolean hasNext() {
return iterator.hasNext();
}
public W next() {
next = iterator.next();
return next;
}
public void remove(Exception e) {
if (next != null) {
skips.add(new SkippedItem<W>(next, e));
}
iterator.remove();
}
public void remove() {
throw new UnsupportedOperationException("To remove an item you must provide an exception.");
}
}
/**
* Wrapper for a skipped item and its exception.
*
* @author Dave Syer
*
*/
public static class SkippedItem<T> {
final private Exception exception;
final private T item;
public SkippedItem(T item, Exception e) {
this.item = item;
this.exception = e;
}
/**
* Public getter for the exception.
* @return the exception
*/
public Exception getException() {
return exception;
}
/**
* Public getter for the item.
* @return the item
*/
public T getItem() {
return item;
}
@Override
public String toString() {
return String.format("[exception=%s, item=%s]", exception, item);
}
}
}

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

@@ -108,6 +108,11 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
return ExitStatus.CONTINUABLE;
}
});
// If there is no input we don't have to do anything more
if (inputs.isEmpty()) {
return result;
}
storeInputs(attributes, inputs);
@@ -141,7 +146,6 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
clearAll(attributes);
}
logger.info("Contribution: " + contribution);
return result;
}

33
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java
View File

@@ -1,33 +0,0 @@
/*
* Copyright 2006-2007 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.step.item;
import org.springframework.batch.core.Step;
import org.springframework.batch.repeat.RepeatOperations;
/**
* Factory bean for {@link Step} implementations allowing registration of
* listeners and also direct injection of the {@link RepeatOperations} needed at
* step and chunk level.
*
* @deprecated use the {@link SimpleStepFactoryBean} instead
*
* @author Dave Syer
*
*/
public class RepeatOperationsStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
}

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

@@ -2,6 +2,7 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -374,38 +375,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
* the next transaction automatically.<br/>
*/
@Override
protected void write(final Chunk<S> chunk, StepContribution contribution) throws Exception {
if (!chunk.canRetry()) {
logger.debug("Run items: " + chunk.getItems());
runChunk(chunk, contribution);
}
else {
logger.debug(String.format("Retry items: %s", chunk.getItems()));
retryChunk(chunk, contribution);
}
chunk.rethrow();
chunk.clear();
}
/**
* @param chunk
*/
private void runChunk(Chunk<S> chunk, final StepContribution contribution) throws Exception {
try {
doWrite(chunk.getItems());
}
catch (Exception e) {
chunk.rethrow(e);
}
}
/**
* @param chunk
*/
private void retryChunk(final Chunk<S> chunk, final StepContribution contribution) throws Exception {
protected void write(final Chunk<S> chunk, final StepContribution contribution) throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
@@ -422,36 +392,42 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
Exception t = (Exception) context.getLastThrowable();
if (!chunk.canSkip()) {
throw t;
for (Chunk<S>.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) {
S item = iterator.next();
try {
doWrite(Collections.singletonList(item));
}
catch (Exception e) {
if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
iterator.remove(e);
contribution.incrementWriteSkipCount();
throw e;
}
else {
throw new RetryException("Non-skippable exception in recoverer", t);
}
}
}
if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
contribution.incrementWriteSkipCount();
S item = chunk.getSkippedItem();
try {
getListener().onSkipInWrite(item, t);
return null;
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
}
}
else {
throw new RetryException("Non-skippable exception in recoverer", t);
}
return null;
}
};
try {
retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk));
}
catch (Exception e) {
// only if the retry failed do we re-arrange the chunk
chunk.rethrow(e);
retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk));
for (Chunk.SkippedItem<S> skip : chunk.getSkips()) {
Exception exception = skip.getException();
try {
getListener().onSkipInWrite(skip.getItem(), exception);
}
catch (RuntimeException e) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception);
}
}
chunk.clear();
}
}

98
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java
View File

@@ -37,10 +37,6 @@ import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
public class AlmostStatefulRetryChunkTests {
private enum CallType {
RUN, RETRY;
}
private Log logger = LogFactory.getLog(getClass());
private final Chunk<String> chunk;
@@ -53,8 +49,6 @@ public class AlmostStatefulRetryChunkTests {
private int count = 0;
private Object lastCallType;
public AlmostStatefulRetryChunkTests(String[] args, int limit) {
chunk = new Chunk<String>();
for (String string : args) {
@@ -67,88 +61,78 @@ public class AlmostStatefulRetryChunkTests {
public void testRetry() throws Exception {
logger.debug("Starting simple scenario");
List<String> items = new ArrayList<String>(chunk.getItems());
int before = items.size();
items.removeAll(Collections.singleton("fail"));
boolean error = true;
while (error && count++ < BACKSTOP_LIMIT) {
try {
if (!chunk.canRetry()) {
// success
logger.debug("Run items: " + chunk.getItems());
lastCallType = CallType.RUN;
runChunk(chunk);
}
else {
logger.debug(String.format("Retry (attempts=%d) items: %s", retryAttempts, chunk.getItems()));
lastCallType = CallType.RETRY;
try {
retryChunk(chunk);
}
catch (Exception e) {
chunk.rethrow(e);
}
}
chunk.rethrow();
statefulRetry(chunk);
error = false;
}
catch (Exception e) {
error = true;
}
}
logger.debug("Items: " + chunk.getItems());
logger.debug("Chunk: " + chunk);
assertTrue("Backstop reached. Probably an infinite loop...", count < BACKSTOP_LIMIT);
assertEquals(CallType.RETRY, lastCallType);
assertFalse(chunk.getItems().contains("fail"));
assertEquals(items, chunk.getItems());
assertEquals(before-chunk.getItems().size(), chunk.getSkips().size());
}
/**
* @param chunk
* @throws Exception
*/
private void retryChunk(Chunk<String> chunk) throws Exception {
try {
// N.B. a classic stateful retry goes straight to recovery here
doWrite(chunk);
retryAttempts = 0;
}
catch (Exception e) {
if (++retryAttempts > retryLimit) {
// recovery
private void statefulRetry(Chunk<String> chunk) throws Exception {
if (retryAttempts <= retryLimit) {
try {
// N.B. a classic stateful retry goes straight to recovery here
logger.debug(String.format("Retry (attempts=%d) chunk: %s", retryAttempts, chunk));
doWrite(chunk.getItems());
retryAttempts = 0;
if (chunk.canSkip()) {
chunk.getSkippedItem();
}
else {
throw e;
}
}
else {
catch (Exception e) {
retryAttempts++;
// stateful retry always rethrow
throw e;
}
}
else {
try {
logger.debug(String.format("Recover (attempts=%d) chunk: %s", retryAttempts, chunk));
recover(chunk);
}
finally {
retryAttempts = 0;
}
}
// recovery
return;
}
/**
* @param chunk
* @throws Exception
*/
private void recover(Chunk<String> chunk) throws Exception {
for (Chunk<String>.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) {
String string = iterator.next();
try {
doWrite(Collections.singletonList(string));
} catch (Exception e) {
iterator.remove(e);
throw e;
}
}
}
/**
* @param chunk
* @throws Exception
*/
private void runChunk(Chunk<String> chunk) throws Exception {
try {
doWrite(chunk);
}
catch (Exception e) {
chunk.rethrow(e);
}
}
/**
* @param chunk
* @throws Exception
*/
private void doWrite(Chunk<String> chunk) throws Exception {
List<String> items = chunk.getItems();
private void doWrite(List<String> items) throws Exception {
if (items.contains("fail")) {
throw new Exception();
}

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

@@ -93,7 +93,7 @@ public class SkipLimitStepFactoryBeanTests {
// only write exception caused rollback, but more than once because it
// has to go back and split the chunk up to isolate the failed item
assertEquals(3, stepExecution.getRollbackCount());
assertEquals(2, stepExecution.getRollbackCount());
// writer did not skip "2" as it never made it to writer, only "4" did
assertTrue(reader.processed.contains("4"));
@@ -229,8 +229,8 @@ public class SkipLimitStepFactoryBeanTests {
assertFalse(reader.processed.contains("2"));
assertTrue(reader.processed.contains("4"));
// "1" was sent to writer but never comitted
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
// only "1" was ever committed
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1"));
assertEquals(expectedOutput, writer.written);
}
@@ -333,16 +333,14 @@ public class SkipLimitStepFactoryBeanTests {
StepExecution stepExecution = jobExecution.createStepExecution(step);
// TODO: uncomment this!
// step.execute(stepExecution);
// assertEquals(4, stepExecution.getSkipCount());
// assertEquals(3, stepExecution.getReadSkipCount());
// assertEquals(1, stepExecution.getWriteSkipCount());
//
// // skipped 2,3,4,5
// List<String> expectedOutput =
// Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
// assertEquals(expectedOutput, writer.written);
step.execute(stepExecution);
assertEquals(4, stepExecution.getSkipCount());
assertEquals(3, stepExecution.getReadSkipCount());
assertEquals(1, stepExecution.getWriteSkipCount());
// skipped 2,3,4,5
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
assertEquals(expectedOutput, writer.written);
}
@@ -366,16 +364,14 @@ public class SkipLimitStepFactoryBeanTests {
StepExecution stepExecution = jobExecution.createStepExecution(step);
// TODO: uncomment this!
// step.execute(stepExecution);
// assertEquals(4, stepExecution.getSkipCount());
// assertEquals(2, stepExecution.getReadSkipCount());
// assertEquals(2, stepExecution.getWriteSkipCount());
//
// // skipped 2,3,4,5
// List<String> expectedOutput =
// Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
// assertEquals(expectedOutput, writer.written);
step.execute(stepExecution);
assertEquals(4, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getReadSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
// skipped 2,3,4,5
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
assertEquals(expectedOutput, writer.written);
}

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

@@ -52,7 +52,9 @@ import org.springframework.batch.retry.policy.MapRetryContextCache;
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;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
@@ -70,6 +72,8 @@ public class StatefulRetryStepFactoryBeanTests {
private List<Object> provided = new ArrayList<Object>();
private List<Object> written = TransactionAwareProxyFactory.createTransactionalList();
int count = 0;
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
@@ -137,8 +141,7 @@ public class StatefulRetryStepFactoryBeanTests {
*/
@Test
public void testSuccessfulRetryWithReadFailure() throws Exception {
List<String> items = Arrays.asList(new String[] { "a", "b", "c" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -177,8 +180,7 @@ public class StatefulRetryStepFactoryBeanTests {
}
});
factory.setSkipLimit(2);
List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
public String read() {
String item = super.read();
count++;
@@ -216,8 +218,7 @@ public class StatefulRetryStepFactoryBeanTests {
}
} });
factory.setSkipLimit(2);
List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
public String read() {
String item = super.read();
logger.debug("Read Called! Item: [" + item + "]");
@@ -231,6 +232,7 @@ public class StatefulRetryStepFactoryBeanTests {
public void write(List<? extends String> item) throws Exception {
logger.debug("Write Called! Item: [" + item + "]");
processed.addAll(item);
written.addAll(item);
if (item.contains("b") || item.contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
@@ -253,10 +255,13 @@ public class StatefulRetryStepFactoryBeanTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f"));
assertEquals(expectedOutput, written);
// [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());
// [a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]
assertEquals(16, processed.size());
// [b, d]
assertEquals(2, recovered.size());
}
@@ -277,8 +282,7 @@ public class StatefulRetryStepFactoryBeanTests {
}
} });
factory.setSkipLimit(2);
List<String> items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
public String read() {
String item = super.read();
logger.debug("Read Called! Item: [" + item + "]");
@@ -292,6 +296,7 @@ public class StatefulRetryStepFactoryBeanTests {
public void write(List<? extends String> item) throws Exception {
logger.debug("Write Called! Item: [" + item + "]");
processed.addAll(item);
written.addAll(item);
if (item.contains("b") || item.contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
}
@@ -314,10 +319,13 @@ public class StatefulRetryStepFactoryBeanTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f"));
assertEquals(expectedOutput, written);
// [a, b, c, d, e, f, null]
assertEquals(7, provided.size());
// [a, b, c, a, b, c, b, b, b, b, b, c, a, c, d, e, f, d, d, d, d, e, f, e, f]
assertEquals(25, processed.size());
// [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f]
assertEquals(37, processed.size());
// [b, d]
assertEquals(2, recovered.size());
}
@@ -331,8 +339,7 @@ public class StatefulRetryStepFactoryBeanTests {
});
factory.setRetryLimit(4);
factory.setSkipLimit(0);
List<String> items = Arrays.asList(new String[] { "b" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -343,6 +350,7 @@ public class StatefulRetryStepFactoryBeanTests {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
@@ -360,11 +368,15 @@ public class StatefulRetryStepFactoryBeanTests {
// expected
}
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
// [b]
assertEquals(1, provided.size());
// [b, b, b, b]
assertEquals(4, processed.size());
// the failed items are tried one more time than the limit (TODO: maybe fix this?)
// [b, b, b, b, b]
assertEquals(5, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(1, stepExecution.getItemCount());
@@ -383,8 +395,7 @@ public class StatefulRetryStepFactoryBeanTests {
factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>());
factory.setSkipLimit(1);
List<String> items = Arrays.asList(new String[] { "b" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -395,6 +406,7 @@ public class StatefulRetryStepFactoryBeanTests {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but not skippable.");
}
@@ -414,11 +426,14 @@ public class StatefulRetryStepFactoryBeanTests {
assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable."));
}
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
// [b]
assertEquals(1, provided.size());
// [b]
assertEquals(1, processed.size());
// [b, b]
assertEquals(2, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(1, stepExecution.getItemCount());
@@ -428,8 +443,7 @@ public class StatefulRetryStepFactoryBeanTests {
public void testRetryPolicy() throws Exception {
factory.setRetryPolicy(new SimpleRetryPolicy(4));
factory.setSkipLimit(0);
List<String> items = Arrays.asList(new String[] { "b" });
ItemReader<String> provider = new ListItemReader<String>(items) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -440,6 +454,7 @@ public class StatefulRetryStepFactoryBeanTests {
ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> item) throws Exception {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
}
@@ -457,11 +472,14 @@ public class StatefulRetryStepFactoryBeanTests {
// expected
}
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
// [b]
assertEquals(1, provided.size());
// [b, b, b, b]
assertEquals(4, processed.size());
// [b, b, b, b, b]
assertEquals(5, processed.size());
// []
assertEquals(0, recovered.size());
assertEquals(1, stepExecution.getItemCount());

159
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatelessRetryChunkTests.java
View File

@@ -1,159 +0,0 @@
/*
* Copyright 2006-2007 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.step.item;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
/**
* @author Dave Syer
*
*/
@RunWith(Parameterized.class)
public class StatelessRetryChunkTests {
private Log logger = LogFactory.getLog(getClass());
private final Chunk<String> chunk;
private static final int BACKSTOP_LIMIT = 1000;
private int count = 0;
public StatelessRetryChunkTests(String[] args) {
chunk = new Chunk<String>();
for (String string : args) {
chunk.add(string);
}
}
@Test
public void testRetry() throws Exception {
logger.debug("Starting simple scenario");
List<String> items = new ArrayList<String>(chunk.getItems());
int before = items.size();
items.removeAll(Collections.singleton("fail"));
int errors = 0;
boolean error = true;
while (error && count++ < BACKSTOP_LIMIT) {
try {
// success
logger.debug("Run items: " + chunk.getItems());
retryChunk(chunk);
error = false;
}
catch (SpecialException e) {
error = true;
}
catch (Exception e) {
errors++;
error = true;
}
}
logger.debug("Items: " + chunk.getItems());
assertTrue("Backstop reached. Probably an infinite loop...", count < BACKSTOP_LIMIT);
assertFalse(chunk.getItems().contains("fail"));
assertEquals(items, chunk.getItems());
int after = chunk.getItems().size();
logger.debug(String.format("Error count: %d, size before: %d, size after: %d", errors, before, after));
}
/**
* @param chunk
* @throws Exception
*/
private void retryChunk(Chunk<String> chunk) throws Exception {
boolean complete = chunk.isComplete();
try {
doWrite(chunk);
}
catch (Exception e) {
if (chunk.canSkip()) {
chunk.getSkippedItem();
}
else {
if (complete) {
chunk.rethrow(e);
} else {
chunk.rethrow(new SpecialException());
}
}
}
chunk.rethrow();
}
/**
* @param chunk
* @throws Exception
*/
private void doWrite(Chunk<String> chunk) throws Exception {
List<String> items = chunk.getItems();
if (items.contains("fail")) {
throw new Exception();
}
}
@Parameters
public static List<Object[]> data() {
List<Object[]> params = new ArrayList<Object[]>();
params.add(new Object[] { new String[] { "foo" } });
params.add(new Object[] { new String[] { "foo", "bar" } });
params.add(new Object[] { new String[] { "foo", "bar", "spam" } });
params.add(new Object[] { new String[] { "foo", "bar", "spam", "maps", "rab", "oof" } });
params.add(new Object[] { new String[] { "fail" } });
params.add(new Object[] { new String[] { "foo", "fail" } });
params.add(new Object[] { new String[] { "fail", "bar" } });
params.add(new Object[] { new String[] { "foo", "fail", "spam" } });
params.add(new Object[] { new String[] { "fail", "bar", "spam" } });
params.add(new Object[] { new String[] { "foo", "fail", "spam", "maps", "rab", "oof" } });
params.add(new Object[] { new String[] { "foo", "fail", "spam", "fail", "rab", "oof" } });
params.add(new Object[] { new String[] { "fail", "bar", "spam", "fail", "rab", "oof" } });
params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } });
params.add(new Object[] { new String[] { "fail" } });
params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } });
params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } });
return params;
}
/**
* @author Dave Syer
*
*/
public class SpecialException extends Exception {
}
}

5
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
View File

@@ -96,9 +96,8 @@ public class SimpleRetryPolicy implements RetryPolicy {
* attempts so far is less than the limit.
*/
public boolean canRetry(RetryContext context) {
SimpleRetryContext simpleContext = ((SimpleRetryContext) context);
Throwable t = simpleContext.getLastThrowable();
return (t == null || retryForException(t)) && simpleContext.getRetryCount() < maxAttempts;
Throwable t = context.getLastThrowable();
return (t == null || retryForException(t)) && context.getRetryCount() < maxAttempts;
}
/**