diff --git a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs index 38984f7ff..b6a8e1b4d 100644 --- a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs +++ b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs @@ -1,4 +1,7 @@ -#Thu Apr 08 11:36:17 BST 2010 +#Thu Apr 08 16:49:54 BST 2010 +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml=\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/FileToMessagesJobIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/step/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/tasklet/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index 234caa683..7d7eb74bd 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -20,9 +20,9 @@ src/test/resources/org/springframework/batch/integration/item/MessagingGatewayIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/partition/VanillaIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml - src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/file/FileToMessagesJobIntegrationTests-context.xml src/test/resources/org/springframework/batch/integration/step/StepGatewayIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index 91838b0a9..a6afdff2f 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java @@ -16,7 +16,11 @@ package org.springframework.batch.integration.chunk; +import java.util.ArrayList; +import java.util.Collection; import java.util.List; +import java.util.Queue; +import java.util.concurrent.LinkedBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -32,7 +36,8 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.integration.gateway.MessagingGateway; import org.springframework.util.Assert; -public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream { +public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, + ItemStream, StepContributionSource { private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class); @@ -57,7 +62,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * result from the remote workers. This is a multiplier on the receive * timeout set separately on the gateway. The ideal value is a compromise * between allowing slow workers time to finish, and responsiveness if there - * is a dead worker. Defaults to 40. + * is a dead worker. Defaults to 40. * * @param maxWaitTimeouts the maximum number of wait timeouts */ @@ -119,6 +124,11 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo stepExecution.setStatus(BatchStatus.FAILED); return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage()); } + finally { + for (StepContribution contribution : getStepContributions()) { + stepExecution.apply(contribution); + } + } if (timedOut) { stepExecution.setStatus(BatchStatus.FAILED); throw new ItemStreamException("Timed out waiting for back log at end of step"); @@ -145,17 +155,32 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo executionContext.putLong(ACTUAL, localState.actual); } + public Collection getStepContributions() { + return localState.pollContributions(); + } + /** * Wait until all the results that are in the pipeline come back to the * reply channel. * * @return true if successfully received a result, false if timed out */ - private boolean waitForResults() { + private boolean waitForResults() throws AsynchronousFailureException { int count = 0; int maxCount = maxWaitTimeouts; + Throwable failure = null; while (localState.getExpecting() > 0 && count++ < maxCount) { - getNextResult(); + try { + getNextResult(); + } + catch (Throwable t) { + logger.error("Detected error in remote result. Trying to recover " + localState.getExpecting() + + " outstanding results before completing.", t); + failure = t; + } + } + if (failure != null) { + throw wrapIfNecessary(failure); } return count < maxCount; } @@ -178,7 +203,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id [" + jobInstanceId + "] should have been [" + localState.getJobId() + "]."); localState.actual++; - // TODO: apply the skip count + localState.pushContribution(payload.getStepContribution()); if (!payload.isSuccessful()) { throw new AsynchronousFailureException("Failure or interrupt detected in handler: " + payload.getMessage()); @@ -186,6 +211,22 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } } + /** + * Re-throws the original throwable if it is unchecked, wraps checked + * exceptions into {@link AsynchronousFailureException}. + */ + private static AsynchronousFailureException wrapIfNecessary(Throwable throwable) { + if (throwable instanceof Error) { + throw (Error) throwable; + } + else if (throwable instanceof AsynchronousFailureException) { + return (AsynchronousFailureException) throwable; + } + else { + return new AsynchronousFailureException("Exception in remote process", throwable); + } + } + private static class LocalState { private long actual; @@ -193,10 +234,26 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private StepExecution stepExecution; + private Queue contributions = new LinkedBlockingQueue(); + public long getExpecting() { return expected - actual; } + public Collection pollContributions() { + Collection set = new ArrayList(); + StepContribution item = contributions.poll(); + while (item != null) { + set.add(item); + item = contributions.poll(); + } + return set; + } + + public void pushContribution(StepContribution stepContribution) { + contributions.add(stepContribution); + } + public StepContribution createStepContribution() { return stepExecution.createStepContribution(); } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java index 6401088a5..4c638a7e5 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java @@ -18,9 +18,15 @@ package org.springframework.batch.integration.chunk; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.step.item.Chunk; import org.springframework.batch.core.step.item.ChunkProcessor; +import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor; +import org.springframework.batch.core.step.skip.NonSkippableReadException; +import org.springframework.batch.core.step.skip.SkipLimitExceededException; +import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.retry.RetryException; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; @@ -63,16 +69,66 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, Initializ StepContribution stepContribution = chunkRequest.getStepContribution(); try { - chunkProcessor.process(stepContribution, new Chunk(chunkRequest.getItems())); + process(chunkRequest, stepContribution); } catch (Exception e) { logger.debug("Failed chunk", e); return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": " + e.getMessage()); + } catch (Throwable e) { + // The handler might throw an Error or other non-exception + logger.debug("Failed chunk with non-exception", e); + return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": " + + e.getMessage()); } logger.debug("Completed chunk handling with " + stepContribution); return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution); } + + /** + * @param chunkRequest the current request + * @param stepContribution the step contribution to update + * @throws Exception if there is a fatal exception + */ + private void process(ChunkRequest chunkRequest, StepContribution stepContribution) throws Exception { + + Chunk chunk = new Chunk(chunkRequest.getItems()); + + if (chunkProcessor instanceof FaultTolerantChunkProcessor) { + + boolean processed = false; + + while (!processed) { + try { + chunkProcessor.process(stepContribution, chunk); + processed = true; + } + catch (SkipLimitExceededException e) { + throw e; + } + catch (NonSkippableReadException e) { + throw e; + } + catch (SkipListenerFailedException e) { + throw e; + } + catch (RetryException e) { + throw e; + } + catch (JobInterruptedException e) { + throw e; + } + catch (Exception e) { + // try again... + } + } + + } + else { + chunkProcessor.process(stepContribution, chunk); + } + + } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java new file mode 100644 index 000000000..9b92020a8 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java @@ -0,0 +1,216 @@ +/* + * Copyright 2006-2010 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.integration.chunk; + +import java.lang.reflect.Field; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.step.item.Chunk; +import org.springframework.batch.core.step.item.ChunkOrientedTasklet; +import org.springframework.batch.core.step.item.ChunkProcessor; +import org.springframework.batch.core.step.item.SimpleChunkProcessor; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.PassThroughItemProcessor; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * Convenient factory bean for a chunk handler that also converts an existing + * chunk-oriented step into a remote chunk master. The idea is to lift the + * existing chunk processor out of a step that works locally, and replace it + * with a chunk writer that is already configured to write chunks into a message + * channel. The existing step hands its business chunk processing responsibility + * over to the handler produced by the factory, which then needs to be set up as + * a remote worker on the other end of the channel the chunks are being sent to. + * + * @author Dave Syer + * + */ +public class RemoteChunkHandlerFactoryBean implements FactoryBean> { + + private static Log logger = LogFactory.getLog(RemoteChunkHandlerFactoryBean.class); + + private TaskletStep step; + + private ItemWriter chunkWriter; + + private StepContributionSource stepContributionSource; + + /** + * @param step the step to set + */ + public void setStep(TaskletStep step) { + this.step = step; + } + + /** + * @param chunkWriter the chunk writer to set + */ + public void setChunkWriter(ItemWriter chunkWriter) { + this.chunkWriter = chunkWriter; + } + + /** + * @param stepContributionSource the step contribution source to set + * (defaults to the chunk writer) + */ + public void setStepContributionSource(StepContributionSource stepContributionSource) { + this.stepContributionSource = stepContributionSource; + } + + public Class getObjectType() { + return ChunkHandler.class; + } + + public boolean isSingleton() { + return true; + } + + public ChunkHandler getObject() throws Exception { + + if (stepContributionSource == null) { + Assert.state(chunkWriter instanceof StepContributionSource, + "The chunk writer must be a StepContributionSource or else the source must be provided explicitly"); + stepContributionSource = (StepContributionSource) chunkWriter; + } + + Assert.state(step instanceof TaskletStep, "Step [" + step.getName() + "] must be a TaskletStep"); + logger.debug("Converting TaskletStep with name=" + step.getName()); + + Tasklet tasklet = getTasklet((TaskletStep) step); + Assert.state(tasklet instanceof ChunkOrientedTasklet, "Tasklet must be ChunkOrientedTasklet in step=" + + step.getName()); + + ChunkProcessor chunkProcessor = getChunkProcessor((ChunkOrientedTasklet) tasklet); + Assert.state(chunkProcessor != null, "ChunkProcessor must be accessible in Tasklet in step=" + step.getName()); + + ItemWriter itemWriter = getItemWriter(chunkProcessor); + Assert.state(!(itemWriter instanceof ChunkMessageChannelItemWriter), "Cannot adapt step [" + step.getName() + + "] because it already has a remote chunk writer. Use a local writer in the step."); + + replaceChunkProcessor((ChunkOrientedTasklet) tasklet, chunkWriter, stepContributionSource); + if (chunkWriter instanceof StepExecutionListener) { + step.registerStepExecutionListener((StepExecutionListener) chunkWriter); + } + + ChunkProcessorChunkHandler handler = new ChunkProcessorChunkHandler(); + handler.setChunkProcessor(chunkProcessor); + // TODO: create step context for the processor in case it has scope="step" dependencies + handler.afterPropertiesSet(); + + return handler; + + } + + /** + * @param tasklet + * @param chunkWriter + */ + private void replaceChunkProcessor(ChunkOrientedTasklet tasklet, ItemWriter chunkWriter, + final StepContributionSource stepContributionSource) { + setField(tasklet, "chunkProcessor", new SimpleChunkProcessor(new PassThroughItemProcessor(), + chunkWriter) { + @Override + protected void write(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { + doWrite(outputs.getItems()); + // Do not update the step contribution until the chunks are + // actually processed + updateStepContribution(contribution, stepContributionSource); + } + }); + } + + /** + * @param contribution + * @param chunkWriter + */ + private void updateStepContribution(StepContribution contribution, StepContributionSource stepContributionSource) { + for (StepContribution result : stepContributionSource.getStepContributions()) { + contribution.incrementFilterCount(result.getFilterCount()); + contribution.incrementWriteCount(result.getWriteCount()); + for (int i = 0; i < result.getProcessSkipCount(); i++) { + contribution.incrementProcessSkipCount(); + } + for (int i = 0; i < result.getWriteSkipCount(); i++) { + contribution.incrementWriteSkipCount(); + } + contribution.setExitStatus(contribution.getExitStatus().and(result.getExitStatus())); + } + } + + /** + * @param chunkProcessor + * @return + */ + @SuppressWarnings("unchecked") + private ItemWriter getItemWriter(ChunkProcessor chunkProcessor) { + return (ItemWriter) getField(chunkProcessor, "itemWriter"); + } + + /** + * @param tasklet + * @return + */ + @SuppressWarnings("unchecked") + private ChunkProcessor getChunkProcessor(ChunkOrientedTasklet tasklet) { + return (ChunkProcessor) getField(tasklet, "chunkProcessor"); + } + + /** + * @param bean + * @return + */ + private Tasklet getTasklet(TaskletStep bean) { + return (Tasklet) getField(bean, "tasklet"); + } + + private static Object getField(Object target, String name) { + Assert.notNull(target, "Target object must not be null"); + Field field = ReflectionUtils.findField(target.getClass(), name); + if (field == null) { + logger.debug("Could not find field [" + name + "] on target [" + target + "]"); + return null; + } + + if (logger.isDebugEnabled()) { + logger.debug("Getting field [" + name + "] from target [" + target + "]"); + } + ReflectionUtils.makeAccessible(field); + return ReflectionUtils.getField(field, target); + } + + private static void setField(Object target, String name, Object value) { + Assert.notNull(target, "Target object must not be null"); + Field field = ReflectionUtils.findField(target.getClass(), name); + if (field == null) { + throw new IllegalStateException("Could not find field [" + name + "] on target [" + target + "]"); + } + + if (logger.isDebugEnabled()) { + logger.debug("Getting field [" + name + "] from target [" + target + "]"); + } + ReflectionUtils.makeAccessible(field); + ReflectionUtils.setField(field, target, value); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/StepContributionSource.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/StepContributionSource.java new file mode 100644 index 000000000..35838bd06 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/StepContributionSource.java @@ -0,0 +1,31 @@ +/* + * Copyright 2006-2010 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.integration.chunk; + +import java.util.Collection; + +import org.springframework.batch.core.StepContribution; + +/** + * @author Dave Syer + * + */ +public interface StepContributionSource { + + Collection getStepContributions(); + +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java index ed4080671..d583f65c5 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java @@ -200,7 +200,7 @@ public class ChunkMessageItemWriterIntegrationTests { public void testEarlyCompletionSignalledInHandler() throws Exception { factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils - .commaDelimitedListToStringArray("1,bad,3,4,5,6")))); + .commaDelimitedListToStringArray("1,fail,3,4,5,6")))); factory.setCommitInterval(2); Step step = (Step) factory.getObject(); @@ -210,7 +210,7 @@ public class ChunkMessageItemWriterIntegrationTests { assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode()); String message = stepExecution.getExitStatus().getExitDescription(); - assertTrue("Message does not contain 'bad': " + message, message.contains("bad")); + assertTrue("Message does not contain 'fail': " + message, message.contains("fail")); waitForResults(2, 10); @@ -262,7 +262,7 @@ public class ChunkMessageItemWriterIntegrationTests { public void testFailureInStepListener() throws Exception { factory.setItemReader(new ListItemReader(Arrays.asList(StringUtils - .commaDelimitedListToStringArray("wait,bad,3,4,5,6")))); + .commaDelimitedListToStringArray("wait,fail,3,4,5,6")))); Step step = (Step) factory.getObject(); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests.java new file mode 100644 index 000000000..314ed7901 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests.java @@ -0,0 +1,75 @@ +package org.springframework.batch.integration.chunk; + +import static org.junit.Assert.assertEquals; + +import java.util.Collections; + +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameter; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RemoteChunkFaultTolerantStepIntegrationTests { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private Job job; + + @Test + public void testFailedStep() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", + new JobParameter("unsupported")))); + assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(9, stepExecution.getReadCount()); + // In principle the write count could be more than 2 and less than 9... + assertEquals(7, stepExecution.getWriteCount()); + } + + @Test + @Ignore + public void testFailedStepOnError() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", + new JobParameter("error")))); + assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(9, stepExecution.getReadCount()); + // In principle the write count could be more than 2 and less than 9... + assertEquals(7, stepExecution.getWriteCount()); + } + + @Test + public void testSunnyDayFaultTolerant() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", + new JobParameter("3")))); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(9, stepExecution.getReadCount()); + assertEquals(9, stepExecution.getWriteCount()); + } + + @Test + public void testSkipsInWriter() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParametersBuilder().addString("item.three", "fail") + .addLong("run.id", 1L).toJobParameters()); + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(9, stepExecution.getReadCount()); + assertEquals(8, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + } +} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests.java similarity index 57% rename from spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java rename to spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests.java index b538e91d6..0086e1672 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests.java @@ -2,11 +2,14 @@ package org.springframework.batch.integration.chunk; import static org.junit.Assert.assertEquals; +import java.util.Collections; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameter; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.launch.JobLauncher; @@ -16,7 +19,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class ChunkStepIntegrationTests { +public class RemoteChunkStepIntegrationTests { @Autowired private JobLauncher jobLauncher; @@ -25,12 +28,24 @@ public class ChunkStepIntegrationTests { private Job job; @Test - public void testOpenWithNoState() throws Exception { - JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + public void testSunnyDaySimpleStep() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", + new JobParameter("3")))); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); assertEquals(9, stepExecution.getReadCount()); assertEquals(9, stepExecution.getWriteCount()); } + @Test + public void testFailedStep() throws Exception { + JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", + new JobParameter("fail")))); + assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(9, stepExecution.getReadCount()); + // In principle the write count could be more than 2 and less than 9... + assertEquals(7, stepExecution.getWriteCount()); + } + } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java index 50883237a..1da3c4255 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemReader.java @@ -18,7 +18,7 @@ public class TestItemReader implements ItemReader { /** * Counts the number of chunks processed in the handler. */ - public volatile static int count = 0; + public volatile int count = 0; /** * Item that causes failure in handler. diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemWriter.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemWriter.java index c2e16a474..e4519ca6e 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemWriter.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemWriter.java @@ -20,7 +20,17 @@ public class TestItemWriter implements ItemWriter { /** * Item that causes failure in handler. */ - public final static String FAIL_ON = "bad"; + public final static String FAIL_ON = "fail"; + + /** + * Item that causes error in handler. + */ + public final static String UNSUPPORTED_ON = "unsupported"; + + /** + * Item that causes error in handler. + */ + public final static String ERROR_ON = "error"; /** * Item that causes handler to wait to simulate delayed processing. @@ -49,6 +59,14 @@ public class TestItemWriter implements ItemWriter { throw new IllegalStateException("Planned failure on: " + FAIL_ON); } + if (item.equals(UNSUPPORTED_ON)) { + throw new UnsupportedOperationException("Planned failure on: " + UNSUPPORTED_ON); + } + + if (item.equals(ERROR_ON)) { + throw new Error("Planned failure on: " + ERROR_ON); + } + } } diff --git a/spring-batch-integration/src/test/resources/log4j.properties b/spring-batch-integration/src/test/resources/log4j.properties index ed4cdc90c..63126f81c 100644 --- a/spring-batch-integration/src/test/resources/log4j.properties +++ b/spring-batch-integration/src/test/resources/log4j.properties @@ -7,5 +7,5 @@ log4j.appender.stdout.layout.ConversionPattern=%d %5p %t [%c] - <%m>%n log4j.category.org.springframework.context=INFO log4j.category.org.springframework.beans=INFO log4j.category.org.springframework.batch.core=DEBUG -log4j.category.org.springframework.integration=DEBUG +log4j.category.org.springframework.batch.integration=DEBUG log4j.category.org.springframework.transaction=INFO \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml new file mode 100644 index 000000000..0360e426c --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml @@ -0,0 +1,72 @@ + + + + + + + + + + + + + + + + + + + 1 + 2 + #{jobParameters['item.three']} + 4 + 5 + 6 + 7 + 8 + 9 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml similarity index 64% rename from spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml rename to spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml index 50229ea58..30781e6eb 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml @@ -2,10 +2,11 @@ + http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"> @@ -14,13 +15,13 @@ - + 1 2 - 3 + #{jobParameters['item.three']} 4 5 6 @@ -31,44 +32,37 @@ - - + + + + - + - - - + + + - - + - + - - - - - - - - - - - + + + - + - + - + - + \ No newline at end of file