title: JSR 352 with Liberty Profile - how to implement checkpointing when ItemReader does a DB query
body: <p>I have 10 records in my source table and I am having item count as 3.</p>
  
  <p>I have 2 partitions to process these 10 records(i.e first 5 records will be processed in first partition and remaining records processed in 2nd partition while processing records in 2nd partition I am throwing an exception so job will be failed at 2nd chunk of 2nd partition.when I am restarting the job ,failed partition is processing all the records again(that is first chunk and 2nd chunk).
  Restarting  the job  should only process from last failed chunk records but not all the records in that partition.Can you please guide me how to achieve this?</p>
  
  <p>My JSL is like below:  </p>
  
  <pre><code>    &lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;job xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
      xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
      id="readingfrom-db" restartable="true" version="1.0" &gt;
      &lt;properties &gt;
          &lt;property name="numRec" value="#{jobParameters['numRec']}?:5;"/&gt;        
          &lt;property name="chunkSize" value="#{jobParameters['chunkSize']}?:3;"/&gt;
          &lt;property name="whereclauseFrom" value="#{jobParameters['whereclauseFrom']}?:5;"/&gt;
          &lt;property name="whereclauseTo" value="#{jobParameters['whereclauseTo']}?:6;"/&gt;      
          &lt;property name="dsJNDI" value="#{jobParameters['dsJNDI']}?:jdbc/db2;"/&gt;
          &lt;property name="dsJNDI1" value="#{jobParameters['dsJNDI1']}?:jdbc/db2;"/&gt;
          &lt;property name="tableName" value="#{jobParameters['tableName']}?:CISDW.AIF1_CH;"/&gt;
          &lt;property name="ProcesstableName" value="#{jobParameters['ProcesstableName']}?:CISDW.PROC_AIF1_CH;"/&gt;
      &lt;/properties&gt;
      &lt;step id="runcache" next="readFromDB"&gt;
          &lt;batchlet ref="com.cdc.runcache.CacheRunnerBatchlet" /&gt;
      &lt;/step&gt;
      &lt;step id="readFromDB"&gt;
          &lt;listeners&gt;
              &lt;listener ref="com.cdc.dbreader.LogExceptionListener"/&gt;
          &lt;/listeners&gt; 
          &lt;chunk item-count="3" checkpoint-policy="item"&gt;
              &lt;reader ref="com.cdc.dbreader.DBItemReader"&gt;
                  &lt;properties &gt;
                      &lt;property name="dsJNDI" value="#{jobProperties['dsJNDI']}"/&gt;
                      &lt;property name="tableName" value="#{jobProperties['tableName']}"/&gt;
                      &lt;property name="whereclauseFrom" value="#{partitionPlan['modrec']}"/&gt;                   
                  &lt;/properties&gt;
              &lt;/reader&gt;
              &lt;processor ref="com.cdc.dbreader.DBItemProcessor" /&gt;            
              &lt;writer ref="com.cdc.dbreader.DBItemWriter"&gt;
                  &lt;properties &gt;
                      &lt;property name="dsJNDI" value="#{jobProperties['dsJNDI1']}"/&gt;
                      &lt;property name="tableName" value="#{jobProperties['ProcesstableName']}"/&gt;
                  &lt;/properties&gt;
              &lt;/writer&gt;
          &lt;/chunk&gt;
          &lt;partition&gt;
              &lt;plan partitions="2" threads="2"&gt;
                  &lt;properties partition="0"&gt;
                      &lt;property name="modrec" value="#{jobProperties['whereclauseFrom']}"/&gt;                   
                  &lt;/properties&gt;
                  &lt;properties partition="1"&gt;
                      &lt;property name="modrec" value="#{jobProperties['whereclauseTo']}"/&gt;
                  &lt;/properties&gt;
              &lt;/plan&gt;
          &lt;/partition&gt;        
      &lt;/step&gt;                     
  &lt;/job&gt;
  </code></pre>
  
  <p>My Item Reader is like below:</p>
  
  <pre><code> public class DBItemReader implements ItemReader {  
      @Inject
      @BatchProperty
      private String dsJNDI;
  
      @Inject
      @BatchProperty
      private String whereclauseFrom;
  
  
      @Inject
      @BatchProperty
      private String tableName;
  
      private Connection conn =null;
      private int totalRecords=0;
  
      private DataSource ds = null;
      List&lt;RecObj&gt; listRecObj=new ArrayList&lt;RecObj&gt;();    
  
      @Override
      public Object readItem() throws SQLException {
           if (listRecObj.size() == 0) {             
               return null;
           } else { 
               RecObj rec =null;           
               Iterator&lt;RecObj&gt; iter =listRecObj.iterator();
               while (iter.hasNext()) {               
                  rec = iter.next();               
  
                 if (Integer.parseInt(rec.getRec())  == 7) {                      
                    throw new IllegalStateException("Thrown Error");
                  }
                  iter.remove();
                  return rec;
               }
               return rec;
           }
  
  
       @Override
      public void open(Serializable arg0) throws NamingException, SQLException {
            ds = DataSource.class.cast(new InitialContext().lookup(dsJNDI)); 
  //        System.out.println("whereclauseFrom: " + whereclauseFrom);          
            conn = ds.getConnection(); 
            String sql ="";
            if(Integer.parseInt(whereclauseFrom) == 5){
                sql = "SELECT * FROM " + tableName + " WHERE  CAST(REC AS INTEGER) &lt;= "+ whereclauseFrom;
            }else if(Integer.parseInt(whereclauseFrom) == 6){
                sql = "SELECT * FROM " + tableName + " WHERE  CAST(REC AS INTEGER) &gt;= "+ whereclauseFrom;
            }
  
            PreparedStatement ps = conn.prepareStatement(sql);
            ResultSet rs=ps.executeQuery();
            while(rs.next()){
               totalRecords++;
               String rec=rs.getString("REC"); 
               if(rec != null)
                  listRecObj.add(new RecObj(rec));
  
            }          
            rs.close();          
      }   
      @Override
      public void close() throws SQLException {
          conn.close();       
      }   
      @Override
      public Serializable checkpointInfo() {       
              return null;
      }
  
  }
      }
  </code></pre>
  
  <p>MY Writer class is like below:</p>
  
  <pre><code>public class DBItemWriter extends AbstractItemWriter implements ItemWriter {    
      @Inject
      @BatchProperty
      private String dsJNDI;
  
      @Inject
      @BatchProperty
      private String tableName;
  
      private DataSource ds = null; 
  
      @Override
      public void open(Serializable arg0) throws NamingException {
           ds = DataSource.class.cast(new InitialContext().lookup(dsJNDI));            
      }
  
      @Override
      public void writeItems(List&lt;java.lang.Object&gt; items) throws BatchUpdateException,SQLException{      
          Connection conn = ds.getConnection();           
          String sql = "INSERT INTO "+tableName+ "(MOD_REC) VALUES(?) ";       
          PreparedStatement ps = conn.prepareStatement(sql);        
          for (Object obj : items) {          
              RecObj v = (RecObj)obj;
              System.out.println("=======Writer values===="+v.getRec());                
              ps.setString(1, v.getRec());            
              ps.addBatch();
          }        
          ps.executeBatch();
          ps.clearBatch();
          ps.close();
          conn.close();
      }
  }
  </code></pre>
  
  <p>Below is my Processor:</p>
  
  <pre><code>public class DBItemProcessor implements ItemProcessor {
      Integer count=0;   
      @Override
      public Object processItem(Object arg0) {
          count++;
          RecObj v=(RecObj)arg0;
          String vname=v.getRec();
          System.out.println("=========Processer Values==="+vname);
          return new RecObj(vname+count);
      }
  }
  </code></pre>
  
  <p>Below is my Bean class</p>
  
  <pre><code>public class RecObj {
     private String rec;
  
  
    public RecObj(String rec) {
      this.rec=rec;
  }
  </code></pre>