I have 10 records in my source table and I am having item count as 3.
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?
My JSL is like below:
<?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>
My Item Reader is like below:
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;
}
}
}
MY Writer class is like below:
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();
}
}
Below is my Processor:
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);
}
}
Below is my Bean class
public class RecObj {
private String rec;
public RecObj(String rec) {
this.rec=rec;
}
You need to return a checkpoint value in your reader's checkpointInfo() which will be passed into your reader's open() method on restart. This is how a reader and the batch container coordinate to provide checkpointing on restart.
So you could have something like (look for the CHECKPOINT comments):
public class DBItemReader implements ItemReader {
// ...
// CHECKPOINT field defined
private String checkpoint = null;
@Override
public void open(Serializable checkpoint) throws NamingException, SQLException {
// CHECKPOINT-based positioning through query value.
// Initial position = whereclauseFrom, on restart set to checkpoint
String queryVal = (String)(checkpoint == null ? whereclauseFrom : checkpoint);
if(Integer.parseInt(whereclauseFrom) == 5){
sql = "SELECT * FROM " + tableName + " WHERE CAST(REC AS INTEGER) <= "+ queryVal;
}else if(Integer.parseInt(whereclauseFrom) == 6){
sql = "SELECT * FROM " + tableName + " WHERE CAST(REC AS INTEGER) >= "+ queryVal;
}
// ..
}
@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();
// CHECKPOINT updated
checkpoint = rec.getRec();
if (Integer.parseInt(rec.getRec()) == 7) {
throw new IllegalStateException("Thrown Error");
}
}
}
// ...
}
@Override
public Serializable checkpointInfo() {
// CHECKPOINT returned at end of chunk
return checkpoint;
}
}