diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java new file mode 100644 index 000000000..c62e1591e --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/SkipListener.java @@ -0,0 +1,49 @@ +/* + * 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; + +/** + * Interface for listener to skipped items. Callbacks will be called by + * {@link Step} implementations at the appropriate time in the step lifecycle. + * Callbacks must always be made (where relevant) in a transaction that is still + * valid: i.e. possibly on a read error, but not on a write error. + * + * @author Dave Syer + * + */ +public interface SkipListener extends StepListener { + + /** + * Callback for a failure on read that is legal, so is not going to be + * re-thrown. + * + * @param t + */ + void onSkipInRead(Throwable t); + + /** + * This item failed on write with the given exception, and a skip was called + * for. The callback is deferred until a new transaction is available. This + * callback might occur more than once for the same item, but only once in + * successful transaction. + * + * + * @param item the failed item + * @param t the cause of the failure + */ + void onSkipInWrite(Object item, Throwable t); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java index f98d2e73d..7285e3e71 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java @@ -50,23 +50,24 @@ public class SimpleJob extends AbstractJob { private CompositeExecutionJobListener listener = new CompositeExecutionJobListener(); /** - * Public setter for injecting {@link JobExecutionListener}s. They will all be given - * the {@link JobExecutionListener} callbacks at the appropriate point in the job. + * Public setter for injecting {@link JobExecutionListener}s. They will all + * be given the listener callbacks at the appropriate point in the job. * * @param listeners the listeners to set. */ - public void setJobListeners(JobExecutionListener[] listeners) { + public void setJobExecutionListeners(JobExecutionListener[] listeners) { for (int i = 0; i < listeners.length; i++) { this.listener.register(listeners[i]); } } /** - * Register a single listener for the {@link JobExecutionListener} callbacks. + * Register a single listener for the {@link JobExecutionListener} + * callbacks. * * @param listener a {@link JobExecutionListener} */ - public void registerListener(JobExecutionListener listener) { + public void registerJobExecutionListener(JobExecutionListener listener) { this.listener.register(listener); } @@ -75,7 +76,8 @@ public class SimpleJob extends AbstractJob { * {@link Step}. * * @see org.springframework.batch.core.Job#execute(org.springframework.batch.core.JobExecution) - * @throws StartLimitExceededException if start limit of one of the steps was exceeded + * @throws StartLimitExceededException if start limit of one of the steps + * was exceeded */ public void execute(JobExecution execution) throws JobExecutionException { @@ -110,8 +112,8 @@ public class SimpleJob extends AbstractJob { StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step); - boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 - && !lastStepExecution.getExitStatus().equals(ExitStatus.FINISHED)) ? true : false; + boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution + .getExitStatus().equals(ExitStatus.FINISHED)) ? true : false; if (isRestart && lastStepExecution != null) { currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java new file mode 100644 index 000000000..415d15776 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java @@ -0,0 +1,88 @@ +/* + * 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.listener; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + +import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.repeat.ExitStatus; + +/** + * @author Dave Syer + * + */ +public class CompositeSkipListener implements SkipListener { + + private List listeners = new ArrayList(); + + /** + * Public setter for the listeners. + * + * @param listeners + */ + public void setListeners(SkipListener[] listeners) { + this.listeners = Arrays.asList(listeners); + } + + /** + * Register additional listener. + * + * @param stepExecutionListener + */ + public void register(SkipListener listener) { + if (!listeners.contains(listener)) { + listeners.add(listener); + } + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.domain.StepListener#onError(java.lang.Throwable) + */ + public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { + ExitStatus status = null; + for (Iterator iterator = listeners.iterator(); iterator.hasNext();) { + StepExecutionListener listener = (StepExecutionListener) iterator.next(); + ExitStatus close = listener.onErrorInStep(stepExecution, e); + status = status!=null ? status.and(close): close; + } + return status; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable) + */ + public void onSkipInRead(Throwable t) { + for (Iterator iterator = listeners.iterator(); iterator.hasNext();) { + SkipListener listener = (SkipListener) iterator.next(); + listener.onSkipInRead(t); + } + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable) + */ + public void onSkipInWrite(Object item, Throwable t) { + for (Iterator iterator = listeners.iterator(); iterator.hasNext();) { + SkipListener listener = (SkipListener) iterator.next(); + listener.onSkipInWrite(item, t); + } + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java index 01c60929f..624831dc8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java @@ -15,12 +15,13 @@ */ package org.springframework.batch.core.listener; -import org.springframework.batch.core.StepListener; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemReadListener; import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.StepListener; import org.springframework.batch.item.ItemStream; import org.springframework.batch.repeat.ExitStatus; @@ -29,7 +30,7 @@ import org.springframework.batch.repeat.ExitStatus; * */ public class MulticasterBatchListener implements StepExecutionListener, ChunkListener, ItemReadListener, - ItemWriteListener { + ItemWriteListener, SkipListener { private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener(); @@ -39,6 +40,8 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener(); + private CompositeSkipListener skipListener = new CompositeSkipListener(); + /** * Initialise the listener instance. */ @@ -77,6 +80,9 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis if (listener instanceof ItemWriteListener) { this.itemWriteListener.register((ItemWriteListener) listener); } + if (listener instanceof SkipListener) { + this.skipListener.register((SkipListener) listener); + } } /** @@ -170,4 +176,21 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis itemWriteListener.onWriteError(ex, item); } + /** + * @param t + * @see org.springframework.batch.core.listener.CompositeSkipListener#onSkipInRead(java.lang.Throwable) + */ + public void onSkipInRead(Throwable t) { + skipListener.onSkipInRead(t); + } + + /** + * @param item + * @param t + * @see org.springframework.batch.core.listener.CompositeSkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable) + */ + public void onSkipInWrite(Object item, Throwable t) { + skipListener.onSkipInWrite(item, t); + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java index b8d261077..6fd9e967c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java @@ -18,11 +18,12 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; import java.util.List; -import org.springframework.batch.core.StepListener; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemReadListener; import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.StepListener; import org.springframework.batch.core.listener.CompositeChunkListener; import org.springframework.batch.core.listener.CompositeItemReadListener; import org.springframework.batch.core.listener.CompositeItemWriteListener; @@ -169,4 +170,19 @@ class BatchListenerFactoryHelper { return (StepExecutionListener[]) list.toArray(new StepExecutionListener[list.size()]); } + /** + * @param listeners + * @return + */ + public SkipListener[] getSkipListeners(StepListener[] listeners) { + List list = new ArrayList(); + for (int i = 0; i < listeners.length; i++) { + StepListener listener = listeners[i]; + if (listener instanceof SkipListener) { + list.add(listener); + } + } + return (SkipListener[]) list.toArray(new SkipListener[list.size()]); + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java index 03265caed..386d93e65 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java @@ -168,9 +168,9 @@ public class ItemOrientedStep extends AbstractStep { * * @param listeners an array of listener objects of known types. */ - public void setStepListeners(StepExecutionListener[] listeners) { + public void setStepExecutionListeners(StepExecutionListener[] listeners) { for (int i = 0; i < listeners.length; i++) { - registerStepListener(listeners[i]); + registerStepExecutionListener(listeners[i]); } } @@ -180,7 +180,7 @@ public class ItemOrientedStep extends AbstractStep { * * @param listener a {@link StepExecutionListener} */ - public void registerStepListener(StepExecutionListener listener) { + public void registerStepExecutionListener(StepExecutionListener listener) { this.listener.register(listener); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java index 74a7b3508..9bd7fcd63 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java @@ -15,12 +15,14 @@ */ package org.springframework.batch.core.step.item; -import java.util.HashSet; -import java.util.Set; +import java.util.HashMap; +import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.listener.CompositeSkipListener; import org.springframework.batch.core.step.skip.ItemSkipPolicy; import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; import org.springframework.batch.item.ItemKeyGenerator; @@ -44,13 +46,39 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - private ItemKeyGenerator itemKeyGenerator = new ItemKeyGenerator() { + private ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() { public Object getKey(Object item) { return item; } }; - private Set skippedItems = new HashSet(); + private ItemKeyGenerator itemKeyGenerator = defaultItemKeyGenerator; + + private CompositeSkipListener listener = new CompositeSkipListener(); + + private Map skippedExceptions = new HashMap(); + + /** + * Register some {@link SkipListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage if a skip occurs. + * + * @param listeners + */ + public void setSkipListeners(SkipListener[] listeners) { + for (int i = 0; i < listeners.length; i++) { + registerSkipListener(listeners[i]); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a skip + * process. + * + * @param listener a {@link SkipListener} + */ + public void registerSkipListener(SkipListener listener) { + this.listener.register(listener); + } /** * Public setter for the {@link ItemKeyGenerator}. Defaults to just return @@ -60,9 +88,13 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { * reader does any buffering the key generator might need to take care to * only use data that do not change on write). * - * @param itemKeyGenerator the itemKeyGenerator to set + * @param itemKeyGenerator the {@link ItemKeyGenerator} to set. If null + * resets to default value. */ public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { + if (itemKeyGenerator == null) { + itemKeyGenerator = defaultItemKeyGenerator; + } this.itemKeyGenerator = itemKeyGenerator; } @@ -95,9 +127,14 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { try { Object item = doRead(); - while (item != null && skippedItems.contains(itemKeyGenerator.getKey(item))) { - logger.debug("Skipping item on input: " + item); + Object key = itemKeyGenerator.getKey(item); + while (item != null && skippedExceptions.containsKey(key)) { + logger.debug("Skipping item on input, previously failed on output; key=[" + key + "]"); + if (listener != null) { + listener.onSkipInWrite(item, (Throwable) skippedExceptions.get(key)); + } item = doRead(); + key = itemKeyGenerator.getKey(item); } return item; @@ -106,6 +143,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { // increment skip count and try again contribution.incrementSkipCount(); + if (listener != null) { + listener.onSkipInRead(e); + } + logger.debug("Skipping failed input", e); } else { // re-throw only when the skip policy runs out of patience @@ -135,7 +176,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { catch (Exception e) { if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { contribution.incrementSkipCount(); - skippedItems.add(key); + // don't call the listener here - the transaction is going to + // roll back + skippedExceptions.put(key, e); + logger.debug("Added item to skip list; key=" + key); } // always re-throw exception on write throw e; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java index a31293de0..6a9a9bd8d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java @@ -108,13 +108,13 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean { step.registerStream((ItemStream) itemReader); } if (itemReader instanceof StepExecutionListener) { - step.registerStepListener((StepExecutionListener) itemReader); + step.registerStepExecutionListener((StepExecutionListener) itemReader); } if (itemWriter instanceof ItemStream) { step.registerStream((ItemStream) itemWriter); } if (itemWriter instanceof StepExecutionListener) { - step.registerStepListener((StepExecutionListener) itemWriter); + step.registerStepExecutionListener((StepExecutionListener) itemWriter); } BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); @@ -128,7 +128,7 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean { setItemReader(itemReader); setItemWriter(itemWriter); - step.setStepListeners(stepListeners); + step.setStepExecutionListeners(stepListeners); step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter)); step.setChunkOperations(chunkOperations); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 45ee32414..db621b3dc 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,10 +15,9 @@ */ package org.springframework.batch.core.step.item; -import org.springframework.batch.core.StepListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecutionListener; -import org.springframework.batch.core.listener.MulticasterBatchListener; +import org.springframework.batch.core.StepListener; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; @@ -49,8 +48,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { private StepListener[] listeners = new StepListener[0]; - private MulticasterBatchListener listener = new MulticasterBatchListener(); - private TaskExecutor taskExecutor; private ItemHandler itemHandler; @@ -89,6 +86,14 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { public void setListeners(StepListener[] listeners) { this.listeners = listeners; } + + /** + * Protected getter for the {@link StepListener}s. + * @return the listeners + */ + protected StepListener[] getListeners() { + return listeners; + } /** * Protected getter for the step operations to make them available in @@ -151,16 +156,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { step.setStreams(streams); - for (int i = 0; i < listeners.length; i++) { - StepListener listener = listeners[i]; - if (listener instanceof StepExecutionListener) { - step.registerStepListener((StepExecutionListener) listener); - } - else { - this.listener.register(listener); - } - } - ItemReader itemReader = getItemReader(); ItemWriter itemWriter = getItemWriter(); @@ -171,13 +166,13 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { step.registerStream((ItemStream) itemReader); } if (itemReader instanceof StepExecutionListener) { - step.registerStepListener((StepExecutionListener) itemReader); + step.registerStepExecutionListener((StepExecutionListener) itemReader); } if (itemWriter instanceof ItemStream) { step.registerStream((ItemStream) itemWriter); } if (itemWriter instanceof StepExecutionListener) { - step.registerStepListener((StepExecutionListener) itemWriter); + step.registerStepExecutionListener((StepExecutionListener) itemWriter); } BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); @@ -197,7 +192,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean { setItemReader(itemReader); setItemWriter(itemWriter); - step.setStepListeners(stepListeners); + step.setStepExecutionListeners(stepListeners); stepOperations = new RepeatTemplate(); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index ab75b99a8..7afe64e45 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -4,6 +4,7 @@ import java.util.Arrays; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; +import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler; /** @@ -29,6 +30,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { private Class[] fatalExceptionClasses = new Class[] { Error.class }; + private ItemKeyGenerator itemKeyGenerator; + /** * Public setter for a limit that determines skip policy. If this value is * positive then an exception in chunk processing will cause the item to be @@ -62,6 +65,17 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { this.fatalExceptionClasses = fatalExceptionClasses; } + /** + * Public setter for the {@link ItemKeyGenerator}. This is used to identify + * failed items so they can be skipped if encountered again, generally in + * another transaction. + * + * @param itemKeyGenerator the {@link ItemKeyGenerator} to set. + */ + public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { + this.itemKeyGenerator = itemKeyGenerator; + } + /** * Uses the {@link #skipLimit} value to configure item handler and and * exception handler. @@ -72,6 +86,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { ItemSkipPolicyItemHandler itemHandler = new ItemSkipPolicyItemHandler(getItemReader(), getItemWriter()); if (skipLimit > 0) { + /* * If there is a skip limit (not the default) then we are prepared * to absorb exceptions at the step level because the failed items @@ -85,6 +100,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses); setExceptionHandler(exceptionHandler); getStepOperations().setExceptionHandler(getExceptionHandler()); + itemHandler.setItemKeyGenerator(itemKeyGenerator); + + BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); + itemHandler.setSkipListeners(helper.getSkipListeners(getListeners())); + } else { // This is the default in ItemOrientedStep anyway... diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java index 18f1d6375..954560a7f 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java @@ -173,7 +173,7 @@ public class SimpleJobTests extends TestCase { } public void testRunNormallyWithListener() throws Exception { - job.setJobListeners(new JobExecutionListenerSupport[] { new JobExecutionListenerSupport() { + job.setJobExecutionListeners(new JobExecutionListenerSupport[] { new JobExecutionListenerSupport() { public void beforeJob(JobExecution jobExecution) { list.add("before"); } @@ -251,7 +251,7 @@ public class SimpleJobTests extends TestCase { } public void testFailedWithListener() throws Exception { - job.setJobListeners(new JobExecutionListenerSupport[] { new JobExecutionListenerSupport() { + job.setJobExecutionListeners(new JobExecutionListenerSupport[] { new JobExecutionListenerSupport() { public void onError(JobExecution jobExecution, Throwable t) { list.add(t); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeJobExecutionListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeJobExecutionListenerTests.java index b48a34ae9..55a434ebe 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeJobExecutionListenerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeJobExecutionListenerTests.java @@ -57,7 +57,7 @@ public class CompositeJobExecutionListenerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.core.listener.CompositeExecutionJobListener#registerListener(org.springframework.batch.core.JobExecutionListener)}. + * {@link org.springframework.batch.core.listener.CompositeExecutionJobListener#registerJobExecutionListener(org.springframework.batch.core.JobExecutionListener)}. */ public void testSetListener() { listener.register(new JobExecutionListenerSupport() { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeStepExecutionListenerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeStepExecutionListenerTests.java index efa6ed491..c8ca65737 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeStepExecutionListenerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/CompositeStepExecutionListenerTests.java @@ -59,7 +59,7 @@ public class CompositeStepExecutionListenerTests extends TestCase { /** * Test method for - * {@link org.springframework.batch.core.listener.CompositeStepExecutionListener#registerListener(org.springframework.batch.core.StepExecutionListener)}. + * {@link org.springframework.batch.core.listener.CompositeStepExecutionListener#registerJobExecutionListener(org.springframework.batch.core.StepExecutionListener)}. */ public void testSetListener() { listener.register(new StepExecutionListenerSupport() { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java index 444da9def..09b471174 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepTests.java @@ -231,7 +231,7 @@ public class ItemOrientedStepTests extends TestCase { }; itemOrientedStep.setItemHandler(new SimpleItemHandler(itemReader, itemWriter)); - itemOrientedStep.registerStepListener(new StepExecutionListenerSupport() { + itemOrientedStep.registerStepExecutionListener(new StepExecutionListenerSupport() { public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { return ExitStatus.FAILED.addExitDescription("FOO"); } @@ -394,7 +394,7 @@ public class ItemOrientedStepTests extends TestCase { } public void testDirectlyInjectedListener() throws Exception { - itemOrientedStep.registerStepListener(new StepExecutionListenerSupport() { + itemOrientedStep.registerStepExecutionListener(new StepExecutionListenerSupport() { public void beforeStep(StepExecution stepExecution) { list.add("foo"); } @@ -421,7 +421,7 @@ public class ItemOrientedStepTests extends TestCase { } }; itemOrientedStep.setStreams(new ItemStream[] { reader }); - itemOrientedStep.registerStepListener(reader); + itemOrientedStep.registerStepExecutionListener(reader); StepExecution stepExecution = new StepExecution(itemOrientedStep, new JobExecution(jobInstance)); itemOrientedStep.execute(stepExecution); assertEquals(1, list.size()); @@ -431,7 +431,7 @@ public class ItemOrientedStepTests extends TestCase { final ExitStatus customStatus = new ExitStatus(false, "custom code"); - itemOrientedStep.setStepListeners(new StepExecutionListener[] { new StepExecutionListenerSupport() { + itemOrientedStep.setStepExecutionListeners(new StepExecutionListener[] { new StepExecutionListenerSupport() { public ExitStatus afterStep(StepExecution stepExecution) { list.add("afterStepCalled"); return customStatus; @@ -452,7 +452,7 @@ public class ItemOrientedStepTests extends TestCase { } public void testDirectlyInjectedListenerOnError() throws Exception { - itemOrientedStep.registerStepListener(new StepExecutionListenerSupport() { + itemOrientedStep.registerStepExecutionListener(new StepExecutionListenerSupport() { public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) { list.add(e); return null; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java index 5adf68b47..7f6372f02 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java @@ -17,9 +17,8 @@ package org.springframework.batch.item; /** - * Mixin interface for {@link ItemReader} implementations if they can - * distinguish a new item from one that has been processed before and failed, - * e.g. by examining a message flag. + * Strategy interface to distinguish a new item from one that has been processed + * before and failed, e.g. by examining a message flag. * * @author Dave Syer * diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java index 89477adc7..ed9dd9836 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java @@ -18,6 +18,7 @@ package org.springframework.batch.retry.callback; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.FailedItemIdentifier; import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; @@ -54,6 +55,8 @@ public class ItemReaderRetryCallback implements RetryCallback { private ItemKeyGenerator keyGenerator; + private FailedItemIdentifier failedItemIdentifier; + private ItemKeyGenerator defaultKeyGenerator = new ItemKeyGenerator() { public Object getKey(Object item) { return item; @@ -94,6 +97,17 @@ public class ItemReaderRetryCallback implements RetryCallback { this.keyGenerator = keyGenerator; } + /** + * Public setter for the {@link FailedItemIdentifier}. If it is not + * injected but the reader or writer implement {@link FailedItemIdentifier}, + * one of those will be used instead (preferring the reader to the writer if + * both would be appropriate). + * @param failedItemIdentifier the {@link FailedItemIdentifier} to set + */ + public void setFailedItemIdentifier(FailedItemIdentifier failedItemIdentifier) { + this.failedItemIdentifier = failedItemIdentifier; + } + public Object doWithRetry(RetryContext context) throws Throwable { // This requires a collaboration with the RetryPolicy... if (!context.isExhaustedOnly()) { @@ -152,6 +166,27 @@ public class ItemReaderRetryCallback implements RetryCallback { return defaultKeyGenerator; } + /** + * Accessor for the {@link FailedItemIdentifier}. If the handler is null + * but the {@link ItemReader} or {@link ItemWriter} is an instance of + * {@link FailedItemIdentifier}, then it will be returned instead. If none + * of those strategies works returns null. + * + * @return the {@link FailedItemIdentifier}. + */ + public FailedItemIdentifier getFailedItemIdentifier() { + if (failedItemIdentifier != null) { + return failedItemIdentifier; + } + if (reader instanceof FailedItemIdentifier) { + return (FailedItemIdentifier) reader; + } + if (writer instanceof FailedItemIdentifier) { + return (FailedItemIdentifier) writer; + } + return null; + } + /** * Accessor for the {@link ItemRecoverer}. If the handler is null but the * {@link ItemReader} is an instance of {@link ItemRecoverer}, then it will diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java index 5cc3ef6be..8e1c1127d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java @@ -20,7 +20,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.FailedItemIdentifier; import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.repeat.support.RepeatSynchronizationManager; import org.springframework.batch.retry.RetryCallback; @@ -99,7 +98,8 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy { * Create a new context for the execution of the callback, which must be an * instance of {@link ItemReaderRetryCallback}. * - * @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback, RetryContext) + * @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback, + * RetryContext) * * @throws IllegalStateException if the callback is not of the required * type. @@ -139,18 +139,18 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy { // The delegate context... private RetryContext delegateContext; - private ItemReader reader; - private ItemRecoverer recoverer; private ItemKeyGenerator keyGenerator; + private FailedItemIdentifier failedItemIdentifier; + public ItemReaderRetryContext(ItemReaderRetryCallback callback, RetryContext parent) { super(parent); item = callback.next(this); - this.reader = callback.getReader(); this.recoverer = callback.getRecoverer(); this.keyGenerator = callback.getKeyGenerator(); + this.failedItemIdentifier = callback.getFailedItemIdentifier(); } public boolean canRetry(RetryContext context) { @@ -162,7 +162,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy { } public RetryContext open(RetryCallback callback, RetryContext parent) { - if (hasFailed(reader, keyGenerator, item)) { + if (hasFailed(failedItemIdentifier, keyGenerator, item)) { this.delegateContext = retryContextCache.get(keyGenerator.getKey(item)); } if (this.delegateContext == null) { @@ -220,19 +220,18 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy { * a messaging environment where the item is a message, it can be inspected * to see if it has been delivered before.
* - * The default implementation of this method checks the provider for a mixin - * interface {@link FailedItemIdentifier}. If the interface is present the - * decision is delegated to the provider. Otherwise we just check the cache - * for the item key. + * The default implementation of this method checks for a non-null + * {@link FailedItemIdentifier}. Otherwise we just check the cache for the + * item key. * - * @param reader + * @param failedItemIdentifier * @param keyGenerator * @param item * @return */ - protected boolean hasFailed(ItemReader reader, ItemKeyGenerator keyGenerator, Object item) { - if (reader instanceof FailedItemIdentifier) { - return ((FailedItemIdentifier) reader).hasFailed(item); + protected boolean hasFailed(FailedItemIdentifier failedItemIdentifier, ItemKeyGenerator keyGenerator, Object item) { + if (failedItemIdentifier != null) { + return failedItemIdentifier.hasFailed(item); } return retryContextCache.containsKey(keyGenerator.getKey(item)); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java index 7f74632a6..7af6064ce 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java @@ -26,7 +26,6 @@ import junit.framework.TestCase; import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.FailedItemIdentifier; import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.context.RepeatContextSupport; @@ -42,7 +41,7 @@ public class ItemReaderRetryPolicyTests extends TestCase { private ItemReaderRetryPolicy policy = new ItemReaderRetryPolicy(); - private ItemReader reader; + private ListItemReaderRecoverer reader; private int count = 0; @@ -303,7 +302,7 @@ public class ItemReaderRetryPolicyTests extends TestCase { MapRetryContextCache cache = new MapRetryContextCache(); policy.setRetryContextCache(cache); cache.put("foo", new RetryContextSupport(null)); - assertTrue(policy.hasFailed(reader, keyGenerator , "foo")); + assertTrue(policy.hasFailed(null, keyGenerator , "foo")); } private static class MockFailedItemProvider extends ListItemReader implements ItemKeyGenerator, FailedItemIdentifier {