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 274d10db0..fb9097303 100644 --- a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs +++ b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs @@ -1,8 +1,9 @@ -#Fri Apr 09 09:28:44 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=\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/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 -eclipse.preferences.version=1 +#Fri Apr 09 18:01:26 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/RemoteChunkFaultTolerantStepJmsIntegrationTests-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 +eclipse.preferences.version=1 diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index 7d7eb74bd..71e07211f 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -23,6 +23,9 @@ 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 + src/test/resources/jms-context.xml + src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml + src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml index dd0870e87..c97f8718d 100644 --- a/spring-batch-integration/pom.xml +++ b/spring-batch-integration/pom.xml @@ -1,5 +1,6 @@ - + 4.0.0 spring-batch-integration Spring Batch Integration @@ -10,36 +11,6 @@ 1.0.0.BUILD-SNAPSHOT ../spring-batch-admin-parent - - - - true - - - - - maven-surefire-plugin - - true - - - - - - - test - - - - maven-surefire-plugin - - false - - - - - - org.springframework.batch @@ -58,6 +29,12 @@ geronimo-jms_1.1_spec true + + org.apache.activemq + activemq-core + 5.3.0 + test + junit junit @@ -93,6 +70,12 @@ ${spring.integration.version} compile + + org.springframework.integration + spring-integration-jms + ${spring.integration.version} + test + org.springframework.integration spring-integration-adapter @@ -135,6 +118,21 @@ com.springsource.bundlor com.springsource.bundlor.maven + + org.apache.maven.plugins + maven-clean-plugin + + + + activemq-data + + ** + * + + + + + diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java index 9eddc4e39..cb15bc38f 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java @@ -18,6 +18,6 @@ package org.springframework.batch.integration.chunk; public interface ChunkHandler { - ChunkResponse handleChunk(ChunkRequest chunk); + ChunkResponse handleChunk(ChunkRequest chunk) throws Exception; } \ No newline at end of file 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 a6afdff2f..107d2591f 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 @@ -195,9 +195,12 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo * @throws IllegalStateException if the result contains the wrong job * instance id (maybe we are sharing a channel and we shouldn't be) */ - private void getNextResult() { + private void getNextResult() throws AsynchronousFailureException { ChunkResponse payload = (ChunkResponse) messagingGateway.receive(); if (payload != null) { + if (logger.isDebugEnabled()) { + logger.debug("Found result: "+payload); + } Long jobInstanceId = payload.getJobId(); Assert.state(jobInstanceId != null, "Message did not contain job instance id."); Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id [" 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 4c638a7e5..f9fb868ff 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 @@ -22,7 +22,6 @@ 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; @@ -63,23 +62,39 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, Initializ * @see ChunkHandler#handleChunk(ChunkRequest) */ @ServiceActivator - public ChunkResponse handleChunk(ChunkRequest chunkRequest) { + public ChunkResponse handleChunk(ChunkRequest chunkRequest) throws Exception { logger.debug("Handling chunk: " + chunkRequest); StepContribution stepContribution = chunkRequest.getStepContribution(); + Throwable failure = null; try { process(chunkRequest, stepContribution); } + catch (SkipLimitExceededException e) { + failure = e; + } + catch (NonSkippableReadException e) { + failure = e; + } + catch (SkipListenerFailedException e) { + failure = e; + } + catch (RetryException e) { + failure = e; + } + catch (JobInterruptedException e) { + failure = e; + } 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()); + // try again... + throw e; + } + + if (failure != null) { + logger.debug("Failed chunk", failure); + return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, failure.getClass().getName() + + ": " + failure.getMessage()); } logger.debug("Completed chunk handling with " + stepContribution); @@ -95,40 +110,7 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, Initializ 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); - } + 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 index 9b92020a8..0fd16a9d5 100644 --- 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 @@ -25,6 +25,7 @@ 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.FaultTolerantChunkProcessor; import org.springframework.batch.core.step.item.SimpleChunkProcessor; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.core.step.tasklet.TaskletStep; @@ -114,6 +115,7 @@ public class RemoteChunkHandlerFactoryBean implements FactoryBean handler = new ChunkProcessorChunkHandler(); + setNonBuffering(chunkProcessor); handler.setChunkProcessor(chunkProcessor); // TODO: create step context for the processor in case it has scope="step" dependencies handler.afterPropertiesSet(); @@ -122,6 +124,15 @@ public class RemoteChunkHandlerFactoryBean implements FactoryBean chunkProcessor) { + if (chunkProcessor instanceof FaultTolerantChunkProcessor) { + ((FaultTolerantChunkProcessor)chunkProcessor).setBuffering(false); + } + } + /** * @param tasklet * @param chunkWriter diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java index 8dcc20dd2..908d51ea2 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java @@ -17,7 +17,7 @@ public class ChunkProcessorChunkHandlerTests { protected int count = 0; @Test - public void testVanillaHandleChunk() { + public void testVanillaHandleChunk() throws Exception { handler.setChunkProcessor(new ChunkProcessor() { public void process(StepContribution contribution, Chunk chunk) throws Exception { count += chunk.size(); 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 index 314ed7901..77792203e 100644 --- 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 @@ -4,7 +4,6 @@ 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; @@ -41,7 +40,6 @@ public class RemoteChunkFaultTolerantStepIntegrationTests { } @Test - @Ignore public void testFailedStepOnError() throws Exception { JobExecution jobExecution = jobLauncher.run(job, new JobParameters(Collections.singletonMap("item.three", new JobParameter("error")))); @@ -69,7 +67,8 @@ public class RemoteChunkFaultTolerantStepIntegrationTests { assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); assertEquals(9, stepExecution.getReadCount()); - assertEquals(8, stepExecution.getWriteCount()); - assertEquals(1, stepExecution.getWriteSkipCount()); + assertEquals(7, stepExecution.getWriteCount()); + // The whole chunk gets skipped... + assertEquals(2, stepExecution.getWriteSkipCount()); } } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests.java new file mode 100644 index 000000000..4940e75d3 --- /dev/null +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests.java @@ -0,0 +1,73 @@ +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.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 RemoteChunkFaultTolerantStepJmsIntegrationTests { + + @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 + 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(7, stepExecution.getWriteCount()); + assertEquals(2, stepExecution.getWriteSkipCount()); + } +} diff --git a/spring-batch-integration/src/test/resources/jms-context.xml b/spring-batch-integration/src/test/resources/jms-context.xml new file mode 100644 index 000000000..2d86ef606 --- /dev/null +++ b/spring-batch-integration/src/test/resources/jms-context.xml @@ -0,0 +1,41 @@ + + + + + + + + + + + + + vm://localhost + + + + + + + + + vm://localhost + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/log4j.properties b/spring-batch-integration/src/test/resources/log4j.properties index 63126f81c..20474e2a4 100644 --- a/spring-batch-integration/src/test/resources/log4j.properties +++ b/spring-batch-integration/src/test/resources/log4j.properties @@ -6,6 +6,7 @@ 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.retry=DEBUG log4j.category.org.springframework.batch.core=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 index 0360e426c..b9f0184cb 100644 --- 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 @@ -1,19 +1,22 @@ - - + + - + @@ -43,20 +46,19 @@ - - + + - - - - - + + + + - + - + @@ -68,5 +70,23 @@ - + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml new file mode 100644 index 000000000..bcefb0278 --- /dev/null +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml @@ -0,0 +1,83 @@ + + + + + + + + + + + + + + + + + + + + + 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/RemoteChunkStepIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml index 429bee27f..7eb5d9d60 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml @@ -1,8 +1,11 @@ @@ -43,19 +46,14 @@ - - - - + - - - + + - - +