BATCH-2005: Implement JSR RetryListener along side Spring Batch's RetryListener

This commit is contained in:
Chris Schaefer
2013-09-30 02:51:49 -04:00
committed by Michael Minella
parent df53ea12b3
commit e53533e9e3
22 changed files with 813 additions and 37 deletions

View File

@@ -23,6 +23,9 @@ import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.batch.api.chunk.listener.SkipProcessListener;
import javax.batch.api.chunk.listener.SkipReadListener;
import javax.batch.api.chunk.listener.SkipWriteListener;
@@ -41,6 +44,9 @@ import org.springframework.batch.core.jsr.ChunkListenerAdapter;
import org.springframework.batch.core.jsr.ItemProcessListenerAdapter;
import org.springframework.batch.core.jsr.ItemReadListenerAdapter;
import org.springframework.batch.core.jsr.ItemWriteListenerAdapter;
import org.springframework.batch.core.jsr.RetryProcessListenerAdapter;
import org.springframework.batch.core.jsr.RetryReadListenerAdapter;
import org.springframework.batch.core.jsr.RetryWriteListenerAdapter;
import org.springframework.batch.core.jsr.SkipListenerAdapter;
import org.springframework.batch.core.jsr.StepListenerAdapter;
import org.springframework.batch.core.launch.JobLauncher;
@@ -96,6 +102,7 @@ import org.springframework.util.Assert;
* @author Dan Garrette
* @author Josh Long
* @author Michael Minella
* @author Chris Schaefer
* @see SimpleStepFactoryBean
* @see FaultTolerantStepFactoryBean
* @see TaskletStep
@@ -222,6 +229,8 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
private Set<SkipListener<I, O>> skipListeners = new LinkedHashSet<SkipListener<I, O>>();
private Set<org.springframework.batch.core.jsr.RetryListener> jsrRetryListeners = new LinkedHashSet<org.springframework.batch.core.jsr.RetryListener>();
//
// Additional
//
@@ -333,6 +342,10 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
builder.listener(listener);
}
for (org.springframework.batch.core.jsr.RetryListener listener : jsrRetryListeners) {
builder.listener(listener);
}
registerItemListeners(builder);
if (skipPolicy != null) {
@@ -797,6 +810,15 @@ public class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAwa
ItemProcessListener itemListener = new ItemProcessListenerAdapter((javax.batch.api.chunk.listener.ItemProcessListener) listener);
processListeners.add(itemListener);
}
if(listener instanceof RetryReadListener) {
jsrRetryListeners.add(new RetryReadListenerAdapter((RetryReadListener) listener));
}
if(listener instanceof RetryProcessListener) {
jsrRetryListeners.add(new RetryProcessListenerAdapter((RetryProcessListener) listener));
}
if(listener instanceof RetryWriteListener) {
jsrRetryListeners.add(new RetryWriteListenerAdapter((RetryWriteListener) listener));
}
}
}

View File

@@ -24,6 +24,9 @@ import javax.batch.api.chunk.listener.ChunkListener;
import javax.batch.api.chunk.listener.ItemProcessListener;
import javax.batch.api.chunk.listener.ItemReadListener;
import javax.batch.api.chunk.listener.ItemWriteListener;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.batch.api.chunk.listener.SkipProcessListener;
import javax.batch.api.chunk.listener.SkipReadListener;
import javax.batch.api.chunk.listener.SkipWriteListener;
@@ -37,11 +40,11 @@ import org.springframework.batch.core.listener.StepListenerFactoryBean;
* ties together the names of methods, their interfaces, and expected arguments.
*
* @author Michael Minella
* @author Chris Schaefer
* @since 3.0
* @see StepListenerFactoryBean
*/
public enum JsrStepListenerMetaData implements ListenerMetaData {
BEFORE_STEP("beforeStep", "jsr-before-step", StepListener.class),
AFTER_STEP("afterStep", "jsr-after-step", StepListener.class),
BEFORE_CHUNK("beforeChunk", "jsr-before-chunk", ChunkListener.class),
@@ -58,7 +61,10 @@ public enum JsrStepListenerMetaData implements ListenerMetaData {
AFTER_WRITE_ERROR("onWriteError", "jsr-after-write-error", ItemWriteListener.class, List.class, Exception.class),
SKIP_READ("onSkipReadItem", "jsr-skip-read", SkipReadListener.class, Exception.class),
SKIP_PROCESS("onSkipProcessItem", "jsr-skip-process", SkipProcessListener.class, Object.class, Exception.class),
SKIP_WRITE("onSkipWriteItem", "jsr-skip-write", SkipWriteListener.class, List.class, Exception.class);
SKIP_WRITE("onSkipWriteItem", "jsr-skip-write", SkipWriteListener.class, List.class, Exception.class),
RETRY_READ("onRetryReadException", "jsr-retry-read", RetryReadListener.class, Exception.class),
RETRY_PROCESS("onRetryProcessException", "jsr-retry-process", RetryProcessListener.class, Object.class, Exception.class),
RETRY_WRITE("onRetryWriteException", "jsr-retry-write", RetryWriteListener.class, List.class, Exception.class);
private final String methodName;
private final String propertyName;

View File

@@ -0,0 +1,30 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.jsr;
import org.springframework.batch.core.StepListener;
/**
* <p>
* Marker interface to be implemented by JSR-352 retry listeners. Extends {@link StepListener}
* to allow registration with existing listener methods.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public interface RetryListener extends StepListener {
}

View File

@@ -0,0 +1,44 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.jsr;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.batch.operations.BatchRuntimeException;
/**
* <p>
* Wrapper class to adapt a {@link RetryProcessListener} to a {@link RetryListener}.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class RetryProcessListenerAdapter implements RetryListener, RetryProcessListener {
private RetryProcessListener retryProcessListener;
public RetryProcessListenerAdapter(RetryProcessListener retryProcessListener) {
this.retryProcessListener = retryProcessListener;
}
@Override
public void onRetryProcessException(Object item, Exception ex) throws Exception {
try {
retryProcessListener.onRetryProcessException(item, ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
}

View File

@@ -0,0 +1,44 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.jsr;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.batch.operations.BatchRuntimeException;
/**
* <p>
* Wrapper class to adapt a {@link RetryReadListener} to a {@link RetryListener}.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class RetryReadListenerAdapter implements RetryListener, RetryReadListener {
private RetryReadListener retryReadListener;
public RetryReadListenerAdapter(RetryReadListener retryReadListener) {
this.retryReadListener = retryReadListener;
}
@Override
public void onRetryReadException(Exception ex) throws Exception {
try {
retryReadListener.onRetryReadException(ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
}

View File

@@ -0,0 +1,45 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.jsr;
import java.util.List;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.batch.operations.BatchRuntimeException;
/**
* <p>
* Wrapper class to adapt a {@link RetryWriteListener} to a {@link RetryListener}.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class RetryWriteListenerAdapter implements RetryListener, RetryWriteListener {
private RetryWriteListener retryWriteListener;
public RetryWriteListenerAdapter(RetryWriteListener retryWriteListener) {
this.retryWriteListener = retryWriteListener;
}
@Override
public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
try {
retryWriteListener.onRetryWriteException(items, ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
}

View File

@@ -69,6 +69,6 @@ public class JobParser extends AbstractSingleBeanDefinitionParser {
parserContext.getRegistry().registerBeanDefinition("stepContextFactory", stepContextBeanDefinition);
new ListnerParser(JobListenerFactoryBean.class, "jobExecutionListeners").parseListeners(element, parserContext, builder);
new ListenerParser(JobListenerFactoryBean.class, "jobExecutionListeners").parseListeners(element, parserContext, builder);
}
}

View File

@@ -33,9 +33,10 @@ import org.w3c.dom.Element;
* and not JSR interfaces
*
* @author Michael Minella
* @author Chris Schaefer
* @since 3.0
*/
public class ListnerParser {
public class ListenerParser {
private static final String REF_ATTRIBUTE = "ref";
private static final String LISTENER_ELEMENT = "listener";
private static final String LISTENERS_ELEMENT = "listeners";
@@ -44,7 +45,7 @@ public class ListnerParser {
private String propertyKey;
@SuppressWarnings("rawtypes")
public ListnerParser(Class listenerType, String propertyKey) {
public ListenerParser(Class listenerType, String propertyKey) {
this.propertyKey = propertyKey;
this.listenerType = listenerType;
}

View File

@@ -70,7 +70,7 @@ public class StepParser extends AbstractSingleBeanDefinitionParser {
allowStartIfComplete);
}
new ListnerParser(StepListenerFactoryBean.class, "listeners").parseListeners(element, parserContext, bd, stepName);
new ListenerParser(StepListenerFactoryBean.class, "listeners").parseListeners(element, parserContext, bd, stepName);
new PropertyParser(stepName, parserContext, BatchArtifact.BatchArtifactType.STEP).parseProperties(element);
// look at all nested elements
@@ -91,8 +91,6 @@ public class StepParser extends AbstractSingleBeanDefinitionParser {
}
}
Collection<BeanDefinition> nextElements = FlowParser.getNextElements(parserContext, stepName, stateBuilder.getBeanDefinition(), element);
return nextElements;
return FlowParser.getNextElements(parserContext, stepName, stateBuilder.getBeanDefinition(), element);
}
}

View File

@@ -42,7 +42,7 @@ public class JsrFlowExecutor extends JobFlowExecutor {
*/
@Override
public void addExitStatus(String code) {
if((exitStatus != null && isNonDefaultExitStauts(exitStatus.getExitCode())) && !isNonDefaultExitStauts(code)) {
if((exitStatus != null && isNonDefaultExitStatus(exitStatus.getExitCode())) && !isNonDefaultExitStatus(code)) {
exitStatus = exitStatus.and(new ExitStatus(code));
}
}
@@ -57,7 +57,7 @@ public class JsrFlowExecutor extends JobFlowExecutor {
execution.setStatus(findBatchStatus(status));
ExitStatus curStatus = execution.getExitStatus();
if(isNonDefaultExitStauts(curStatus.getExitCode())) {
if(isNonDefaultExitStatus(curStatus.getExitCode())) {
exitStatus = exitStatus.and(new ExitStatus(status.getName()));
execution.setExitStatus(exitStatus);
}
@@ -67,7 +67,7 @@ public class JsrFlowExecutor extends JobFlowExecutor {
* @param curStatus the exit code to be evaluated
* @return true if the value matches a known exit code
*/
protected boolean isNonDefaultExitStauts(String curStatus) {
protected boolean isNonDefaultExitStatus(String curStatus) {
return curStatus == null ||
curStatus.equals(ExitStatus.COMPLETED.getExitCode()) ||
curStatus.equals(ExitStatus.EXECUTING.getExitCode()) ||

View File

@@ -17,6 +17,7 @@ package org.springframework.batch.core.jsr.step.builder;
import java.util.ArrayList;
import java.util.List;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.jsr.step.item.JsrChunkProvider;
import org.springframework.batch.core.jsr.step.item.JsrFaultTolerantChunkProcessor;
@@ -33,6 +34,7 @@ import org.springframework.batch.core.step.skip.SkipPolicy;
* pattern defined by the spec as well as skip/retry logic.
*
* @author Michael Minella
* @author Chris Schaefer
*
* @param <I> The input type for the step
* @param <O> The output type for the step
@@ -70,10 +72,17 @@ public class JsrFaultTolerantStepBuilder<I, O> extends FaultTolerantStepBuilder<
chunkProcessor.setRollbackClassifier(getRollbackClassifier());
detectStreamInReader();
chunkProcessor.setChunkMonitor(getChunkMonitor());
ArrayList<StepListener> listeners = new ArrayList<StepListener>(getItemListeners());
listeners.addAll(getSkipListeners());
chunkProcessor.setListeners(listeners);
chunkProcessor.setListeners(getChunkListeners());
return chunkProcessor;
}
private List<StepListener> getChunkListeners() {
List<StepListener> listeners = new ArrayList<StepListener>();
listeners.addAll(getItemListeners());
listeners.addAll(getSkipListeners());
listeners.addAll(getJsrRetryListeners());
return listeners;
}
}

View File

@@ -17,6 +17,7 @@ package org.springframework.batch.core.jsr.step.item;
import java.util.List;
import javax.batch.operations.BatchRuntimeException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
@@ -45,12 +46,12 @@ import org.springframework.util.Assert;
* Extension of the {@link JsrChunkProcessor} that adds skip and retry functionality.
*
* @author Michael Minella
* @author Chris Schaefer
*
* @param <I> input type for the step
* @param <O> output type for the step
*/
public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O> {
protected final Log logger = LogFactory.getLog(getClass());
private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy();
private Classifier<Throwable, Boolean> rollbackClassifier = new BinaryExceptionClassifier(true);
@@ -58,10 +59,6 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
private ChunkMonitor chunkMonitor = new ChunkMonitor();
private boolean hasProcessor = false;
public JsrFaultTolerantChunkProcessor() {
this(null, null, null, null, null);
}
public JsrFaultTolerantChunkProcessor(ItemReader<I> reader, ItemProcessor<I,O> processor, ItemWriter<O> writer, RepeatOperations repeatTemplate, BatchRetryTemplate batchRetryTemplate) {
super(reader, processor, writer, repeatTemplate);
hasProcessor = processor != null;
@@ -148,10 +145,14 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
logger.debug("Skipping failed input", e);
}
else {
if (rollbackClassifier.classify(e)) {
getListener().onRetryReadException(e);
if(rollbackClassifier.classify(e)) {
throw e;
}
else {
throw e;
}
logger.debug("No-rollback for non-skippable exception (ignored)", e);
}
}
}
@@ -174,7 +175,8 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
// allows us to continue
throw new RetryException("Non-skippable exception in recoverer while reading", e);
}
return null;
throw new BatchRuntimeException(e);
}
}
@@ -234,13 +236,17 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
// If not re-throwing then the listener will not be
// called in next chunk.
getListener().onSkipInProcess(item, e);
} else if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw e;
}
else {
throw e;
} else {
getListener().onRetryProcessException(item, e);
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw e;
}
else {
throw e;
}
}
}
return null;
@@ -249,7 +255,6 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
};
RecoveryCallback<O> recoveryCallback = new RecoveryCallback<O>() {
@Override
public O recover(RetryContext context) throws Exception {
Throwable e = context.getLastThrowable();
@@ -264,10 +269,10 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
// allows us to continue
throw new RetryException("Non-skippable exception in recoverer while processing", e);
}
return null;
throw new BatchRuntimeException(e);
}
}
};
return batchRetryTemplate.execute(retryCallback, recoveryCallback);
@@ -295,8 +300,12 @@ public class JsrFaultTolerantChunkProcessor<I,O> extends JsrChunkProcessor<I, O>
catch (Exception e) {
if(shouldSkip(skipPolicy, e, contribution.getStepSkipCount())) {
getListener().onSkipInWrite(chunk.getItems(), e);
} else if (rollbackClassifier.classify(e)) {
throw e;
} else {
getListener().onRetryWriteException(chunk.getItems(), e);
if (rollbackClassifier.classify(e)) {
throw e;
}
}
/*
* If the exception is marked as no-rollback, we need to

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.listener;
import java.util.Iterator;
import java.util.List;
import javax.batch.api.chunk.listener.RetryProcessListener;
/**
* <p>
* Composite class holding {@link RetryProcessListener}'s.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class CompositeRetryProcessListener implements RetryProcessListener {
private OrderedComposite<RetryProcessListener> listeners = new OrderedComposite<RetryProcessListener>();
/**
* <p>
* Public setter for the {@link RetryProcessListener}'s.
* </p>
*
* @param listeners the {@link RetryProcessListener}'s to set
*/
public void setListeners(List<? extends RetryProcessListener> listeners) {
this.listeners.setItems(listeners);
}
/**
* <p>
* Register an additional {@link RetryProcessListener}.
* </p>
*
* @param listener the {@link RetryProcessListener} to register
*/
public void register(RetryProcessListener listener) {
listeners.add(listener);
}
@Override
public void onRetryProcessException(Object item, Exception ex) throws Exception {
for (Iterator<RetryProcessListener> iterator = listeners.reverse(); iterator.hasNext();) {
RetryProcessListener listener = iterator.next();
listener.onRetryProcessException(item, ex);
}
}
}

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.listener;
import java.util.Iterator;
import java.util.List;
import javax.batch.api.chunk.listener.RetryReadListener;
/**
* <p>
* Composite class holding {@link RetryReadListener}'s.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class CompositeRetryReadListener implements RetryReadListener {
private OrderedComposite<RetryReadListener> listeners = new OrderedComposite<RetryReadListener>();
/**
* <p>
* Public setter for the {@link RetryReadListener}'s.
* </p>
*
* @param listeners the {@link RetryReadListener}'s to set
*/
public void setListeners(List<? extends RetryReadListener> listeners) {
this.listeners.setItems(listeners);
}
/**
* <p>
* Register an additional {@link RetryReadListener}.
* </p>
*
* @param listener the {@link RetryReadListener} to register
*/
public void register(RetryReadListener listener) {
listeners.add(listener);
}
@Override
public void onRetryReadException(Exception ex) throws Exception {
for (Iterator<RetryReadListener> iterator = listeners.reverse(); iterator.hasNext();) {
RetryReadListener listener = iterator.next();
listener.onRetryReadException(ex);
}
}
}

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.listener;
import java.util.Iterator;
import java.util.List;
import javax.batch.api.chunk.listener.RetryWriteListener;
/**
* <p>
* Composite class holding {@link RetryWriteListener}'s.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class CompositeRetryWriteListener implements RetryWriteListener {
private OrderedComposite<RetryWriteListener> listeners = new OrderedComposite<RetryWriteListener>();
/**
* <p>
* Public setter for the {@link RetryWriteListener}'s.
* </p>
*
* @param listeners the {@link RetryWriteListener}'s to set
*/
public void setListeners(List<? extends RetryWriteListener> listeners) {
this.listeners.setItems(listeners);
}
/**
* <p>
* Register an additional {@link RetryWriteListener}.
* </p>
*
* @param listener the {@link RetryWriteListener} to register
*/
public void register(RetryWriteListener listener) {
listeners.add(listener);
}
@Override
public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
for (Iterator<RetryWriteListener> iterator = listeners.reverse(); iterator.hasNext();) {
RetryWriteListener listener = iterator.next();
listener.onRetryWriteException(items, ex);
}
}
}

View File

@@ -17,6 +17,10 @@ package org.springframework.batch.core.listener;
import java.util.List;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.batch.operations.BatchRuntimeException;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.ItemProcessListener;
@@ -32,9 +36,10 @@ import org.springframework.batch.item.ItemStream;
/**
* @author Dave Syer
* @author Michael Minella
* @author Chris Schaefer
*/
public class MulticasterBatchListener<T, S> implements StepExecutionListener, ChunkListener, ItemReadListener<T>,
ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T, S> {
ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T, S>, RetryReadListener, RetryProcessListener, RetryWriteListener {
private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener();
@@ -48,6 +53,12 @@ ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T, S> {
private CompositeSkipListener<T, S> skipListener = new CompositeSkipListener<T, S>();
private CompositeRetryReadListener retryReadListener = new CompositeRetryReadListener();
private CompositeRetryProcessListener retryProcessListener = new CompositeRetryProcessListener();
private CompositeRetryWriteListener retryWriteListener = new CompositeRetryWriteListener();
/**
* Initialize the listener instance.
*/
@@ -99,6 +110,15 @@ ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T, S> {
SkipListener<T, S> skipListener = (SkipListener<T, S>) listener;
this.skipListener.register(skipListener);
}
if(listener instanceof RetryReadListener) {
this.retryReadListener.register((RetryReadListener) listener);
}
if(listener instanceof RetryProcessListener) {
this.retryProcessListener.register((RetryProcessListener) listener);
}
if(listener instanceof RetryWriteListener) {
this.retryWriteListener.register((RetryWriteListener) listener);
}
}
/**
@@ -327,4 +347,31 @@ ItemProcessListener<T, S>, ItemWriteListener<S>, SkipListener<T, S> {
throw new StepListenerFailedException("Error in afterFailedChunk.", e);
}
}
@Override
public void onRetryReadException(Exception ex) throws Exception {
try {
retryReadListener.onRetryReadException(ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
@Override
public void onRetryProcessException(Object item, Exception ex) throws Exception {
try {
retryProcessListener.onRetryProcessException(item, ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
@Override
public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
try {
retryWriteListener.onRetryWriteException(items, ex);
} catch (Exception e) {
throw new BatchRuntimeException(e);
}
}
}

View File

@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.batch.operations.BatchRuntimeException;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.SkipListener;
@@ -80,6 +81,7 @@ import org.springframework.util.Assert;
* additional properties for retry and skip of failed items.
*
* @author Dave Syer
* @author Chris Schaefer
*
* @since 2.2
*/
@@ -113,6 +115,8 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
private Set<SkipListener<? super I, ? super O>> skipListeners = new LinkedHashSet<SkipListener<? super I, ? super O>>();
private Set<org.springframework.batch.core.jsr.RetryListener> jsrRetryListeners = new LinkedHashSet<org.springframework.batch.core.jsr.RetryListener>();
private int skipLimit = 0;
private SkipPolicy skipPolicy;
@@ -186,6 +190,11 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
return this;
}
public FaultTolerantStepBuilder<I, O> listener(org.springframework.batch.core.jsr.RetryListener listener) {
jsrRetryListeners.add(listener);
return this;
}
@Override
public FaultTolerantStepBuilder<I, O> listener(ChunkListener listener) {
super.listener(new TerminateOnExceptionChunkListenerDelegate(listener));
@@ -434,7 +443,8 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
TransactionException.class, FatalStepExecutionException.class, SkipListenerFailedException.class,
SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class,
BatchRuntimeException.class);
}
protected void detectStreamInReader() {
@@ -592,6 +602,10 @@ public class FaultTolerantStepBuilder<I, O> extends SimpleStepBuilder<I, O> {
return skipListeners;
}
protected Set<org.springframework.batch.core.jsr.RetryListener> getJsrRetryListeners() {
return jsrRetryListeners;
}
/**
* Wrap the provided {@link #setRetryPolicy(RetryPolicy)} so that it never retries explicitly non-retryable
* exceptions.

View File

@@ -0,0 +1,244 @@
/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.jsr.configuration.xml;
import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import javax.batch.api.chunk.ItemProcessor;
import javax.batch.api.chunk.ItemReader;
import javax.batch.api.chunk.ItemWriter;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.batch.operations.BatchRuntimeException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.retry.RetryException;
import org.springframework.util.Assert;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
/**
* <p>
* Test cases around JSR-352 retry listeners.
* </p>
*
* @author Chris Schaefer
* @since 3.0
*/
public class RetryListenerTests {
private static final Log LOG = LogFactory.getLog(RetryListenerTests.class);
@Test
public void testReadRetryExhausted() throws Exception {
ApplicationContext context = new ClassPathXmlApplicationContext("org/springframework/batch/core/jsr/configuration/xml/RetryReadListenerExhausted.xml");
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
JobExecution jobExecution = jobLauncher.run(context.getBean(Job.class), new JobParameters());
List<Throwable> failureExceptions = jobExecution.getAllFailureExceptions();
assertTrue("Expected 1 failure exceptions", failureExceptions.size() == 1);
assertTrue("Failure exception must be of type RetryException", (failureExceptions.get(0) instanceof RetryException));
assertTrue("Exception cause must be of type IllegalArgumentException", (failureExceptions.get(0).getCause() instanceof IllegalArgumentException));
assertEquals(ExitStatus.FAILED, jobExecution.getExitStatus());
}
@Test
public void testReadRetryOnce() throws Exception {
ApplicationContext context = new ClassPathXmlApplicationContext("org/springframework/batch/core/jsr/configuration/xml/RetryReadListenerRetryOnce.xml");
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
JobExecution jobExecution = jobLauncher.run(context.getBean(Job.class), new JobParameters());
Collection<StepExecution> stepExecutions = jobExecution.getStepExecutions();
assertEquals(1, stepExecutions.size());
StepExecution stepExecution = stepExecutions.iterator().next();
assertEquals(1, stepExecution.getCommitCount());
assertEquals(2, stepExecution.getReadCount());
assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus());
}
@Test
public void testReadRetryExceptionInListener() throws Exception {
ApplicationContext context = new ClassPathXmlApplicationContext("org/springframework/batch/core/jsr/configuration/xml/RetryReadListenerListenerException.xml");
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
JobExecution jobExecution = jobLauncher.run(context.getBean(Job.class), new JobParameters());
List<Throwable> failureExceptions = jobExecution.getAllFailureExceptions();
assertTrue("Failure exceptions must equal one", failureExceptions.size() == 1);
assertTrue("Failure exception must be of type RetryException", (failureExceptions.get(0) instanceof RetryException));
assertTrue("Exception cause must be of type BatchRuntimeException", (failureExceptions.get(0).getCause() instanceof BatchRuntimeException));
assertEquals(ExitStatus.FAILED, jobExecution.getExitStatus());
}
public static class ExceptionThrowingRetryReadListener implements RetryReadListener {
@Override
public void onRetryReadException(Exception ex) throws Exception {
Assert.isInstanceOf(IllegalArgumentException.class, ex);
throw new IllegalStateException();
}
}
public static class TestRetryReadListener implements RetryReadListener {
@Override
public void onRetryReadException(Exception ex) throws Exception {
Assert.isInstanceOf(IllegalArgumentException.class, ex);
}
}
public static class TestRetryProcessListener implements RetryProcessListener {
@Override
public void onRetryProcessException(Object item, Exception ex) throws Exception {
Assert.isInstanceOf(String.class, item);
String currentItem = (String) item;
Assert.isTrue("three".equals(currentItem));
Assert.isInstanceOf(IllegalArgumentException.class, ex);
}
}
public static class TestRetryWriteListener implements RetryWriteListener {
@Override
public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
Assert.isTrue(items.size() == 2, "Must be two items to write");
Assert.isTrue(items.contains("three"), "Items must contain the string 'three'");
Assert.isTrue(items.contains("one"), "Items must contain the string 'one'");
Assert.isInstanceOf(IllegalArgumentException.class, ex);
}
}
public static class AlwaysFailItemReader implements ItemReader {
@Override
public void open(Serializable checkpoint) throws Exception {
}
@Override
public void close() throws Exception {
}
@Override
public Object readItem() throws Exception {
throw new IllegalArgumentException();
}
@Override
public Serializable checkpointInfo() throws Exception {
return null;
}
}
public static class FailOnceItemReader implements ItemReader {
private int cnt;
@Override
public void open(Serializable checkpoint) throws Exception {
}
@Override
public void close() throws Exception {
}
@Override
public Object readItem() throws Exception {
if(cnt == 0) {
cnt++;
return "one";
} else if (cnt == 1) {
cnt++;
throw new IllegalArgumentException();
} else if (cnt == 2) {
cnt++;
return "three";
}
return null;
}
@Override
public Serializable checkpointInfo() throws Exception {
return null;
}
}
public static class FailOnceItemProcessor implements ItemProcessor {
private int cnt;
@Override
public Object processItem(Object item) throws Exception {
if(cnt == 0) {
cnt++;
return "one";
} else if (cnt == 1) {
cnt++;
throw new IllegalArgumentException();
} else if (cnt == 2) {
cnt++;
return "three";
}
return null;
}
}
public static class FailOnceItemWriter implements ItemWriter {
private int cnt;
@Override
public void open(Serializable checkpoint) throws Exception {
}
@Override
public void close() throws Exception {
}
@Override
public void writeItems(List<Object> items) throws Exception {
for(Object item : items) {
if(cnt == 0) {
cnt++;
LOG.info("one");
} else if (cnt == 1) {
cnt++;
throw new IllegalArgumentException();
} else if (cnt == 2) {
cnt++;
LOG.info("three");
}
}
}
@Override
public Serializable checkpointInfo() throws Exception {
return null;
}
}
}

View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd">
<job id="job1" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<listeners>
<listener ref="readRetryListener" />
<listener ref="processRetryListener" />
<listener ref="writeRetryListener" />
</listeners>
<chunk retry-limit="5">
<reader ref="reader" />
<processor ref="processor" />
<writer ref="writer" />
<retryable-exception-classes>
<include class="java.lang.IllegalArgumentException" />
<exclude class="java.lang.RuntimeException" />
</retryable-exception-classes>
</chunk>
</step>
</job>
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
<property name="jobRepository" ref="jobRepository"/>
</bean>
<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"/>
</beans>

View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="RetryListenerTestBase-context.xml"/>
<bean id="readRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryReadListener"/>
<bean id="processRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryProcessListener"/>
<bean id="writeRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryWriteListener"/>
<bean id="reader" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$AlwaysFailItemReader"/>
<bean id="writer" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemWriter"/>
<bean id="processor" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemProcessor"/>
</beans>

View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="RetryListenerTestBase-context.xml"/>
<bean id="readRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$ExceptionThrowingRetryReadListener"/>
<bean id="processRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryProcessListener"/>
<bean id="writeRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryWriteListener"/>
<bean id="reader" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemReader"/>
<bean id="writer" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemWriter"/>
<bean id="processor" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemProcessor"/>
</beans>

View File

@@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="RetryListenerTestBase-context.xml"/>
<bean id="readRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryReadListener"/>
<bean id="processRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryProcessListener"/>
<bean id="writeRetryListener" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$TestRetryWriteListener"/>
<bean id="reader" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemReader"/>
<bean id="writer" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemWriter"/>
<bean id="processor" class="org.springframework.batch.core.jsr.configuration.xml.RetryListenerTests$FailOnceItemProcessor"/>
</beans>