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

BATCH-1761: attempted fix using new flag in user data

Browse Source 
- the writing phase needs to communicate with the processing phase
- when processing a chunk that is going to be scanned we need to
  only call the processor on the first item, but otherwise all of
  them
This commit is contained in:
Dave Syer
2012-06-28 14:12:02 +01:00
parent d043fa9e30
commit a0a02b237f
5 changed files with 225 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

263
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
View File

@@ -75,13 +75,15 @@ public class FaultTolerantStepFactoryBeanRetryTests {
private List<Object> provided = new ArrayList<Object>();
private List<Object> written = TransactionAwareProxyFactory.createTransactionalList();
private List<Object> written = TransactionAwareProxyFactory
.createTransactionalList();
int count = 0;
boolean fail = false;
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
private SimpleJobRepository repository = new SimpleJobRepository(
new MapJobInstanceDao(), new MapJobExecutionDao(),
new MapStepExecutionDao(), new MapExecutionContextDao());
JobExecution jobExecution;
@@ -99,7 +101,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory = new FaultTolerantStepFactoryBean<String, String>();
factory.setBeanName("step");
factory.setItemReader(new ListItemReader<String>(new ArrayList<String>()));
factory.setItemReader(new ListItemReader<String>(
new ArrayList<String>()));
factory.setItemWriter(writer);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
@@ -108,8 +111,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique")
.toJobParameters();
JobParameters jobParameters = new JobParametersBuilder().addString(
"statefulTest", "make_this_unique").toJobParameters();
jobExecution = repository.createJobExecution("job", jobParameters);
jobExecution.setEndTime(new Date());
@@ -126,11 +129,13 @@ public class FaultTolerantStepFactoryBeanRetryTests {
}
@Test
public void testProcessAllItemsWhenErrorInWriterTransformation() throws Exception{
public void testProcessAllItemsWhenErrorInWriterTransformation()
throws Exception {
FaultTolerantStepFactoryBean<String, Integer> factory = new FaultTolerantStepFactoryBean<String, Integer>();
factory.setBeanName("step");
factory.setItemReader(new ListItemReader<String>(new ArrayList<String>()));
factory.setItemReader(new ListItemReader<String>(
new ArrayList<String>()));
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
@SuppressWarnings("unchecked")
@@ -145,37 +150,40 @@ public class FaultTolerantStepFactoryBeanRetryTests {
}
written.add(item);
}
}};
ItemProcessor<String, Integer> processor = new ItemProcessor<String, Integer>() {
public Integer process(String item) throws Exception {
processed.add(item);
return Integer.parseInt(item);
}
};
ItemReader<String> reader = new ListItemReader<String>(Arrays.asList("1", "2", "3"));
factory.setCommitInterval(3);
factory.setRetryLimit(3);
factory.setSkippableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
factory.setItemReader(reader);
factory.setItemProcessor(processor);
factory.setItemWriter(failingWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
// System.out.println(stepExecution.getWriteCount());
// System.out.println(processed.size());
// System.out.println(processed);
// System.out.println(written);
assertEquals((1+3)*3, processed.size()); //(Initial try + retry limit)*item count
}
};
ItemProcessor<String, Integer> processor = new ItemProcessor<String, Integer>() {
public Integer process(String item) throws Exception {
processed.add(item);
return Integer.parseInt(item);
}
};
ItemReader<String> reader = new ListItemReader<String>(Arrays.asList(
"1", "2", "3"));
factory.setCommitInterval(3);
factory.setRetryLimit(3);
factory.setSkippableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
factory.setItemReader(reader);
factory.setItemProcessor(processor);
factory.setItemWriter(failingWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
System.out.println(stepExecution.getWriteCount());
System.out.println(processed.size());
System.out.println(processed);
System.out.println(written);
assertEquals((1 + 3) * 3, processed.size()); // (Initial try + retry limit)*item count
}
@Test
public void testProcessAllItemsWhenErrorInWriter() throws Exception {
final int RETRY_LIMIT = 3;
final List<String> ITEM_LIST = Arrays.asList("a", " b", "c");
final List<String> ITEM_LIST = Arrays.asList("a", "b", "c");
ItemWriter<String> failingWriter = new ItemWriter<String>() {
public void write(List<? extends String> data) throws Exception {
int count = 0;
@@ -206,16 +214,25 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setItemWriter(failingWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
// System.out.println(processed);
assertEquals(ExitStatus.COMPLETED.getExitCode(), stepExecution.getExitStatus().getExitCode());
assertEquals((1 + RETRY_LIMIT) * ITEM_LIST.size(), processed.size());
System.out.println(processed);
assertEquals(ExitStatus.COMPLETED.getExitCode(), stepExecution
.getExitStatus().getExitCode());
/*
* Each chunk tried up to RETRY_LIMIT, then the scan processes 1 full
* chunk and fails on the scan, identfies the skip, and then
* re-processes the other n-1 items
*/
assertEquals(RETRY_LIMIT * ITEM_LIST.size() + ITEM_LIST.size()
+ ITEM_LIST.size() - 1, processed.size());
}
@Test
public void testNoItemsReprocessedWhenErrorInWriterAndProcessorNotTransactional() throws Exception{
public void testNoItemsReprocessedWhenErrorInWriterAndProcessorNotTransactional()
throws Exception {
ItemWriter<String> failingWriter = new ItemWriter<String>() {
public void write(List<? extends String> data) throws Exception {
int count = 0;
@@ -225,34 +242,36 @@ public class FaultTolerantStepFactoryBeanRetryTests {
}
written.add(item);
}
}};
ItemProcessor<String, String> processor = new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
processed.add(item);
return item;
}
};
ItemReader<String> reader = new ListItemReader<String>(Arrays.asList("a", "b", "c"));
factory.setProcessorTransactional(false);
factory.setCommitInterval(3);
factory.setRetryLimit(3);
factory.setSkippableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
factory.setItemReader(reader);
factory.setItemProcessor(processor);
factory.setItemWriter(failingWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
System.out.println(stepExecution.getWriteCount());
System.out.println(processed.size());
System.out.println(processed);
System.out.println(written);
assertEquals(3, processed.size()); //(Initial try + retry limit)*item count
}
};
ItemProcessor<String, String> processor = new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
processed.add(item);
return item;
}
};
ItemReader<String> reader = new ListItemReader<String>(Arrays.asList(
"a", "b", "c"));
factory.setProcessorTransactional(false);
factory.setCommitInterval(3);
factory.setRetryLimit(3);
factory.setSkippableExceptionClasses(new HashMap<Class<? extends Throwable>, Boolean>());
factory.setItemReader(reader);
factory.setItemProcessor(processor);
factory.setItemWriter(failingWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
// System.out.println(stepExecution.getWriteCount());
// System.out.println(processed.size());
// System.out.println(processed);
// System.out.println(written);
assertEquals(3, processed.size()); // Initial try only, then cached
}
/**
* N.B. this doesn't really test retry, since the retry is only on write
@@ -265,13 +284,15 @@ public class FaultTolerantStepFactoryBeanRetryTests {
@SuppressWarnings("unchecked")
@Test
public void testSuccessfulRetryWithReadFailure() throws Exception {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c")) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList(
"a", "b", "c")) {
public String read() {
String item = super.read();
provided.add(item);
count++;
if (count == 2) {
throw new RuntimeException("Temporary error - retry for success.");
throw new RuntimeException(
"Temporary error - retry for success.");
}
return item;
}
@@ -281,7 +302,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setSkippableExceptionClasses(getExceptionMap());
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
@@ -313,7 +335,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
@Override
protected void doOpen() throws Exception {
reader = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f"));
reader = new ListItemReader<String>(Arrays.asList("a", "b",
"c", "d", "e", "f"));
}
@Override
@@ -338,7 +361,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
Step step = (Step) factory.getObject();
fail = true;
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
@@ -362,12 +386,14 @@ public class FaultTolerantStepFactoryBeanRetryTests {
public void testSkipAndRetry() throws Exception {
factory.setSkipLimit(2);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList(
"a", "b", "c", "d", "e", "f")) {
public String read() {
String item = super.read();
count++;
if ("b".equals(item) || "d".equals(item)) {
throw new RuntimeException("Read error - planned but skippable.");
throw new RuntimeException(
"Read error - planned but skippable.");
}
return item;
}
@@ -376,7 +402,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setRetryLimit(10);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
@@ -393,11 +420,13 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setListeners(new StepListener[] { new SkipListenerSupport<String, String>() {
public void onSkipInWrite(String item, Throwable t) {
recovered.add(item);
assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
assertTrue(TransactionSynchronizationManager
.isActualTransactionActive());
}
} });
factory.setSkipLimit(2);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
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 + "]");
@@ -413,7 +442,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
processed.addAll(item);
written.addAll(item);
if (item.contains("b") || item.contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
throw new RuntimeException(
"Write error - planned but recoverable.");
}
}
};
@@ -423,7 +453,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setRetryableExceptionClasses(getExceptionMap(RuntimeException.class));
AbstractStep step = (AbstractStep) factory.getObject();
step.setName("mytest");
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
@@ -431,27 +462,32 @@ public class FaultTolerantStepFactoryBeanRetryTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f"));
List<String> expectedOutput = Arrays.asList(StringUtils
.commaDelimitedListToStringArray("a,c,e,f"));
assertEquals(expectedOutput, written);
assertEquals("[a, b, c, d, e, f, null]", provided.toString());
assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", processed.toString());
assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]",
processed.toString());
assertEquals("[b, d]", recovered.toString());
}
@SuppressWarnings("unchecked")
@Test
public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() throws Exception {
public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval()
throws Exception {
factory.setCommitInterval(3);
factory.setListeners(new StepListener[] { new SkipListenerSupport<String, String>() {
public void onSkipInWrite(String item, Throwable t) {
recovered.add(item);
assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
assertTrue(TransactionSynchronizationManager
.isActualTransactionActive());
}
} });
factory.setSkipLimit(2);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
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 + "]");
@@ -467,7 +503,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
processed.addAll(item);
written.addAll(item);
if (item.contains("b") || item.contains("d")) {
throw new RuntimeException("Write error - planned but recoverable.");
throw new RuntimeException(
"Write error - planned but recoverable.");
}
}
};
@@ -477,7 +514,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setRetryableExceptionClasses(getExceptionMap(RuntimeException.class));
AbstractStep step = (AbstractStep) factory.getObject();
step.setName("mytest");
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
@@ -485,7 +523,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
assertEquals(2, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getWriteSkipCount());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f"));
List<String> expectedOutput = Arrays.asList(StringUtils
.commaDelimitedListToStringArray("a,c,e,f"));
assertEquals(expectedOutput, written);
// [a, b, c, d, e, f, null]
@@ -503,7 +542,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setRetryLimit(4);
factory.setSkipLimit(0);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
ItemReader<String> provider = new ListItemReader<String>(
Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -516,19 +556,22 @@ public class FaultTolerantStepFactoryBeanRetryTests {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
throw new RuntimeException(
"Write error - planned but retryable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
List<String> expectedOutput = Arrays.asList(StringUtils
.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
@@ -552,7 +595,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
factory.setRetryableExceptionClasses(getExceptionMap());
factory.setSkipLimit(1);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
ItemReader<String> provider = new ListItemReader<String>(
Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -565,20 +609,25 @@ public class FaultTolerantStepFactoryBeanRetryTests {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but not skippable.");
throw new RuntimeException(
"Write error - planned but not skippable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
String message = stepExecution.getFailureExceptions().get(0).getMessage();
assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable."));
String message = stepExecution.getFailureExceptions().get(0)
.getMessage();
assertTrue("Wrong message: " + message,
message.contains("Write error - planned but not skippable."));
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
List<String> expectedOutput = Arrays.asList(StringUtils
.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
@@ -594,9 +643,11 @@ public class FaultTolerantStepFactoryBeanRetryTests {
@Test
public void testRetryPolicy() throws Exception {
factory.setRetryPolicy(new SimpleRetryPolicy(4, Collections
.<Class<? extends Throwable>, Boolean> singletonMap(Exception.class, true)));
.<Class<? extends Throwable>, Boolean> singletonMap(
Exception.class, true)));
factory.setSkipLimit(0);
ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
ItemReader<String> provider = new ListItemReader<String>(
Arrays.asList("b")) {
public String read() {
String item = super.read();
provided.add(item);
@@ -609,19 +660,22 @@ public class FaultTolerantStepFactoryBeanRetryTests {
processed.addAll(item);
written.addAll(item);
logger.debug("Write Called! Item: [" + item + "]");
throw new RuntimeException("Write error - planned but retryable.");
throw new RuntimeException(
"Write error - planned but retryable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
List<String> expectedOutput = Arrays.asList(StringUtils
.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, written);
assertEquals(0, stepExecution.getSkipCount());
@@ -657,14 +711,16 @@ public class FaultTolerantStepFactoryBeanRetryTests {
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.");
throw new RuntimeException(
"Write error - planned but retryable.");
}
};
factory.setItemReader(provider);
factory.setItemWriter(itemWriter);
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepExecution stepExecution = new StepExecution(step.getName(),
jobExecution);
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
@@ -680,7 +736,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
assertEquals(0, recovered.size());
}
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
private Map<Class<? extends Throwable>, Boolean> getExceptionMap(
Class<? extends Throwable>... args) {
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>();
for (Class<? extends Throwable> arg : args) {
map.put(arg, true);

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -434,6 +434,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
assertEquals("[1, 3, 5]", writer.getWritten().toString());
assertEquals("[1, 3, 5]", writer.getCommitted().toString());
// If non-transactional, we should only process each item once
assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
}

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

@@ -717,9 +717,9 @@ public class FaultTolerantStepFactoryBeanTests {
assertEquals(1, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getRollbackCount());
// 1,2,3,4,3,4 - two re-processing attempts until the item is
// 1,2,3,4,3,4,4 - two re-processing attempts until the item is
// identified and finally skipped on the second attempt
assertEquals("[1, 2, 3, 4, 3, 4]", processor.getProcessed().toString());
assertEquals("[1, 2, 3, 4, 3, 4, 4]", processor.getProcessed().toString());
assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step
.getName()));
}