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

Unwrap StepListenerFailedException when determining skip/rollback

Browse Source 
When determining if an exception should be skipped or not, the business
exception should be used.  However, when an exception is thrown in the
listener, it's wrapped by a `StepListenerFailedException`.  This leads
to it hiding the underlying exception.  This commit will unwrap the
cause of a `StepListenerFailedException` and use that to determine if it
should be skipped or not.

Resolves BATCH-2322
This commit is contained in:
Michael Minella
2015-05-08 22:37:25 +01:00
parent ca847e2760
commit eb2f752b16
3 changed files with 334 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -16,9 +16,18 @@
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.listener.StepListenerFailedException;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableProcessException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
@@ -35,13 +44,6 @@ import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryException;
import org.springframework.retry.support.DefaultRetryState;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
/**
* FaultTolerant implementation of the {@link ChunkProcessor} interface, that
* allows for skipping or retry of items that cause exceptions during writing.
@@ -584,16 +586,25 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
outputIterator.remove();
}
catch (Exception e) {
doOnWriteError(e, items);
if (!shouldSkip(itemWriteSkipPolicy, e, -1) && !rollbackClassifier.classify(e)) {
inputIterator.remove();
outputIterator.remove();
try {
doOnWriteError(e, items);
}
else {
checkSkipPolicy(inputIterator, outputIterator, e, contribution, recovery);
}
if (rollbackClassifier.classify(e)) {
throw e;
finally {
Throwable cause = e;
if(e instanceof StepListenerFailedException) {
cause = e.getCause();
}
if (!shouldSkip(itemWriteSkipPolicy, cause, -1) && !rollbackClassifier.classify(cause)) {
inputIterator.remove();
outputIterator.remove();
}
else {
checkSkipPolicy(inputIterator, outputIterator, cause, contribution, recovery);
}
if (rollbackClassifier.classify(cause)) {
throw (Exception) cause;
}
}
}
chunkMonitor.incrementOffset();

304
spring-batch-core/src/test/java/org/springframework/batch/core/listener/ItemListenerErrorTests.java Normal file
View File

@@ -0,0 +1,304 @@
/*
* Copyright 2015 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.listener;
import static org.junit.Assert.assertEquals;
import java.util.Collections;
import java.util.List;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* BATCH-2322
*
* @author Michael Minella
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = {ItemListenerErrorTests.BatchConfiguration.class})
public class ItemListenerErrorTests {
@Autowired
private FailingListener listener;
@Autowired
private FailingItemReader reader;
@Autowired
private FailingItemProcessor processor;
@Autowired
private FailingItemWriter writer;
@Autowired
private JobLauncher jobLauncher;
@Autowired
private Job job;
@Before
public void setUp() {
listener.setMethodToThrowExceptionFrom("");
reader.setGoingToFail(false);
processor.setGoingToFail(false);
writer.setGoingToFail(false);
}
@Test
@DirtiesContext
public void testOnWriteError() throws Exception {
listener.setMethodToThrowExceptionFrom("onWriteError");
writer.setGoingToFail(true);
JobExecution execution = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
}
@Ignore
@Test
@DirtiesContext
public void testOnReadError() throws Exception {
listener.setMethodToThrowExceptionFrom("onReadError");
reader.setGoingToFail(true);
JobExecution execution = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
}
@Test
@DirtiesContext
public void testOnProcessError() throws Exception {
listener.setMethodToThrowExceptionFrom("onProcessError");
processor.setGoingToFail(true);
JobExecution execution = jobLauncher.run(job, new JobParameters());
assertEquals(BatchStatus.COMPLETED, execution.getStatus());
}
@Configuration
@EnableBatchProcessing
public static class BatchConfiguration {
@Bean
public Job testJob(JobBuilderFactory jobs, Step testStep) {
return jobs.get("testJob")
.incrementer(new RunIdIncrementer())
.start(testStep)
.build();
}
@Bean
public Step step1(StepBuilderFactory stepBuilderFactory,
ItemReader<String> fakeItemReader,
ItemProcessor<String, String> fakeProcessor,
ItemWriter<String> fakeItemWriter,
ItemProcessListener<String, String> itemProcessListener) {
return stepBuilderFactory.get("testStep").<String, String>chunk(10)
.reader(fakeItemReader)
.processor(fakeProcessor)
.writer(fakeItemWriter)
.listener(itemProcessListener)
.faultTolerant().skipLimit(50).skip(RuntimeException.class)
.build();
}
@Bean
public FailingListener itemListener() {
return new FailingListener();
}
@Bean
public FailingItemReader fakeReader() {
return new FailingItemReader();
}
@Bean
public FailingItemProcessor fakeProcessor() {
return new FailingItemProcessor();
}
@Bean
public FailingItemWriter fakeItemWriter() {
return new FailingItemWriter();
}
}
public static class FailingItemWriter implements ItemWriter<String> {
private boolean goingToFail = false;
@Override
public void write(List<? extends String> items) throws Exception {
if(goingToFail) {
throw new RuntimeException("failure in the writer");
}
else {
for (String item : items) {
System.out.println(item);
}
}
}
public void setGoingToFail(boolean goingToFail) {
this.goingToFail = goingToFail;
}
}
public static class FailingItemProcessor implements ItemProcessor<String, String> {
private boolean goingToFail = false;
@Override
public String process(String item) throws Exception {
if(goingToFail) {
throw new RuntimeException("failure in the processor");
}
else {
return item;
}
}
public void setGoingToFail(boolean goingToFail) {
this.goingToFail = goingToFail;
}
}
public static class FailingItemReader implements ItemReader<String> {
private boolean goingToFail = false;
private ItemReader<String> delegate = new ListItemReader<String>(Collections.singletonList("1"));
private int count = 0;
@Override
public String read() throws Exception {
count++;
if(goingToFail) {
throw new RuntimeException("failure in the reader");
}
else {
return delegate.read();
}
}
public void setGoingToFail(boolean goingToFail) {
this.goingToFail = goingToFail;
}
public int getCount() {
return count;
}
}
public static class FailingListener extends ItemListenerSupport<String, String> {
private String methodToThrowExceptionFrom;
public void setMethodToThrowExceptionFrom(String methodToThrowExceptionFrom) {
this.methodToThrowExceptionFrom = methodToThrowExceptionFrom;
}
@Override
public void beforeRead() {
if (methodToThrowExceptionFrom.equals("beforeRead")) {
throw new RuntimeException("beforeRead caused this Exception");
}
}
@Override
public void afterRead(String item) {
if (methodToThrowExceptionFrom.equals("afterRead")) {
throw new RuntimeException("afterRead caused this Exception");
}
}
@Override
public void onReadError(Exception ex) {
if (methodToThrowExceptionFrom.equals("onReadError")) {
throw new RuntimeException("onReadError caused this Exception");
}
}
@Override
public void beforeProcess(String item) {
if (methodToThrowExceptionFrom.equals("beforeProcess")) {
throw new RuntimeException("beforeProcess caused this Exception");
}
}
@Override
public void afterProcess(String item, String result) {
if (methodToThrowExceptionFrom.equals("afterProcess")) {
throw new RuntimeException("afterProcess caused this Exception");
}
}
@Override
public void onProcessError(String item, Exception ex) {
if (methodToThrowExceptionFrom.equals("onProcessError")) {
throw new RuntimeException("onProcessError caused this Exception");
}
}
@Override
public void beforeWrite(List<? extends String> items) {
if (methodToThrowExceptionFrom.equals("beforeWrite")) {
throw new RuntimeException("beforeWrite caused this Exception");
}
}
@Override
public void afterWrite(List<? extends String> items) {
if (methodToThrowExceptionFrom.equals("afterWrite")) {
throw new RuntimeException("afterWrite caused this Exception");
}
}
@Override
public void onWriteError(Exception ex, List<? extends String> item) {
if (methodToThrowExceptionFrom.equals("onWriteError")) {
throw new RuntimeException("onWriteError caused this Exception");
}
}
}
}

6
spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/RegisterMultiListenerTests.java
View File

@@ -15,14 +15,12 @@
*/
package org.springframework.batch.core.step.builder;
import static org.junit.Assert.assertEquals;
import java.util.List;
import javax.sql.DataSource;
import org.junit.After;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ExitStatus;
@@ -55,6 +53,8 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
import static org.junit.Assert.assertEquals;
/**
* Test for registering a listener class that implements different listeners interfaces
* just once in java based configuration.