Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - BATCH-550: Add skipCount in BATCH_STEP_EXECUTION table

Browse Source 
added readSkipCount and writeSkipCount to StepExecution class and db table
This commit is contained in:
robokaso
2008-04-30 12:59:17 +00:00
parent 471948a824
commit 0396023d33
15 changed files with 500 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
View File

@@ -30,10 +30,12 @@ public class StepContribution {
private int commitCount;
private int skipCount;
private int readSkipCount;
private int writeSkipCount;
private int uncommitedReadSkipCount;
/**
* @param execution
*/
@@ -78,7 +80,7 @@ public class StepContribution {
* skips.
*/
public int getStepSkipCount() {
return readSkipCount + skipCount + parentSkipCount;
return uncommitedReadSkipCount + readSkipCount + writeSkipCount + parentSkipCount;
}
/**
@@ -87,23 +89,30 @@ public class StepContribution {
* parent {@link StepExecution}.
*/
public int getSkipCount() {
return skipCount;
return readSkipCount + writeSkipCount;
}
/**
* Increment the total skip count for this contribution
* Increment the read skip count for this contribution
*/
public void incrementSkipCount() {
skipCount++;
}
/**
* Increment the counter for skipped reads
*/
public void incrementReadSkipCount() {
public void incrementReadsSkipCount() {
readSkipCount++;
}
/**
* Increment the write skip count for this contribution
*/
public void incrementWriteSkipCount() {
writeSkipCount++;
}
/**
* Increment the counter for temporary skipped reads
*/
public void incrementTemporaryReadSkipCount() {
uncommitedReadSkipCount++;
}
/**
* @return the read skip count
*/
@@ -112,11 +121,19 @@ public class StepContribution {
}
/**
* Combine the skip counts and reset read skips to zero.
* @return the write skip count
*/
public int getWriteSkipCount() {
return writeSkipCount;
}
/**
* Add the temporary read skip count to read skip count and reset the
* temporary counter.
*/
public void combineSkipCounts() {
skipCount += readSkipCount;
readSkipCount = 0;
readSkipCount += uncommitedReadSkipCount;
uncommitedReadSkipCount = 0;
}
/*
@@ -124,8 +141,8 @@ public class StepContribution {
* @see java.lang.Object#toString()
*/
public String toString() {
return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", readSkips=" + readSkipCount
+ ", skips=" + skipCount + "]";
return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", uncommitedReadSkips="
+ uncommitedReadSkipCount + ", readSkips=" + readSkipCount + "writeSkips=" + writeSkipCount + "]";
}
}

34
spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java
View File

@@ -47,7 +47,9 @@ public class StepExecution extends Entity {
private int rollbackCount = 0;
private int skipCount = 0;
private int readSkipCount = 0;
private int writeSkipCount = 0;
private Date startTime = new Date(System.currentTimeMillis());
@@ -301,8 +303,10 @@ public class StepExecution extends Entity {
public synchronized void apply(StepContribution contribution) {
itemCount += contribution.getItemCount();
commitCount += contribution.getCommitCount();
contribution.combineSkipCounts();
skipCount += contribution.getSkipCount();
readSkipCount += contribution.getReadSkipCount();
writeSkipCount += contribution.getWriteSkipCount();
}
/**
@@ -328,11 +332,15 @@ public class StepExecution extends Entity {
}
public int getSkipCount() {
return skipCount;
return readSkipCount + writeSkipCount;
}
public void incrementSkipCountBy(int count) {
skipCount += count;
public void incrementReadSkipCountBy(int count) {
readSkipCount += count;
}
public void incrementWriteSkipCountBy(int count) {
writeSkipCount += count;
}
/**
@@ -348,4 +356,20 @@ public class StepExecution extends Entity {
return jobExecution.getJobInstance().getJobParameters();
}
public int getReadSkipCount() {
return readSkipCount;
}
public int getWriteSkipCount() {
return writeSkipCount;
}
public void setReadSkipCount(int readSkipCount) {
this.readSkipCount = readSkipCount;
}
public void setWriteSkipCount(int writeSkipCount) {
this.writeSkipCount = writeSkipCount;
}
}

22
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/JdbcStepExecutionDao.java
View File

@@ -62,18 +62,18 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
+ " KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,?,?,?,?,?,?)";
private static final String SAVE_STEP_EXECUTION = "INSERT into %PREFIX%STEP_EXECUTION(STEP_EXECUTION_ID, VERSION, STEP_NAME, JOB_EXECUTION_ID, START_TIME, "
+ "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE) "
+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ "END_TIME, STATUS, COMMIT_COUNT, ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT) "
+ "values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String UPDATE_STEP_EXECUTION_CONTEXT = "UPDATE %PREFIX%STEP_EXECUTION_CONTEXT set "
+ "TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where STEP_EXECUTION_ID = ? and KEY_NAME = ?";
private static final String UPDATE_STEP_EXECUTION = "UPDATE %PREFIX%STEP_EXECUTION set START_TIME = ?, END_TIME = ?, "
+ "STATUS = ?, COMMIT_COUNT = ?, ITEM_COUNT = ?, CONTINUABLE = ? , EXIT_CODE = ?, "
+ "EXIT_MESSAGE = ?, VERSION = ? where STEP_EXECUTION_ID = ? and VERSION = ?";
+ "EXIT_MESSAGE = ?, VERSION = ?, READ_SKIP_COUNT = ?, WRITE_SKIP_COUNT = ? where STEP_EXECUTION_ID = ? and VERSION = ?";
private static final String GET_STEP_EXECUTION = "SELECT STEP_EXECUTION_ID, STEP_NAME, START_TIME, END_TIME, STATUS, COMMIT_COUNT,"
+ " ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE from %PREFIX%STEP_EXECUTION where STEP_NAME = ? and JOB_EXECUTION_ID = ?";
+ " ITEM_COUNT, CONTINUABLE, EXIT_CODE, EXIT_MESSAGE, READ_SKIP_COUNT, WRITE_SKIP_COUNT from %PREFIX%STEP_EXECUTION where STEP_NAME = ? and JOB_EXECUTION_ID = ?";
private static final String CURRENT_VERSION_STEP_EXECUTION = "SELECT VERSION FROM %PREFIX%STEP_EXECUTION WHERE STEP_EXECUTION_ID=?";
@@ -185,13 +185,14 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
stepExecution.getStepName(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
stepExecution.getItemCount(), stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
stepExecution.getExitStatus().getExitCode(), stepExecution.getExitStatus().getExitDescription() };
stepExecution.getExitStatus().getExitCode(), stepExecution.getExitStatus().getExitDescription(),
Integer.valueOf(stepExecution.getReadSkipCount()), Integer.valueOf(stepExecution.getWriteSkipCount()) };
getJdbcTemplate().update(
getQuery(SAVE_STEP_EXECUTION),
parameters,
new int[] { Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.INTEGER, Types.TIMESTAMP,
Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.CHAR, Types.VARCHAR,
Types.VARCHAR });
Types.VARCHAR, Types.INTEGER, Types.INTEGER });
}
/**
@@ -319,13 +320,16 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getItemCount(),
stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getId(),
stepExecution.getExitStatus().getExitCode(), exitDescription, version,
Integer.valueOf(stepExecution.getReadSkipCount()),
Integer.valueOf(stepExecution.getWriteSkipCount()), stepExecution.getId(),
stepExecution.getVersion() };
int count = getJdbcTemplate().update(
getQuery(UPDATE_STEP_EXECUTION),
parameters,
new int[] { Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER,
Types.CHAR, Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER });
Types.CHAR, Types.VARCHAR, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.INTEGER,
Types.INTEGER, Types.INTEGER });
// Avoid concurrent modifications...
if (count == 0) {
@@ -362,6 +366,8 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
stepExecution.setItemCount(rs.getInt(7));
stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(8)), rs.getString(9), rs.getString(10)));
stepExecution.setExecutionContext(findExecutionContext(stepExecution));
stepExecution.setReadSkipCount(rs.getInt(11));
stepExecution.setWriteSkipCount(rs.getInt(12));
return stepExecution;
}

3
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
View File

@@ -339,7 +339,8 @@ public class ItemOrientedStep extends AbstractStep {
private void processRollback(final StepExecution stepExecution, final StepContribution contribution,
final ExceptionHolder fatalException, TransactionStatus transaction) {
stepExecution.incrementSkipCountBy(contribution.getSkipCount());
stepExecution.incrementReadSkipCountBy(contribution.getReadSkipCount());
stepExecution.incrementWriteSkipCountBy(contribution.getWriteSkipCount());
/*
* Any exception thrown within the transaction should automatically
* cause the transaction to rollback.

4
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
View File

@@ -183,7 +183,7 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
try {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementReadSkipCount();
contribution.incrementTemporaryReadSkipCount();
if (listener != null) {
listener.onSkipInRead(e);
}
@@ -233,7 +233,7 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
}
catch (Exception e) {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
contribution.incrementSkipCount();
contribution.incrementWriteSkipCount();
// don't call the listener here - the transaction is going to
// roll back
addSkippedException(key, e);