tm.begin();
try {         
      itemReader.open(stepOrPartitionExecution.getReaderCheckpointInfo());
      processingInfo.readPosition = processingInfo.checkpointPosition;          
      itemWriter.open(stepOrPartitionExecution.getWriterCheckpointInfo());
      tm.commit();
} catch (Exception e) {
      tm.rollback();
      // An error occurred, safely close the reader and writer
      safeClose();
      throw e;
}