Through this article, we are going to show you Spring batch exception handling, No job is perfect! Errors happen. You may receive bad data. You may forget one null check that causes a NullPointerException at the worst of times. How you handle errors using Spring Batch is our topic today. There are many scenarios where exceptions encountered while processing should not result in Step failure, but should be skipped instead.
This is usually a decision that totally depends on the data itself and what meaning it has. For example, our last article Spring Batch ETL Job which calculates the financial stock market prices (Open, Low, High, Close) may got a bad trade record which was formatted incorrectly or was missing necessary information, then there probably won’t be issues. Usually these bad records should be skipped and logged as well using listeners. However, banking data may not be skippable because it results in money being transferred, which needs to be completely accurate.
Today’s example will cover Spring batch skip technique, and how they can be used for handling Spring batch exceptions. We will leverage the skip technique for handling some bad stock data records in our last Spring Batch ETL Job which raises a FlatFileParseException while reading CSV file trades.csv.
1. Spring batch Skip technique
With the skip technique you may specify certain exception types and a maximum number of skipped items, and whenever one of those skippable exceptions is thrown, the batch job doesn’t fail but skip the item and goes on with the next one. Only when the maximum number of skipped items is reached, the batch job will fail. For example, Spring Batch provides the ability to skip a record when a specified Exception is throw when there is an error reading a record from your input. This section will look at how to use this technique to skip records based upon specific Exceptions. There are two pieces involved in choosing when a record is skipped.
1.1. Exception
Under what conditions to skip the record, specifically what exceptions you will ignore. When any error occurs during the reading process, Spring Batch throws an exception. In order to determine what to skip, you need to identify what exceptions to skip.
1.2. Skipped records
How many input records you will allow the step to skip before considering the step execution failed. If you skip one or two records out of a million, not a big deal; however, skipping half a million out of a million is probably wrong. It’s your responsibility to determine the threshold.
2. Creating Spring batch custom SkipPolicy
To actually skip records, all you need to do is tweak your configuration to specify the exceptions you want to skip and how many times it’s okay to do so. Say you want to skip the first 10 records that throw any ParseException.
Using the following FileVerificationSkipper.java to specify what Exceptions to skip and how many times to skip them. Spring Batch provides an interface called SkipPolicy. This interface, with its single method shouldSkip(java.lang.Throwable t, int skipCount), takes the Exception that was thrown and the number of times records have been skipped then returns true or false, indicating whether or not processing should continue with the given throwable.
From there, any implementation can determine what Exceptions they should skip and how many times. FileVerificationSkipper.java class is a SkipPolicy implementation that will not allow a FileNotFoundException to be skipped but 10 ParseException to be skipped.
FileVerificationSkipper.java:
package com.quantvalley.batch;
import java.io.FileNotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.file.FlatFileParseException;
/**
* The Class FileVerificationSkipper.
*
* @author ashraf
*/
public class FileVerificationSkipper implements SkipPolicy {
private static final Logger logger = LoggerFactory.getLogger("badRecordLogger");
@Override
public boolean shouldSkip(Throwable exception, int skipCount) throws SkipLimitExceededException {
if (exception instanceof FileNotFoundException) {
return false;
} else if (exception instanceof FlatFileParseException && skipCount <= 5) {
FlatFileParseException ffpe = (FlatFileParseException) exception;
StringBuilder errorMessage = new StringBuilder();
errorMessage.append("An error occured while processing the " + ffpe.getLineNumber()
+ " line of the file. Below was the faulty " + "input.\n");
errorMessage.append(ffpe.getInput() + "\n");
logger.error("{}", errorMessage.toString());
return true;
} else {
return false;
}
}
}
Also, we added a Logback logger to FileVerificationSkipper.java class to log the bad records, the logback.xml file contains the following configuration.
logback.xml:
<configuration>
<statusListener class="ch.qos.logback.core.status.NopStatusListener" />
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- encoders are assigned the type ch.qos.logback.classic.encoder.PatternLayoutEncoder
by default -->
<encoder>
<pattern>%d{HH:mm:ss.SSS} %-5level %class{0} - %msg%n
</pattern>
</encoder>
</appender>
<!-- Insert the current time formatted as "yyyyMMdd'T'HHmmss" under the
key "bySecond" into the logger context. This value will be available to all
subsequent configuration elements. -->
<timestamp key="bySecond" datePattern="yyyyMMdd'T'HHmmss" timeReference="contextBirth" />
<property name="LOG_FOLDER" value="logs/" />
<appender name="badRecordLoggerFile" class="ch.qos.logback.core.FileAppender">
<file>${LOG_FOLDER}/bad_records_${bySecond}.log</file>
<encoder>
<pattern>%d{HH:mm:ss.SSS} - %msg%n</pattern>
</encoder>
<param name="Append" value="false" />
</appender>
<root level="info">
<appender-ref ref="STDOUT" />
</root>
<logger name="badRecordLogger" level="error" additivity="false">
<appender-ref ref="badRecordLoggerFile" />
</logger>
</configuration>
3. Configuring and Running a Job
3.1. Enable job skipping feature
To enable the skip functionality, we will need to activate fault-tolerance on the builder, which is done with the method faultTolerant. Like explained below, the builder type switches, this time to FaultTolerantStepBuilder, and we used the skipPolicy(SkipPolicy skipPolicy) method to set FileVerificationSkipper.java class instance as a SkipPolicy implementation. A Step configuration may look like this:
@Bean
public SkipPolicy fileVerificationSkipper() {
return new FileVerificationSkipper();
}
@Bean
public Step etlStep() {
return stepBuilderFactory.get("Extract -> Transform -> Aggregate -> Load"). chunk(10000)
.reader(fxMarketEventReader()).faultTolerant().skipPolicy(fileVerificationSkipper()).processor(fxMarketEventProcessor())
.writer(stockPriceAggregator())
.build();
}
3.2. Running a Job
We added the following bad formatted records the trades.csv file which cause ParseException while the job reads the file.
trades.csv:
OMC,09:30:00.00,74.53,24,jk5kcg0oka8gvivuiv909lq5db TWTR,09:30:00.00,64.89,100,7102vs1mkukslit9smvcl6rbaj TWTR,09:30:00.00,64.89,25,875g607hfq600i1h5di6egugk3 TWTR,09:30:00.00,64.89,245,4qda2rhsr0lrqcof2cpe8f7psb TWTR,09:30:00.00,64.89,55,7dv3h155sl6dald6rra1qefuu9 USB,09:30:00.00,39.71,400,21798cg4n8nf4k0p0dgptu1pbh USB,09:30:00.00,39.71,359,s4cgm5p6hmph0jno7de76dvjlq
Finally, our job was successfully finished and the bad records were printed in the log file below.
Output:
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v1.3.3.RELEASE)
20:44:40.926 INFO StartupInfoLogger - Starting Application on HP-ProBook with PID 18310 (started by ashraf in /home/ashraf/jcg/examples/Spring Batch Exception Handling Example/spring-batch-exception-handling-example)
20:44:40.957 INFO SpringApplication - No active profile set, falling back to default profiles: default
20:44:41.018 INFO AbstractApplicationContext - Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@400cff1a: startup date [Sat May 28 20:44:41 EEST 2016]; root of context hierarchy
20:44:41.800 WARN ConfigurationClassEnhancer$BeanMethodInterceptor - @Bean method ScopeConfiguration.stepScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
20:44:41.808 WARN ConfigurationClassEnhancer$BeanMethodInterceptor - @Bean method ScopeConfiguration.jobScope is non-static and returns an object assignable to Spring's BeanFactoryPostProcessor interface. This will result in a failure to process annotations such as @Autowired, @Resource and @PostConstruct within the method's declaring @Configuration class. Add the 'static' modifier to this method to avoid these container lifecycle issues; see @Bean javadoc for complete details.
20:44:42.106 INFO EmbeddedDatabaseFactory - Starting embedded database: url='jdbc:hsqldb:mem:testdb', username='sa'
20:44:43.264 INFO ScriptUtils - Executing SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql]
20:44:43.274 INFO ScriptUtils - Executed SQL script from class path resource [org/springframework/batch/core/schema-hsqldb.sql] in 10 ms.
20:44:43.357 INFO MBeanExporter - Registering beans for JMX exposure on startup
20:44:43.374 INFO JobLauncherCommandLineRunner - Running default command line with: []
20:44:43.384 INFO JobRepositoryFactoryBean - No database type set, using meta data indicating: HSQL
20:44:43.763 INFO SimpleJobLauncher - No TaskExecutor has been set, defaulting to synchronous executor.
20:44:43.814 INFO SimpleJobLauncher$1 - Job: [FlowJob: [name=FxMarket Prices ETL Job]] launched with the following parameters: [{run.id=1}]
20:44:43.831 INFO SimpleStepHandler - Executing step: [Extract -> Transform -> Aggregate -> Load]
20:45:05.299 INFO SimpleJobLauncher$1 - Job: [FlowJob: [name=FxMarket Prices ETL Job]] completed with the following parameters: [{run.id=1}] and the following status: [COMPLETED]
20:45:05.334 INFO StartupInfoLogger - Started Application in 24.767 seconds (JVM running for 27.634)
20:45:05.353 INFO AbstractApplicationContext - Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@400cff1a: startup date [Sat May 28 20:44:41 EEST 2016]; root of context hierarchy
20:45:05.355 INFO MBeanExporter - Unregistering JMX-exposed beans on shutdown
20:45:05.356 INFO EmbeddedDatabaseFactory - Shutting down embedded database: url='jdbc:hsqldb:mem:testdb'
bad_records_20160528T190816.log:
19:08:18.459 - An error occured while processing the 7 line of the file. Below was the faulty input. OMC,09:30:00.00,74.53,24,jk5kcg0oka8gvivuiv909lq5db 19:08:18.460 - An error occured while processing the 8 line of the file. Below was the faulty input. TWTR,09:30:00.00,64.89,100,7102vs1mkukslit9smvcl6rbaj 19:08:18.460 - An error occured while processing the 9 line of the file. Below was the faulty input. TWTR,09:30:00.00,64.89,25,875g607hfq600i1h5di6egugk3 19:08:18.460 - An error occured while processing the 10 line of the file. Below was the faulty input. TWTR,09:30:00.00,64.89,245,4qda2rhsr0lrqcof2cpe8f7psb 19:08:18.460 - An error occured while processing the 11 line of the file. Below was the faulty input. TWTR,09:30:00.00,64.89,55,7dv3h155sl6dald6rra1qefuu9 19:08:18.460 - An error occured while processing the 12 line of the file. Below was the faulty input. USB,09:30:00.00,39.71,400,21798cg4n8nf4k0p0dgptu1pbh 19:08:18.460 - An error occured while processing the 13 line of the file. Below was the faulty input. USB,09:30:00.00,39.71,359,s4cgm5p6hmph0jno7de76dvjlq
4. Download the Source Code
This was an example to show how to handle Spring Batch Exception.
You can download the full source code of this example here: SpringBatchExceptionHandlingExampleCode.zip
Thank you for the excellent tutorial about the spring batch and the skip policy. I tried the example and it works fine. One issue I have when I’m handling xml files. I’m using the StaxEventItemReader and setup the policy to skip all the time, however that does not work properly. The batch runs and when there is bad record it stops there as completed. is there any way to let the batch move to the next record.
Why method documentation says that when the skip policy is explicit, the skip limit is ignored?
So when use the skip policy the skip count will return -1 every time. Here I have to implement the skip count manually.
I think that this skipCount <= 5 will not work.
Nice article buddy. Helped me a lot…
Thank you