Tenho 10 registros na minha tabela de origem e estou tendo a contagem de itens como 3.

Eu tenho 2 partições para processar esses 10 registros (ou seja, os 5 primeiros registros serão processados na primeira partição e os demais registros serão processados na 2ª partição enquanto processando registros na 2ª partição. Estou lançando uma exceção para que o trabalho falhe no 2º bloco da 2ª partição. Quando estou reiniciando o trabalho, a partição com falha está processando todos os registros novamente (ou seja, o primeiro bloco e o segundo bloco). Reiniciar o trabalho deve processar apenas os últimos registros de blocos com falha, mas nem todos os registros dessa partição. como conseguir isso?

Meu JSL é como abaixo:

    <?xml version="1.0" encoding="UTF-8"?>
<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" >
    <properties >
        <property name="numRec" value="#{jobParameters['numRec']}?:5;"/>        
        <property name="chunkSize" value="#{jobParameters['chunkSize']}?:3;"/>
        <property name="whereclauseFrom" value="#{jobParameters['whereclauseFrom']}?:5;"/>
        <property name="whereclauseTo" value="#{jobParameters['whereclauseTo']}?:6;"/>      
        <property name="dsJNDI" value="#{jobParameters['dsJNDI']}?:jdbc/db2;"/>
        <property name="dsJNDI1" value="#{jobParameters['dsJNDI1']}?:jdbc/db2;"/>
        <property name="tableName" value="#{jobParameters['tableName']}?:CISDW.AIF1_CH;"/>
        <property name="ProcesstableName" value="#{jobParameters['ProcesstableName']}?:CISDW.PROC_AIF1_CH;"/>
    </properties>
    <step id="runcache" next="readFromDB">
        <batchlet ref="com.cdc.runcache.CacheRunnerBatchlet" />
    </step>
    <step id="readFromDB">
        <listeners>
            <listener ref="com.cdc.dbreader.LogExceptionListener"/>
        </listeners> 
        <chunk item-count="3" checkpoint-policy="item">
            <reader ref="com.cdc.dbreader.DBItemReader">
                <properties >
                    <property name="dsJNDI" value="#{jobProperties['dsJNDI']}"/>
                    <property name="tableName" value="#{jobProperties['tableName']}"/>
                    <property name="whereclauseFrom" value="#{partitionPlan['modrec']}"/>                   
                </properties>
            </reader>
            <processor ref="com.cdc.dbreader.DBItemProcessor" />            
            <writer ref="com.cdc.dbreader.DBItemWriter">
                <properties >
                    <property name="dsJNDI" value="#{jobProperties['dsJNDI1']}"/>
                    <property name="tableName" value="#{jobProperties['ProcesstableName']}"/>
                </properties>
            </writer>
        </chunk>
        <partition>
            <plan partitions="2" threads="2">
                <properties partition="0">
                    <property name="modrec" value="#{jobProperties['whereclauseFrom']}"/>                   
                </properties>
                <properties partition="1">
                    <property name="modrec" value="#{jobProperties['whereclauseTo']}"/>
                </properties>
            </plan>
        </partition>        
    </step>                     
</job>

Meu leitor de itens é como abaixo:

 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<RecObj> listRecObj=new ArrayList<RecObj>();    

    @Override
    public Object readItem() throws SQLException {
         if (listRecObj.size() == 0) {             
             return null;
         } else { 
             RecObj rec =null;           
             Iterator<RecObj> 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) <= "+ whereclauseFrom;
          }else if(Integer.parseInt(whereclauseFrom) == 6){
              sql = "SELECT * FROM " + tableName + " WHERE  CAST(REC AS INTEGER) >= "+ 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;
    }

}
    }

A minha classe Writer é como abaixo:

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<java.lang.Object> 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();
    }
}

Abaixo está o meu processador:

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);
    }
}

Abaixo está a minha classe Bean

public class RecObj {
   private String rec;


  public RecObj(String rec) {
    this.rec=rec;
}