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

OPEN - issue BATCH-1383: onSkipInProcess is not called

Browse Source 
http://jira.springframework.org/browse/BATCH-1383

Tidied up the skip sample.  Could not reproduce issue.
This commit is contained in:
dsyer
2009-09-04 10:36:45 +00:00
parent 9e5de91923
commit cdf94a7148
10 changed files with 203 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java
View File

@@ -11,13 +11,15 @@ import javax.sql.DataSource;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.JobParametersNotFoundException;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.sample.domain.trade.internal.ItemTrackingTradeItemWriter;
import org.springframework.batch.sample.domain.trade.internal.TradeWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
@@ -38,15 +40,12 @@ public class SkipSampleFunctionalTests {
private SimpleJdbcTemplate simpleJdbcTemplate;
@Autowired
private JobExplorer jobExplorer;
@Autowired
private JobOperator jobOperator;
@Autowired
private TradeWriter tradeWriter;
@Autowired
private ItemTrackingTradeItemWriter itemTrackingWriter;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
@@ -60,9 +59,6 @@ public class SkipSampleFunctionalTests {
simpleJdbcTemplate.update("INSERT INTO CUSTOMER VALUES (" + i + ", 0, 'customer" + i + "', 100000)");
}
simpleJdbcTemplate.update("DELETE from ERROR_LOG");
itemTrackingWriter.clearItems();
itemTrackingWriter.setWriteFailureISIN("UK21341EAH47");
}
/**
@@ -131,10 +127,10 @@ public class SkipSampleFunctionalTests {
// Launch 1
//
long id1 = launchJobWithIncrementer();
Map<String, Object> execution1 = this.getJobExecutionAsMap(id1);
assertEquals("COMPLETED", execution1.get("STATUS"));
JobExecution execution1 = jobExplorer.getJobExecution(id1);
assertEquals(BatchStatus.COMPLETED, execution1.getStatus());
validateLaunchWithSkips(id1);
validateLaunchWithSkips(execution1);
//
// Clear the data
@@ -145,59 +141,59 @@ public class SkipSampleFunctionalTests {
// Launch 2
//
long id2 = launchJobWithIncrementer();
Map<String, Object> execution2 = getJobExecutionAsMap(id2);
assertEquals("COMPLETED", execution2.get("STATUS"));
JobExecution execution2 = jobExplorer.getJobExecution(id2);
assertEquals(BatchStatus.COMPLETED, execution2.getStatus());
validateLaunchWithoutSkips(id2);
validateLaunchWithoutSkips(execution2);
//
// Make sure that the launches were separate executions and separate
// instances
//
assertTrue(id1 != id2);
assertTrue(!execution1.get("JOB_INSTANCE_ID").equals(execution2.get("JOB_INSTANCE_ID")));
assertTrue(!execution1.getJobId().equals(execution2.getJobId()));
}
private void validateLaunchWithSkips(long jobExecutionId) {
// Step1: 9 input records, 1 skipped in process, 1 skipped in write =>
private void validateLaunchWithSkips(JobExecution jobExecution) {
// Step1: 9 input records, 1 skipped in read, 1 skipped in write =>
// 7 written to output
assertEquals(7, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "TRADE"));
// Step2: 7 input records, 1 skipped => 6 written to output
assertEquals(6, itemTrackingWriter.getItems().size());
// Step2: 7 input records, 1 skipped on process, 1 on write => 5 written
// to output
// System.err.println(simpleJdbcTemplate.queryForList("SELECT * FROM TRADE"));
assertEquals(5, simpleJdbcTemplate.queryForInt("SELECT COUNT(*) from TRADE where VERSION=?", 1));
// Both steps contained skips
assertEquals(2, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "ERROR_LOG"));
for (int i = 1; i <= 2; i++) {
assertEquals(1, simpleJdbcTemplate.queryForInt(
"SELECT Count(*) from ERROR_LOG where JOB_NAME = ? and STEP_NAME = ?", "skipJob", "step" + i));
}
assertEquals("2 records were skipped!", simpleJdbcTemplate.queryForObject(
"SELECT MESSAGE from ERROR_LOG where JOB_NAME = ? and STEP_NAME = ?", String.class, "skipJob", "step1"));
assertEquals("2 records were skipped!", simpleJdbcTemplate.queryForObject(
"SELECT MESSAGE from ERROR_LOG where JOB_NAME = ? and STEP_NAME = ?", String.class, "skipJob", "step2"));
assertEquals(new BigDecimal("340.45"), tradeWriter.getTotalPrice());
System.err.println(jobExecution.getExecutionContext());
assertEquals(new BigDecimal("340.45"), jobExecution.getExecutionContext().get(TradeWriter.TOTAL_AMOUNT_KEY));
Map<String, Object> step1Execution = getStepExecutionAsMap(jobExecutionId, "step1");
Map<String, Object> step1Execution = getStepExecutionAsMap(jobExecution.getId(), "step1");
assertEquals(new Long(4), step1Execution.get("COMMIT_COUNT"));
assertEquals(new Long(8), step1Execution.get("READ_COUNT"));
assertEquals(new Long(7), step1Execution.get("WRITE_COUNT"));
}
private void validateLaunchWithoutSkips(long jobExecutionId) {
private void validateLaunchWithoutSkips(JobExecution jobExecution) {
// Step1: 5 input records => 5 written to output
assertEquals(5, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "TRADE"));
// Step2: 5 input records => 5 written to output
assertEquals(5, itemTrackingWriter.getItems().size());
assertEquals(5, simpleJdbcTemplate.queryForInt("SELECT COUNT(*) from TRADE where VERSION=?", 1));
// Neither step contained skips
assertEquals(0, SimpleJdbcTestUtils.countRowsInTable(simpleJdbcTemplate, "ERROR_LOG"));
assertEquals(new BigDecimal("270.75"), tradeWriter.getTotalPrice());
}
assertEquals(new BigDecimal("270.75"), jobExecution.getExecutionContext().get(TradeWriter.TOTAL_AMOUNT_KEY));
private Map<String, Object> getJobExecutionAsMap(long jobExecutionId) {
return simpleJdbcTemplate.queryForMap("SELECT * from BATCH_JOB_EXECUTION where JOB_EXECUTION_ID = ?",
jobExecutionId);
}
private Map<String, Object> getStepExecutionAsMap(long jobExecutionId, String stepName) {
@@ -214,15 +210,20 @@ public class SkipSampleFunctionalTests {
public long launchJobWithIncrementer() {
try {
return this.jobOperator.startNextInstance("skipJob");
} catch (NoSuchJobException e) {
}
catch (NoSuchJobException e) {
throw new RuntimeException(e);
} catch (JobExecutionAlreadyRunningException e) {
}
catch (JobExecutionAlreadyRunningException e) {
throw new RuntimeException(e);
} catch (JobParametersNotFoundException e) {
}
catch (JobParametersNotFoundException e) {
throw new RuntimeException(e);
} catch (JobRestartException e) {
}
catch (JobRestartException e) {
throw new RuntimeException(e);
} catch (JobInstanceAlreadyCompleteException e) {
}
catch (JobInstanceAlreadyCompleteException e) {
throw new RuntimeException(e);
}
}

24
spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java
View File

@@ -21,19 +21,38 @@ import org.springframework.util.Assert;
public class ErrorLogTasklet implements Tasklet, StepExecutionListener {
protected final Log logger = LogFactory.getLog(getClass());
private SimpleJdbcTemplate simpleJdbcTemplate;
private String jobName;
private StepExecution stepExecution;
private String stepName;
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
Assert.notNull(this.stepName, "Step name not set. Either this class was not registered as a listener "
+ "or the key 'stepName' was not found in the Job's ExecutionContext.");
this.simpleJdbcTemplate.update("insert into ERROR_LOG values (?, ?, 'Some records were skipped!')", jobName,
stepName);
this.simpleJdbcTemplate.update("insert into ERROR_LOG values (?, ?, '"+getSkipCount()+" records were skipped!')",
jobName, stepName);
return RepeatStatus.FINISHED;
}
/**
* @return
*/
private int getSkipCount() {
if (stepExecution == null || stepName == null) {
return 0;
}
for (StepExecution execution : stepExecution.getJobExecution().getStepExecutions()) {
if (execution.getStepName().equals(stepName)) {
return execution.getSkipCount();
}
}
return 0;
}
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
@@ -41,6 +60,7 @@ public class ErrorLogTasklet implements Tasklet, StepExecutionListener {
public void beforeStep(StepExecution stepExecution) {
this.jobName = stepExecution.getJobExecution().getJobInstance().getJobName().trim();
this.stepName = (String) stepExecution.getJobExecution().getExecutionContext().get("stepName");
this.stepExecution = stepExecution;
stepExecution.getJobExecution().getExecutionContext().remove("stepName");
}

17
spring-batch-samples/src/test/java/org/springframework/batch/sample/common/SkipCheckingListener.java
View File

@@ -1,9 +1,14 @@
package org.springframework.batch.sample.common;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.AfterStep;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.core.annotation.OnSkipInProcess;
import org.springframework.batch.core.annotation.OnSkipInWrite;
import org.springframework.batch.sample.domain.trade.Trade;
/**
* @author Dan Garrette
@@ -11,6 +16,8 @@ import org.springframework.batch.core.annotation.BeforeStep;
*/
public class SkipCheckingListener {
private static final Log logger = LogFactory.getLog(SkipCheckingListener.class);
@AfterStep
public ExitStatus checkForSkips(StepExecution stepExecution) {
if (!stepExecution.getExitStatus().getExitCode().equals(ExitStatus.FAILED.getExitCode())
@@ -22,6 +29,16 @@ public class SkipCheckingListener {
}
}
@OnSkipInWrite
public void skipWrite(Trade trade, Throwable t) {
logger.debug("Skipped writing " + trade);
}
@OnSkipInProcess
public void skipProcess(Trade trade, Throwable t) {
logger.debug("Skipped processing " + trade);
}
@BeforeStep
public void saveStepName(StepExecution stepExecution) {
stepExecution.getExecutionContext().put("stepName", stepExecution.getStepName());

26
spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/ItemTrackingTradeItemWriter.java
View File

@@ -1,15 +1,27 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import javax.sql.DataSource;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.sample.domain.trade.Trade;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
public class ItemTrackingTradeItemWriter implements ItemWriter<Trade> {
private List<Trade> items = new ArrayList<Trade>();
private String writeFailureISIN;
private SimpleJdbcTemplate jdbcTemplate;
public void setDataSource(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
public void setWriteFailureISIN(String writeFailureISIN) {
this.writeFailureISIN = writeFailureISIN;
}
@@ -21,18 +33,22 @@ public class ItemTrackingTradeItemWriter implements ItemWriter<Trade> {
public List<Trade> getItems() {
return items;
}
public void clearItems(){
public void clearItems() {
this.items.clear();
}
public void write(List<? extends Trade> items) throws Exception {
List<Trade> newItems = new ArrayList<Trade>();
for(Trade t : items){
if (t.getIsin().equals(this.writeFailureISIN)){
throw new RuntimeException("write failed");
for (Trade t : items) {
if (t.getIsin().equals(this.writeFailureISIN)) {
throw new IOException("write failed");
}
newItems.add(t);
if (jdbcTemplate != null) {
jdbcTemplate.update("UPDATE TRADE set VERSION=? where ID=? and version=?", t.getVersion() + 1, t
.getId(), t.getVersion());
}
}
this.items.addAll(newItems);
}