Logging invalid records

Time needed to complete  ~60 minutes

Prerequisite for this tutorial is Configure Spring Batch Admin to use MySQL

1.Introduction

In post Validate input data the batch job stop immediately after finding invalid order record, in real life  scenario if 15 out of 1 million record that needs to be processed are invalid and there is an invalid record in the first 100 records, the batch job will failed, no work is done, the question is can the batch job isolate these records and continues processing and later developers will deal with the invalid records. Spring batch has mechanism  to do that, and in this post will be explained, invalid data will be logged  to the file and for developers there will be hint in what stage(read ,process ,write) the error popped up to know where to look.

2.What is used in this tutorial:

1- Maven 3

2- Jdk 1.7

3- Tomcat 7.0.55

4- Eclipse Luna 4.4

5- Spring Core 3.2.9

6- Spring Batch 2.2.7

7- Spring Batch Admin Manager 1.3.0

8- MySQL 5 Database

3.Project Structure

Picture 1: Project structure

4. Configuration and Other files

- The database configuration where the data will be written is in file database.xml path src/main/resources/META-INF/spring/batch/dbConfig/database.xml, this is not the same as the database used by spring batch admin to store job repository data. First create the database named company_db and in the database.xml change db_username and db_password to suite company_db username and password.

Listing 1: database.xml

<beans xmlns="http://www.springframework.org/schema/beans"  
      xmlns:jdbc="http://www.springframework.org/schema/jdbc"   
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
      xsi:schemaLocation="http://www.springframework.org/schema/beans   
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
           http://www.springframework.org/schema/jdbc   
           http://www.springframework.org/schema/jdbc/spring-jdbc-3.2.xsd">  
   <!-- connect to database -->  
      <bean id="dataSource2"  
           class="org.springframework.jdbc.datasource.DriverManagerDataSource">  
           <property name="driverClassName" value="com.mysql.jdbc.Driver" />  
           <property name="url" value="jdbc:mysql://localhost:3306/company_db" />  
           <property name="username" value="db_username" /> 
           <property name="password" value="db_password" />
      </bean>  
      <bean id="transactionManager"  
           class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />  
 </beans>  

- The job definition that will read,process and write is in src/main/resources/META-INF/spring/batch/jobs/job-orderInvalidLogging.xml , Listing 2. All exceptions will be skipped and data that caused the exception will be logged. It is interesting to see how changing skip-limit value will change the behavior of the job execution. For example changing skip-limit =2 will cause job to throw:

 org.springframework.batch.core.step.skip.SkipLimitExceededException: Skip limit of '2' exceeded

Listing 2: Job definition in job-orderInvalidateLogging.xml

<beans xmlns="http://www.springframework.org/schema/beans"  
      xmlns:batch="http://www.springframework.org/schema/batch"   
      xmlns:task="http://www.springframework.org/schema/task"  
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
      xsi:schemaLocation="http://www.springframework.org/schema/batch  
           http://www.springframework.org/schema/batch/spring-batch-2.2.xsd  
           http://www.springframework.org/schema/beans   
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd">  
      <!-- Database configuration -->  
      <import resource="../dbConfig/database.xml" />  
      <bean id="orderSkipListener" class="com.web.listener.OrderRecordSkipHandler" />  
      <bean id="validator"  
           class="org.springframework.validation.beanvalidation.LocalValidatorFactoryBean" />  
      <!-- Batch job -->  
      <batch:job id="orderLoggingJob">  
           <batch:step id="step1">  
                <batch:tasklet>  
                     <batch:chunk reader="cvsFileItemReader" processor="orderProcessor"  
                          writer="mysqlItemWriter" commit-interval="3" skip-limit="20">  
                          <batch:skippable-exception-classes>  
                               <batch:include class="java.lang.Exception" />  
                          </batch:skippable-exception-classes>  
                     </batch:chunk>  
                     <batch:listeners>  
                          <batch:listener ref="orderSkipListener" />  
                     </batch:listeners>  
                </batch:tasklet>  
           </batch:step>  
      </batch:job>  
      <bean id="orderProcessor" class="com.web.processor.OrderProcessor">  
           <property name="validator" ref="validator" />  
      </bean>  
      <bean id="cvsFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader"  
           scope="step">  
           <property name="linesToSkip" value="1" />  
           <!-- Read order csv file -->  
           <property name="resource" value="#{jobParameters['inputDataFile']}" />  
           <property name="lineMapper">  
                <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">  
                     <!-- Spliting fileds -->  
                     <property name="lineTokenizer">  
                          <bean  
                               class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer" />  
                     </property>  
                     <property name="fieldSetMapper">  
                          <bean class="com.web.mapper.OrderMapper" />  
                     </property>  
                </bean>  
           </property>  
      </bean>  
      <bean id="mysqlItemWriter"  
           class="org.springframework.batch.item.database.JdbcBatchItemWriter">  
           <property name="dataSource" ref="dataSource2" />  
           <property name="sql">  
                <value>  
       <![CDATA[      
            insert into tel_order(code,service_type,quantity,offer,bill_type,order_date)  
         values (:code, :serviceType, :quantity, :offer, :billType,:orderDate)  
       ]]>  
                </value>  
           </property>  
           <!-- map java object property to sql parameter -->  
           <property name="itemSqlParameterSourceProvider">  
                <bean  
                     class="org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider" />  
           </property>  
      </bean>  
 </beans>  

- The Order class using hibernate validator annotations, listing 3, and there is a custom @AccptedValues validator listing 4 and Listing 5.

Listing 3: Order class

 package com.web.model;  
   
 import java.io.Serializable;  
 import java.util.Date;  
   
 import javax.validation.constraints.Max;  
 import javax.validation.constraints.Min;  
 import javax.validation.constraints.NotNull;  
 import javax.validation.constraints.Past;  
 import javax.validation.constraints.Size;  
   
 import org.springframework.format.annotation.DateTimeFormat;  
   
 import com.web.validator.AcceptedValues;  
   
   
 public class Order implements Serializable {  
   
      private static final long serialVersionUID = 1853355651832301663L;  
   
      @Size(min = 5, max = 5, message = "Invalid code value")  
      private String code;  
        
      @AcceptedValues(acceptValues = { "fix", "mobile", "bundle", "internet" }, message = "Invalid service type")  
      private String serviceType;  
   
      @Min(value = 1, message = "Quantity less than 1")  
      @Max(value = 9, message = "Quantity more than 9")  
      private int quantity;  
        
      @AcceptedValues(acceptValues = { "small", "medium", "large", "fiber32" }, message = "Invalid offer")  
      private String offer;  
   
      @AcceptedValues(acceptValues = { "prepaid", "postpaid_mail","postpaid_email" }, message = "Invalid bill type")  
      private String billType;  
   
      @NotNull(message = "Date is required")  
      @Past(message = "Order date should be in the past")  
      @DateTimeFormat(pattern = "dd-MM-yyyy")  
      private Date orderDate; 

    // Setter, getters, toString, etc omitted    

Listing 4: Custom annotation type

 package com.web.validator;  
   
 import java.lang.annotation.Documented;  
 import java.lang.annotation.ElementType;  
 import java.lang.annotation.Retention;  
 import java.lang.annotation.RetentionPolicy;  
 import java.lang.annotation.Target;  
   
 import javax.validation.Constraint;  
 import javax.validation.Payload;  
   
 @Documented  
 @Constraint(validatedBy = AcceptedValidator.class)  
 @Target({ ElementType.METHOD, ElementType.FIELD })  
 @Retention(RetentionPolicy.RUNTIME)  
 public @interface AcceptedValues {  
   
      String[] acceptValues();  
   
      String message() default "Not allowded value";  
   
      Class<?>[] groups() default {};  
   
      Class<? extends Payload>[] payload() default {};  
 }  
   

Listing 5: Custom Validator implementation

 package com.web.validator;  
   
 import java.util.ArrayList;  
 import java.util.List;  
   
 import javax.validation.ConstraintValidator;  
 import javax.validation.ConstraintValidatorContext;  
   
 public class AcceptedValidator implements  
           ConstraintValidator<AcceptedValues, String> {  
   
      private List<String> valueList;  
   
      @Override  
      public void initialize(AcceptedValues constraintAnnotation) {  
           valueList = new ArrayList<String>();  
           for (String val : constraintAnnotation.acceptValues()) {  
                valueList.add(val.toLowerCase());  
           }  
      }  
   
      @Override  
      public boolean isValid(String value, ConstraintValidatorContext context) {  
           if (!valueList.contains(value.toLowerCase())) {  
                return false;  
           }  
           return true;  
      }  
   
 }  
   

- Because of the orderDate property in Order class, mapper is required for the reader, if not implemented
reader will not know the date format.

Listing 6: Order mapper

 package com.web.mapper;  
   
 import org.springframework.batch.item.file.mapping.FieldSetMapper;  
 import org.springframework.batch.item.file.transform.FieldSet;  
 import org.springframework.validation.BindException;  
   
 import com.web.model.Order;  
   
 public class OrderMapper implements FieldSetMapper<Order>{  
   
      @Override  
      public Order mapFieldSet(FieldSet fieldSet) throws BindException {  
           Order order = new Order();  
             
           order.setCode(fieldSet.readString(0));  
           order.setServiceType(fieldSet.readString(1));  
           order.setQuantity(fieldSet.readInt(2));  
           order.setOffer(fieldSet.readString(3));  
           order.setBillType(fieldSet.readString(4));  
           order.setOrderDate(fieldSet.readDate(5, "dd-MM-yyyy"));  
             
           return order;  
      }  
   
 }  
   

- The validation process takes place in OrderProcessor class, listing 7.

Listing 7: OrderProcessor class

 package com.web.processor;  
   
 import org.springframework.batch.item.ItemProcessor;  
 import org.springframework.batch.item.validator.ValidationException;  
 import org.springframework.validation.BindingResult;  
 import org.springframework.validation.DataBinder;  
 import org.springframework.validation.ObjectError;  
 import org.springframework.validation.Validator;  
   
 import com.web.model.Order;  
   
 public class OrderProcessor implements ItemProcessor<Order, Order> {  
   
      private Validator validator;  
   
      public void setValidator(Validator validator) {  
           this.validator = validator;  
      }  
   
      @Override  
      public Order process(Order item) throws Exception {  
   
           BindingResult results = BindAndValidate(item);  
           if (results.hasErrors())  
                buildValidationException(results);  
   
           item.setOffer(item.getOffer().toUpperCase());  
           return item;  
      }  
   
      private BindingResult BindAndValidate(Order item) {  
           DataBinder binder = new DataBinder(item);  
           binder.setValidator(validator);  
   
           binder.validate();  
   
           return binder.getBindingResult();  
   
      }  
   
      private void buildValidationException(BindingResult results) {  
           StringBuilder msg = new StringBuilder();  
           for (ObjectError error : results.getAllErrors()) {  
                msg.append("-*-*-*- \n" + error.toString() + "-*-*-*- \n");  
           }  
           throw new ValidationException(msg.toString());  
      }  
   
 }  
   

- Until now all the logging in the previous posts was console output, logging to file will be added to log the invalid records will in jobApplication.log file, Listing 8, logging property file

Listing 8: Logging properties to log into console and file

# Console output
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n

# File log
log4j.appender.A1=org.apache.log4j.RollingFileAppender
log4j.appender.A1.File=jobApplication.log
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
log4j.category.recordLogger=FATAL, A1
log4j.additivty.A1=false

log4j.category.org.springframework.integration=INFO
log4j.category.org.springframework.batch=INFO
log4j.category.org.springframework.jdbc=INFO
log4j.category.org.springframework.transaction=INFO

5. CSV files

- The input file with invalid data that will be used src/main/resources/cvs/skipInvalidOrderData.csv invalid date at:
 line 3 value fixe and future order date.
 line 5 order date format
 line 7 value prepai

- File with valid data can be found in src/main/resources/cvs/orders_12082014.csv

Listing 9: Invalid order record in data

 code,service_type,quantity,offer,bill_type,order_date  
 234ac,mobile,1,small,prepaid,12-08-2014
 235ab,fixe,1,medium,postpaid_mail,15-08-2015
 237ar,bundle,1,small,postpaid_email,10-08-2014
 236as,bundle,1,small,postpaid_email,12/08-2014
 239ac,fix,1,large,postpaid_mail,12-08-2014
 230mb,mobile,2,small,prepai,12-08-2014
 244rx,mobile,1,large,postpaid_email,12-08-2014
 274cb,mobile,5,medium,prepaid,12-08-2014
 228wb,mobile,1,large,postpaid_email,12-07-2014
 284mn,mobile,1,small,prepaid,12-08-2014
 285rn,fix,1,large,prepaid,12-08-2014
 292wr,bundle,1,medium,postpaid_email,12-08-2014
 293un,mobile,1,medium,prepaid,12-08-2014
 304po,bundle,1,large,postpaid_email,12-08-2014
 305is,bundle,1,large,postpaid_email,12-08-2014
 311jr,bundle,1,small,postpaid_mail,12-08-2014
 312ps,internet,1,fiber32,postpaid_email,12-08-2014
 317tr,mobile,3,small,prepaid,12-08-2014
 329qv,mobile,1,medium,prepaid,12-08-2014  

Listing 10: Valid order data

 code,service_type,quantity,offer,bill_type,order_date  
 234ac,mobile,9,small,prepaid,12-08-2014  
 235ab,fix,1,medium,postpaid_mail,14-08-2014  
 237ar,bundle,1,small,postpaid_email,10-08-2014  
 236as,bundle,1,small,postpaid_email,12-08-2014  
 239ac,fix,1,large,postpaid_mail,12-08-2014  
 230mb,mobile,2,small,prepaid,12-08-2014  
 244rx,mobile,1,large,postpaid_email,12-08-2014  
 274cb,mobile,5,medium,prepaid,12-08-2014  
 228wb,mobile,1,large,postpaid_email,12-07-2014  
 284mn,mobile,1,small,prepaid,12-08-2014  
 285rn,fix,1,large,prepaid,12-08-2014  
 292wr,bundle,1,medium,postpaid_email,12-08-2014  
 293un,mobile,1,medium,prepaid,12-08-2014  
 304po,bundle,1,large,postpaid_email,12-08-2014  
 305is,bundle,1,large,postpaid_email,12-08-2014  
 311jr,bundle,1,small,postpaid_mail,12-08-2014  
 312ps,internet,1,fiber32,postpaid_email,12-08-2014  
 317tr,mobile,3,small,prepaid,12-08-2014  
 329qv,mobile,1,medium,prepaid,12-08-2014  

6. Running orderValidationJob in eclipse

- To run the application use code in class src/main/java/com/web/app/LoggingInvalidOrderApp.java Listing 11.

Listing 11: LoggingInvalidOrderApp Class

 package com.web.app;  
   
 import java.util.Date;  
   
 import org.apache.log4j.Logger;  
 import org.springframework.batch.core.Job;  
 import org.springframework.batch.core.JobExecution;  
 import org.springframework.batch.core.JobParameters;  
 import org.springframework.batch.core.JobParametersBuilder;  
 import org.springframework.batch.core.launch.JobLauncher;  
 import org.springframework.context.support.ClassPathXmlApplicationContext;  
   
 public class LoggingInvalidOrderApp {  
   
      private static final Logger log = Logger.getLogger(LoggingInvalidOrderApp.class);  
      private static ClassPathXmlApplicationContext context;  
   
      public static void main(String[] args) {  
   
           String exitStatus = "Exit Status: ";  
   
           String[] springConfig = { "launch-context.xml",  
                     "META-INF/spring/batch/jobs/job-orderInvalidLogging.xml" };  
   
           context = new ClassPathXmlApplicationContext(springConfig);  
   
           JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");  
           Job job = (Job) context.getBean("orderLoggingJob");  
   
           try {  
                JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();  
   
                jobParametersBuilder.addString("dateTime", (new Date()).toString());  
                jobParametersBuilder.addString("inputDataFile",  
                          "classpath:cvs/skipInvalidOrderData.csv");  
   
                JobParameters param = jobParametersBuilder.toJobParameters();  
                JobExecution execution = jobLauncher.run(job, param);  
   
                exitStatus += execution.getStatus().toString();  
   
           } catch (Exception e) {  
                log.info(e.getStackTrace());  
           } finally {  
                if (context != null)  
                     context.close();  
           }  
   
           log.info(exitStatus);  
           log.info("Job Finished");  
   
      }  
   
 }  
  

- The result of running is in Listing 12.

Listing 12: Output of orderValidationJob execution in eclipse

 .  
 .  
 .  
 .  
INFO main app.LoggingInvalidOrderApp:49 - Exit Status: COMPLETED
INFO main app.LoggingInvalidOrderApp:50 - Job Finished 

7. Running orderJob From Spring Batch Admin

- In pom.xml change username_tc and password_tc to suite your local tomcat. 

- Start tomcat.

- Deploy application from run configuration in goal type clean install tomcat7:redeploy , click apply and 

Run buttons.

Note: If these steps are unfamiliar read steps 4 and 5 deploying spring batch admin Link How To

- open address localhost:8080 /springBatchAdminMysql/jobs  in web browser , picture 2

Picture 2: orderLoggingJob in jobs list

- Click orderLoggingJob, Change dateTime parameter, click launch button.  job will completed with COMPLETED status.

- Click on executions -> ID number of job execution -> COMPLETED under column Status to the 2 skips by processor and 1 skip by reader picture 3.

Picture 3: Step Summary View

- the jobApplication.log file, listing 13, 2 record are skipped in processing phase, while one record was skipped because of parsing error of date in reading phase of the job.

Listing 13: jobApplication.log

 FATAL jobLauncherTaskExecutor-1 recordLogger:32 - Processor skip: Order [code=235ab, type=fixe, quantity=1, offer=medium, billType=postpaid_mail, orderDate=Sat Aug 15 00:00:00 CEST 2015]
 FATAL jobLauncherTaskExecutor-1 recordLogger:32 - Processor skip: Order [code=230mb, type=mobile, quantity=2, offer=small, billType=prepai, orderDate=Tue Aug 12 00:00:00 CEST 2014]
 FATAL jobLauncherTaskExecutor-1 recordLogger:32 - Reader skip:236as,bundle,1,small,postpaid_email,12/08-2014

8. Eclipse Project

 Download project source code