diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index da8223f62..be7fad738 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2006-2018 the original author or authors. + * Copyright 2006-2019 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. @@ -344,7 +344,10 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { } /** - * Explicitly request certain exceptions (and subclasses) to be skipped. + * Explicitly request certain exceptions (and subclasses) to be skipped. These + * exceptions (and their subclasses) might be thrown during any phase of the chunk + * processing (read, process, write) but separate counts are made of skips on + * read, process and write inside the step execution. * * @param type the exception type. * @return this for fluent chaining diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicy.java index 8c81a5a5e..e2309a99a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicy.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicy.java @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2019 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. @@ -20,6 +20,7 @@ package org.springframework.batch.core.step.skip; * * @author Lucas Ward * @author Dave Syer + * @author Mahmoud Ben Hassine */ public interface SkipPolicy { @@ -31,7 +32,7 @@ public interface SkipPolicy { * {@code skipCount<0}. Implementations should avoid throwing any * undeclared exceptions. * - * @param t exception encountered while reading + * @param t exception encountered while processing * @param skipCount currently running count of skips * @return true if processing should continue, false otherwise. * @throws SkipLimitExceededException if a limit is breached diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-3.0.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-3.0.xsd index 64dd6f6f7..cdc96d288 100644 --- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-3.0.xsd +++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-3.0.xsd @@ -841,6 +841,9 @@ are declared as included take precedence over the same value if it is also excluded. Exceptions that are already marked as no-rollback are automatically skippable (but it doesn't hurt to add them again here). + Exceptions (and their subclasses) that are declared might be thrown + during any phase of the chunk processing (read, process, write) but separate counts + are made of skips on read, process and write inside the step execution. ]]> diff --git a/spring-batch-docs/asciidoc/step.adoc b/spring-batch-docs/asciidoc/step.adoc index af648dacd..d87360a5b 100644 --- a/spring-batch-docs/asciidoc/step.adoc +++ b/spring-batch-docs/asciidoc/step.adoc @@ -588,7 +588,9 @@ public Step step1() { In the preceding example, a `FlatFileItemReader` is used. If, at any point, a `FlatFileParseException` is thrown, the item is skipped and counted against the total -skip limit of 10. Separate counts are made of skips on read, process, and write inside +skip limit of 10. Exceptions (and their subclasses) that are declared might be thrown +during any phase of the chunk processing (read, process, write) but separate counts +are made of skips on read, process, and write inside the step execution, but the limit applies across all skips. Once the skip limit is reached, the next exception found causes the step to fail. In other words, the eleventh skip triggers the exception, not the tenth. diff --git a/spring-batch-samples/README.md b/spring-batch-samples/README.md index 42291a8ad..db8c29524 100644 --- a/spring-batch-samples/README.md +++ b/spring-batch-samples/README.md @@ -44,6 +44,7 @@ IO Sample Job | | | | [Restart Sample](#restart-sample) | | | X | | | | | | | | [Retry Sample](#retry-sample) | | X | | | | | | | | | [Skip Sample](#skip-sample) | X | | | | | | | | | | +[Chunk Scanning Sample](#chunk-scanning-sample) | X | | | | | | | | | | [Trade Job](#trade-job) | | | | | | X | | | | | The IO Sample Job has a number of special instances that show different IO features using the same job configuration but with different readers and writers: @@ -812,6 +813,72 @@ The format for the transaction attribute specification is given in the Spring Core documentation (e.g. see the Javadocs for [TransactionAttributeEditor](https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/transaction/interceptor/TransactionAttributeEditor.html)). +### Chunk Scanning Sample + +In a fault tolerant chunk-oriented step, when a skippable exception is thrown during +item writing, the item writer (which receives a chunk of items) does not +know which item caused the issue. Hence, it will "scan" the chunk item by item +and only the faulty item will be skipped. Technically, the commit-interval will +be re-set to 1 and each item will re-processed/re-written in its own transaction. + +The `org.springframework.batch.sample.skip.SkippableExceptionDuringWriteSample` sample +illustrates this behaviour: + +* It reads numbers from 1 to 6 in chunks of 3 items, so two chunks are created: [1, 2 ,3] and [4, 5, 6] +* It processes each item by printing it to the standard output and returning it as is. +* It writes items to the standard output and throws an exception for item 5 + +The expected behaviour when an exception occurs at item 5 is that the second chunk [4, 5, 6] is +scanned item by item. Transactions of items 4 and 6 will be successfully committed, while +the one of item 5 will be rolled back. Here is the output of the sample with some useful comments: + +``` +1. reading item = 1 +2. reading item = 2 +3. reading item = 3 +4. processing item = 1 +5. processing item = 2 +6. processing item = 3 +7. About to write chunk: [1, 2, 3] +8. writing item = 1 +9. writing item = 2 +10. writing item = 3 +11. reading item = 4 +12. reading item = 5 +13. reading item = 6 +14. processing item = 4 +15. processing item = 5 +16. processing item = 6 +17. About to write chunk: [4, 5, 6] +18. writing item = 4 +19. Throwing exception on item 5 +20. processing item = 4 +21. About to write chunk: [4] +22. writing item = 4 +23. processing item = 5 +24. About to write chunk: [5] +25. Throwing exception on item 5 +26. processing item = 6 +27. About to write chunk: [6] +28. writing item = 6 +29. reading item = null +``` + +* Lines 1-10: The first chunk is processed without any issue +* Lines 11-17: The second chunk is read and processed correctly and is about to be written +* Line 18: Item 4 is successfully written +* Line 19: An exception is thrown when attempting to write item 5, the transaction is rolled back and chunk scanning is about to start +* Lines 20-22: Item 4 is re-processed/re-written successfully in its own transaction +* Lines 23-25: Item 5 is re-processed/re-written with an exception. Its transaction is rolled back and is skipped +* Lines 26-28: Item 6 is re-processed/re-written successfully in its own transaction +* Line 29: Attempting to read the next chunk, but the reader returns `null`: +the datasource is exhausted and the step ends here + +Similar examples show the expected behaviour when a skippable exception is thrown +during reading and processing can be found in +`org.springframework.batch.sample.skip.SkippableExceptionDuringReadSample` +and `org.springframework.batch.sample.skip.SkippableExceptionDuringProcessSample`. + ### Tasklet Job The goal is to show the simplest use of the batch framework with a diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringProcessSample.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringProcessSample.java new file mode 100644 index 000000000..5c4486d86 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringProcessSample.java @@ -0,0 +1,104 @@ +/* + * Copyright 2019 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.sample.skip; + +import java.util.Arrays; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +public class SkippableExceptionDuringProcessSample { + + private final JobBuilderFactory jobBuilderFactory; + + private final StepBuilderFactory stepBuilderFactory; + + public SkippableExceptionDuringProcessSample(JobBuilderFactory jobBuilderFactory, + StepBuilderFactory stepBuilderFactory) { + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + } + + @Bean + public ItemReader itemReader() { + return new ListItemReader(Arrays.asList(1, 2, 3, 4, 5, 6)) { + @Override + public Integer read() { + Integer item = super.read(); + System.out.println("reading item = " + item); + return item; + } + }; + } + + @Bean + public ItemProcessor itemProcessor() { + return item -> { + if (item.equals(5)) { + System.out.println("Throwing exception on item " + item); + throw new IllegalArgumentException("Unable to process 5"); + } + System.out.println("processing item = " + item); + return item; + }; + } + + @Bean + public ItemWriter itemWriter() { + return items -> { + System.out.println("About to write chunk: " + items); + for (Integer item : items) { + System.out.println("writing item = " + item); + } + }; + } + + @Bean + public Step step() { + return this.stepBuilderFactory.get("step") + .chunk(3) + .reader(itemReader()) + .processor(itemProcessor()) + .writer(itemWriter()) + .faultTolerant() + .skip(IllegalArgumentException.class) + .skipLimit(3) + .build(); + } + + @Bean + public Job job() { + return this.jobBuilderFactory.get("job") + .start(step()) + .build(); + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringReadSample.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringReadSample.java new file mode 100644 index 000000000..2cf29eba2 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringReadSample.java @@ -0,0 +1,104 @@ +/* + * Copyright 2019 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.sample.skip; + +import java.util.Arrays; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +public class SkippableExceptionDuringReadSample { + + private final JobBuilderFactory jobBuilderFactory; + + private final StepBuilderFactory stepBuilderFactory; + + public SkippableExceptionDuringReadSample(JobBuilderFactory jobBuilderFactory, + StepBuilderFactory stepBuilderFactory) { + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + } + + @Bean + public ItemReader itemReader() { + return new ListItemReader(Arrays.asList(1, 2, 3, 4, 5, 6)) { + @Override + public Integer read() { + Integer item = super.read(); + System.out.println("reading item = " + item); + if (item != null && item.equals(5)) { + System.out.println("Throwing exception on item " + item); + throw new IllegalArgumentException("Sorry, no 5 here!"); + } + return item; + } + }; + } + + @Bean + public ItemProcessor itemProcessor() { + return item -> { + System.out.println("processing item = " + item); + return item; + }; + } + + @Bean + public ItemWriter itemWriter() { + return items -> { + System.out.println("About to write chunk: " + items); + for (Integer item : items) { + System.out.println("writing item = " + item); + } + }; + } + + @Bean + public Step step() { + return this.stepBuilderFactory.get("step") + .chunk(3) + .reader(itemReader()) + .processor(itemProcessor()) + .writer(itemWriter()) + .faultTolerant() + .skip(IllegalArgumentException.class) + .skipLimit(3) + .build(); + } + + @Bean + public Job job() { + return this.jobBuilderFactory.get("job") + .start(step()) + .build(); + } + +} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringWriteSample.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringWriteSample.java new file mode 100644 index 000000000..5631a83aa --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringWriteSample.java @@ -0,0 +1,104 @@ +/* + * Copyright 2019 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.sample.skip; + +import java.util.Arrays; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Mahmoud Ben Hassine + */ +@Configuration +@EnableBatchProcessing +public class SkippableExceptionDuringWriteSample { + + private final JobBuilderFactory jobBuilderFactory; + + private final StepBuilderFactory stepBuilderFactory; + + public SkippableExceptionDuringWriteSample(JobBuilderFactory jobBuilderFactory, + StepBuilderFactory stepBuilderFactory) { + this.jobBuilderFactory = jobBuilderFactory; + this.stepBuilderFactory = stepBuilderFactory; + } + + @Bean + public ItemReader itemReader() { + return new ListItemReader(Arrays.asList(1, 2, 3, 4, 5, 6)) { + @Override + public Integer read() { + Integer item = super.read(); + System.out.println("reading item = " + item); + return item; + } + }; + } + + @Bean + public ItemProcessor itemProcessor() { + return item -> { + System.out.println("processing item = " + item); + return item; + }; + } + + @Bean + public ItemWriter itemWriter() { + return items -> { + System.out.println("About to write chunk: " + items); + for (Integer item : items) { + if (item.equals(5)) { + System.out.println("Throwing exception on item " + item); + throw new IllegalArgumentException("Sorry, no 5 here!"); + } + System.out.println("writing item = " + item); + } + }; + } + + @Bean + public Step step() { + return this.stepBuilderFactory.get("step") + .chunk(3) + .reader(itemReader()) + .processor(itemProcessor()) + .writer(itemWriter()) + .faultTolerant() + .skip(IllegalArgumentException.class) + .skipLimit(3) + .build(); + } + + @Bean + public Job job() { + return this.jobBuilderFactory.get("job") + .start(step()) + .build(); + } + +} diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java index fd6d00608..587afc289 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2008-2014 the original author or authors. + * Copyright 2008-2019 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. @@ -27,10 +27,15 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersInvalidException; +import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.UnexpectedJobExecutionException; import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.launch.JobOperator; import org.springframework.batch.core.launch.JobParametersNotFoundException; import org.springframework.batch.core.launch.NoSuchJobException; @@ -39,6 +44,11 @@ import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteExcep import org.springframework.batch.core.repository.JobRestartException; import org.springframework.batch.sample.common.SkipCheckingListener; import org.springframework.batch.sample.domain.trade.internal.TradeWriter; +import org.springframework.batch.sample.skip.SkippableExceptionDuringProcessSample; +import org.springframework.batch.sample.skip.SkippableExceptionDuringReadSample; +import org.springframework.batch.sample.skip.SkippableExceptionDuringWriteSample; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.test.jdbc.JdbcTestUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -54,6 +64,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * * @author Robert Kasanicky * @author Dan Garrette + * @author Mahmoud Ben Hassine */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = { "/skipSample-job-launcher-context.xml" }) @@ -179,6 +190,76 @@ public class SkipSampleFunctionalTests { assertTrue(!execution1.getJobId().equals(execution2.getJobId())); } + /* + * When a skippable exception is thrown during reading, the item is skipped + * from the chunk and is not passed to the chunk processor (So it will not be + * processed nor written). + */ + @Test + public void testSkippableExceptionDuringRead() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(SkippableExceptionDuringReadSample.class); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + + // when + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + // then + assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(1, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + } + + /* + * When a skippable exception is thrown during processing, items will re-processed + * one by one and the faulty item will be skipped from the chunk (it will not be + * passed to the writer). + */ + @Test + public void testSkippableExceptionDuringProcess() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(SkippableExceptionDuringProcessSample.class); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + + // when + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + // then + assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getProcessSkipCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + } + + /* + * When a skippable exception is thrown during writing, the item writer (which receives a chunk of items) + * does not know which item caused the issue. Hence, it will "scan" the chunk item by item + * and only the faulty item will be skipped (technically, the commit-interval will be re-set to 1 + * and each item will re-processed/re-written in its own transaction). + */ + @Test + public void testSkippableExceptionDuringWrite() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(SkippableExceptionDuringWriteSample.class); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + + // when + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + // then + assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(0, stepExecution.getProcessSkipCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + } + private void validateLaunchWithSkips(JobExecution jobExecution) { // Step1: 9 input records, 1 skipped in read, 1 skipped in write => // 7 written to output