Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update docs with expected behaviour in regards to skippable exceptions

Browse Source 
Before this commit, the expected behaviour when a skippbale exception
occurs in a fault tolerant chunk-oriented step was not documented in
details.

This commit update the docs and adds a sample for each case (when a
skippable exception occurs during read, process and write).

Resolves BATCH-2541
This commit is contained in:
Mahmoud Ben Hassine
2019-11-25 11:24:21 +01:00
parent 2b4a87b1b8
commit dc3fee597d
9 changed files with 475 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java
View File

@@ -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<I, O> extends SimpleStepBuilder<I, O> {
}
/**
* 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

5
spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipPolicy.java
View File

@@ -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&lt;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

3
spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-3.0.xsd
View File

@@ -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.
]]>
</xsd:documentation>
</xsd:annotation>

4
spring-batch-docs/asciidoc/step.adoc
View File

@@ -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.

67
spring-batch-samples/README.md
View File

@@ -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

104
spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringProcessSample.java Normal file
View File

@@ -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<Integer> itemReader() {
return new ListItemReader<Integer>(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<Integer, Integer> 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<Integer> 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")
.<Integer, Integer>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();
}
}

104
spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringReadSample.java Normal file
View File

@@ -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<Integer> itemReader() {
return new ListItemReader<Integer>(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<Integer, Integer> itemProcessor() {
return item -> {
System.out.println("processing item = " + item);
return item;
};
}
@Bean
public ItemWriter<Integer> 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")
.<Integer, Integer>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();
}
}

104
spring-batch-samples/src/main/java/org/springframework/batch/sample/skip/SkippableExceptionDuringWriteSample.java Normal file
View File

@@ -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<Integer> itemReader() {
return new ListItemReader<Integer>(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<Integer, Integer> itemProcessor() {
return item -> {
System.out.println("processing item = " + item);
return item;
};
}
@Bean
public ItemWriter<Integer> 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")
.<Integer, Integer>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();
}
}

83
spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java
View File

@@ -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