Step Partitioning

Time needed to complete  ~75 minutes

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

1.Introduction

In the situation where single threaded batch job can’t finish the work in a given time interval and tuning  JVM didn't do the trick, then it is time for scaling the batch job using multithreading. In spring batch partitioning is one way for scaling batch jobs that can improve performance.  

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 3.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 4.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 from csv files,process and write in database table is in src/main/resources/META-INF/spring/batch/jobs/job-order-multithreading.xml , Listing 4.2. The job has two steps, organizeFilesStep step will create tmp folder in mt_data folder that will contains folders with names as number of partitions then in each folder a number of csv files will be copyed(see DistributeFiles.class). Picture 4.1.

Listing 4.2: Job definition in job-order-multithreading

 <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" />  
      <bean id="partitioner" class="com.web.partition.MakePartition" />  
      <bean id="asyncTaskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />  
      <bean id="distributeFiles" class="com.web.general.DistributeFiles" scope="step">  
           <property name="dataDirectory" value="classpath:cvs/mt_data" />  
           <property name="numberOfPartitions" value="#{jobParameters['numberOfPartitions']}" />  
      </bean>  
   
      <!-- Batch job -->  
      <batch:job id="orderMTJob">  
   
           <batch:step id="organizeFilesStep" next="step1.boss">  
                <batch:tasklet ref="distributeFiles" />                 
           </batch:step>  
   
           <batch:step id="step1.boss">  
                <batch:partition step="worker" partitioner="partitioner" handler="orderPartitionHandler" />  
                     <!-- batch:handler grid-size="3" task-executor="asyncTaskExecutor"/>  
                </batch:partition -->  
           </batch:step>  
   
      </batch:job>  
   
      <batch:step id="worker">  
           <batch:tasklet>  
                <batch:chunk reader="filesReader" processor="orderProcessor"  
                     writer="mysqlItemWriter" commit-interval="15" 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>  
        
      <bean id="orderPartitionHandler" class="org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler" scope="step">  
   <property name="taskExecutor" ref="asyncTaskExecutor"/>  
   <property name="step" ref="worker" />  
   <property name="gridSize" value="#{jobParameters['numberOfPartitions']}" />  
 </bean>  
   
   
      <bean id="orderProcessor" class="com.web.processor.OrderProcessorMT" scope="step">  
           <property name="validator" ref="validator" />            
      </bean>  
        
       <bean id="filesReader"     class=" org.springframework.batch.item.file.MultiResourceItemReader" scope="step">  
      <property name="resources" value="classpath:cvs/mt_data/tmp/#{stepExecutionContext['partitionNo']}/#{jobParameters['inputDataFileFormat']}" />  
      <property name="delegate" ref="cvsFileItemReader" />  
  </bean>  
    
      <bean id="cvsFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="prototype">  
           <property name="linesToSkip" value="1" />  
           <!-- Read order csv file -->  
             
           <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" scope="prototype">  
           <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>  
   

Picture 4.1: organizeFilesStep will copy files in emp/number_of_partition folder

- The second step is step1.boss , picture 4.2, in this step 3 threads (grid-size) will be created, where thread1 will read csv files only from tmp/1 folder, thread2 will read only from tmp/2 folder and thread3 will read only from tmp/3 folder. Each thread will have it's own reader,processor and writer. Readers and Writers bean have scope attribute prototype , and it is a good place to learn what happened if scope="prototype" is removed. Try it and see the data race to lineNumber in reader. Reader by default are singleton's without prototype scope there will be one reader instance and three threads that reading and updating the next lineNumber to be read, causing incorrect behavior. An important parameter is commit-interval changing it's value will have a high impact on the performance, e.g.  value 3 will degenerate performance. 

Picture 4.2: orderMTJob Steps

- The Order class listing 4.3.

Listing 4.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.4: 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;  
      }  
   
 }  
   

- MakePartition, Listing 4.5.

Listing 4.5: orderMTJob Steps

 package com.web.partition;  
   
 import java.util.HashMap;  
 import java.util.Map;  
   
 import org.springframework.batch.core.partition.support.Partitioner;  
 import org.springframework.batch.item.ExecutionContext;  
   
 public class MakePartition implements Partitioner {  
   
      @Override  
      public Map<String, ExecutionContext> partition(int gridSize) {  
             
           Map<String, ExecutionContext> executionContextMap = new HashMap<String, ExecutionContext>();  
             
           for (int i = 1; i <= gridSize; i++) {  
                  
                ExecutionContext executionContext = new ExecutionContext();                 
                executionContext.putString("threadId", "Thread_"+i);  
                executionContext.putInt("partitionNo", i);                 
                executionContextMap.put("Partition_"+i, executionContext);  
           }  
             
           return executionContextMap;  
      }  
   
 }  
   

- SQL for creating table company_db.tel_order

Listing 4.6: SQL for creating tel_order table

 CREATE TABLE tel_order (  
  id INTEGER(11) NOT NULL AUTO_INCREMENT,  
  code VARCHAR(5) NOT NULL,  
  service_type varchar(32) NOT NULL,  
  quantity INTEGER(11) NOT NULL,  
  offer VARCHAR(32) NOT NULL,  
  bill_type VARCHAR(32) NOT NULL,  
  order_date timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  PRIMARY KEY (id)  
 ) ENGINE=InnoDB;  

5. CSV files

- The are 100 csv input files in folder mt_data containing 1900 orders.

Listing 5.1: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 orderMTJob in eclipse

- To run the application use code in class src/main/java/com/web/app/MultithreadingOrderApp.java Listing 6.1. Changing  jobParameters  numberOfPartitions(grid-size) will change the execution time for the batch job, bigger i not better, it depends on the number of threads(logical processor) in the processor. Example 8 cores-16 threads processor.

Listing 6.1: MultithreadingOrderAppClass

 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 MultithreadingOrderApp {  
   
      private static final Logger log = Logger.getLogger(MultithreadingOrderApp.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-order-multithreading.xml" };  
   
           context = new ClassPathXmlApplicationContext(springConfig);  
   
           JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");  
           Job job = (Job) context.getBean("orderMTJob");  
   
           try {  
                JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();  
   
                jobParametersBuilder.addString("dateTime", (new Date()).toString());  
                jobParametersBuilder.addString("numberOfPartitions", "3");       
                jobParametersBuilder.addString("inputDataFileFormat", "orders_*.csv");  
                  
   
                JobParameters param = jobParametersBuilder.toJobParameters();  
                  
                long t1 = System.currentTimeMillis();  
                JobExecution execution = jobLauncher.run(job, param);  
                log.info("TIME:"+(System.currentTimeMillis()- t1));  
                  
   
                exitStatus += execution.getStatus().toString();  
   
           } catch (Exception e) {  
                log.info(e.getStackTrace());  
           } finally {  
                if (context != null)  
                     context.close();  
           }  
   
           log.info(exitStatus);  
           log.info("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 7.1

Picture 7.1: Jobs list

- Click orderMTJob, Change dateTime parameter, click launch button.  job will completed with COMPLETED status. picture 7.2. orderMTJob steps.

Picture 7.2:orderMTJob steps status

- Click on executions -> ID number of job execution -> COMPLETED under column Status picture 7.3.

Picture 7.3: step1.boss Step Summary View

8. Eclipse Project

 Download project source code

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

Validate Input data

Time needed to complete  ~60 minutes

Prerequisite for this tutorial is Configure Spring Batch Admin to use MySQL
and Developing first spring batch job

1.Introduction

What money is for bankers, data is for IT developers . Money can be dirty, data can be dirty too, for this reason developers need to check data using validation  before sending for processing. This article will show how validation can be used in spring batch jobs.

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  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-orderValidate.xml , Listing 2.

Listing 2: Job definition in job-orderValidate.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 which data will be written to order table -->  
      <bean id="order" class="com.web.model.Order" scope="prototype" />  
      <bean id="validator"  
           class="org.springframework.validation.beanvalidation.LocalValidatorFactoryBean" />  
      <!-- Batch job -->  
      <batch:job id="orderValidationJob">  
           <batch:step id="step1">  
                <batch:tasklet>  
                     <batch:chunk reader="cvsFileItemReader" processor="orderProcessor"  
                          writer="mysqlItemWriter" commit-interval="5">  
                     </batch:chunk>  
                </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">  
                     <!-- split 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>  
           <!-- mapping 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());  
      }  
   
 }  
   

5. CSV files

- The input file with invalid data that will be used src/main/resources/cvs/invalidOrderData.csv line 7 mobilex and 12-08-2015.

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

Listing 8: Invalid order record in 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,mobilex,2,small,prepaid,12-08-2015  
 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 9: 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/OrderValidationApp.java Listing 10.

Listing 10: OrderValidationApp 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 OrderValidationApp {  
   
      private static final Logger log = Logger.getLogger(OrderValidationApp.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-orderValidate.xml" };  
   
           context = new ClassPathXmlApplicationContext(springConfig);  
   
           JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");  
           Job job = (Job) context.getBean("orderValidationJob");  
   
           try {  
                JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();  
   
                jobParametersBuilder.addString("dateTime", (new Date()).toString());  
                jobParametersBuilder.addString("inputDataFile",  
                          "classpath:cvs/invalidOrderData.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 11.

Listing 11: Output of orderValidationJob execution in eclipse

 .  
 .  
 .  
 org.springframework.batch.item.validator.ValidationException: -*-*-*-   
 Field error in object 'target' on field 'serviceType': rejected value [mobilex];   
 codes [AcceptedValues.target.serviceType,AcceptedValues.serviceType,AcceptedValues.java.lang.String,AcceptedValues];  
 arguments [org.springframework.context.support.DefaultMessageSourceResolvable: codes [target.serviceType,serviceType];  
 arguments []; default message [serviceType],[Ljava.lang.String;@6d6474ec]; default message [Invalid service type]  
 -*-*-*-*-*-*-   
 Field error in object 'target' on field 'orderDate': rejected value [Wed Aug 12 00:00:00 CEST 2015];  
 codes [Past.target.orderDate,Past.orderDate,Past.java.util.Date,Past];  
 arguments [org.springframework.context.support.DefaultMessageSourceResolvable:   
 codes [target.orderDate,orderDate]; arguments []; default message [orderDate]];  
 default message [Order date should be in the past]-*-*-*-   
 .  
 .  
 .  
 INFO main app.OrderValidationApp:48 - Exit Status: FAILED  
 INFO main app.OrderValidationApp:49 - 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: orderValidationJob in jobs list

- Click orderValidationJob, Change dateTime parameter, click launch button.  job will completed with FAILED status.

- Click on executions -> ID number of job execution -> FAILED under column Status to see validation messages picture 3.

Picture 3: Validation messages

8. Eclipse Project

 Download project source code

Dealing with empty inputs

Time needed to complete  ~20 minutes
Prerequisite for this tutorial is Configure Spring Batch Admin to use MySQL

1.Introduction

Errors in programming are the norm not the exception. Finding and fixing errors is not an easy task, specially when user reports an error that don't exist, both code and data are correct. A common situation in batch processing is empty file, no input. Clients see no output result, the conclusion is batch job didn't finished, report bug. To prevent this situation spring batch defines a listener after step is finished to check the number of reading. 

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  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-emptyFile.xml , Listing 2.

Listing 2: Batch job definition job-emptyFile.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">  
      <import resource="../dbConfig/database.xml" />  
      <bean id="order" class="com.web.model.Order" scope="prototype" />  
      <bean id="afterStepListener" class="com.web.listener.EmptyFileHandler" />
      <bean id="afterJobListener" class="com.web.listener.JobListener" />   
      <batch:job id="orderEmptyFileJob">  
           <batch:step id="step1">  
                <batch:tasklet>  
                     <batch:chunk reader="cvsFileItemReader" processor="orderProcessor"  
                          writer="mysqlItemWriter" commit-interval="3">  
                     </batch:chunk>  
                      <batch:listeners>    
                          <batch:listener ref="afterStepListener" />    
                      </batch:listeners>    
                </batch:tasklet>  
           </batch:step>
           <batch:listeners>
           <batch:listener ref="afterJobListener" />
           </batch:listeners>  
      </batch:job>  
      <bean id="orderProcessor" class="com.web.processor.OrderProcessor" />  
      <bean id="cvsFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader"  
           scope="step">  
           <property name="resource" value="#{jobParameters['inputDataFile']}" />  
           <property name="lineMapper">  
                <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">  
                     <property name="lineTokenizer">  
                          <bean  
                               class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">  
                               <property name="names" value="id,name,surname,age,accountNumber" />  
                          </bean>  
                     </property>  
                     <property name="fieldSetMapper">  
                          <bean  
                               class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">  
                               <property name="prototypeBeanName" value="order" />  
                          </bean>  
                     </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,STR_TO_DATE(:orderDate,'%d-%m-%Y'))  
       ]]>  
                </value>  
           </property>  
           <property name="itemSqlParameterSourceProvider">  
                <bean  
                     class="org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider" />  
           </property>  
      </bean>  
 </beans>  

- A bean afterStepListener is defined and referenced in the orderEmptyFileJob job definition, afterStepListener will use code in a class com.web.listener.EmptyFileHandler, Listing 3.

Listing 3: EmptyFileHandler class

 package com.web.listener;  
 import org.apache.log4j.Logger;  
 import org.springframework.batch.core.ExitStatus;  
 import org.springframework.batch.core.StepExecution;  
 import org.springframework.batch.core.annotation.AfterStep;  
 public class EmptyFileHandler {  
      private static final Logger log = Logger.getLogger(EmptyFileHandler.class);  
      @AfterStep  
      public ExitStatus afterStep(StepExecution execution) {  
           if (execution.getReadCount() > 0) {  
                return execution.getExitStatus();  
           } else {  
                log.error("-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-");  
                log.error("EMPTY FILE");  
                log.error("-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-");  
                return ExitStatus.FAILED;  
           }  
      }  
 }  

- afterStep method is annotated with @afterStep, and the method will be executed after step, checking the readcount and if it is not greater than zero will log a message that file is empty and return job status failed.

- A bean afterJobListener is defined and referenced in the orderEmptyFileJob job definition, afterJobListener will use code in a class com.web.listener.JobListener, Listing 4.

Listing 4: JobListener class

 package com.web.listener;  
   
 import java.util.Collection;  
 import java.util.Iterator;  
   
 import org.springframework.batch.core.ExitStatus;  
 import org.springframework.batch.core.JobExecution;  
 import org.springframework.batch.core.StepExecution;  
 import org.springframework.batch.core.annotation.AfterJob;  
   
 public class JobListener {  
      private static final String STEP_NAME = "step1";  
      private static final String EXIT_MSG = "EMPTY FILE -";  
   
      @AfterJob 
      public void afterJob(JobExecution jobExecution) {  
   
           Collection<StepExecution> stepExecutionCol = jobExecution.getStepExecutions();  
           Iterator<StepExecution> itr = stepExecutionCol.iterator();  
   
           // Only one step, go for it directly no loop  
           StepExecution stepexe = itr.next();  
   
           if (STEP_NAME.equals(stepexe.getStepName())     && stepexe.getReadCount() < 1) {  
                ExitStatus es = jobExecution.getExitStatus();  
                jobExecution.setExitStatus(new ExitStatus(es.getExitCode(),     EXIT_MSG));  
           }  
      }  
 }  
   

5. CSV file

- The input empty file that will be used src/main/resources/cvs/emptyFile.csv
- For non empty file test use data in listing 5.1.

Listing 5.1: Testing data for non empty file

 234ac,mobile,1,small,prepaid,12-08-2014  
 235ab,fix,1,medium,postpaid_mail,14-08-2014  
 237ar,bundle,1,small,postpaid_email,10-08-2014  

6. Running orderEmptyFileJob in eclipse

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

Listing 5: EmptyFileHandlingApp to run orderEmptyFileJob

 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 EmptyFileHandlingApp {  
   
      private static final Logger log = Logger  
                .getLogger(EmptyFileHandlingApp.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-emptyFile.xml" };  
   
           context = new ClassPathXmlApplicationContext(springConfig);  
   
           JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");  
           Job job = (Job) context.getBean("orderEmptyFileJob");  
   
           try {  
                JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();  
   
                jobParametersBuilder.addString("dateTime", (new Date()).toString());  
                jobParametersBuilder.addString("inputDataFile",  
                          "classpath:cvs/emptyFile.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("Finished");  
   
      }  
   
 }  
   

- The result of running is Listing 6.

Listing 6: EmptyFileHandlingApp console output in eclipse

 .  
 .  
 .  
  INFO main job.SimpleStepHandler:139 - Executing step: [step1]  
  ERROR main listener.EmptyFileHandler:17 - -*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-  
  ERROR main listener.EmptyFileHandler:18 - EMPTY FILE  
  ERROR main listener.EmptyFileHandler:19 - -*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-  
 .  
 .  
 .  
  INFO main app.EmptyFileHandlingApp:49 - Exit Status: FAILED  
  INFO main app.EmptyFileHandlingApp:50 - 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: orderEmptyfileJob is listed in spring batch admin job list

- Click orderEmptyFileJob, Change dateTime parameter, click launch button, picture 3.  job will completed with FAILED status.

Picture 3: Job Launch screen

- Click on executions then ID number of job execution, to see exit message EMPTY FILE, picture 4.

Picture 4: EMPTY FILE message

8. Eclipse Project

 Download project source code