From cdf94a714887635bbf5bd20e601a624c36096f37 Mon Sep 17 00:00:00 2001 From: dsyer Date: Fri, 4 Sep 2009 10:36:45 +0000 Subject: [PATCH] OPEN - issue BATCH-1383: onSkipInProcess is not called http://jira.springframework.org/browse/BATCH-1383 Tidied up the skip sample. Could not reproduce issue. --- .../batch/sample/domain/trade/Trade.java | 27 +++- .../domain/trade/internal/JdbcTradeDao.java | 8 +- .../domain/trade/internal/TradeRowMapper.java | 5 +- .../domain/trade/internal/TradeWriter.java | 2 +- .../src/main/resources/jobs/skipSampleJob.xml | 118 ++++++++++-------- .../skipSample-job-launcher-context.xml | 20 --- .../sample/SkipSampleFunctionalTests.java | 79 ++++++------ .../batch/sample/common/ErrorLogTasklet.java | 24 +++- .../sample/common/SkipCheckingListener.java | 17 +++ .../internal/ItemTrackingTradeItemWriter.java | 26 +++- 10 files changed, 203 insertions(+), 123 deletions(-) diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/Trade.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/Trade.java index d21cd53dc..4e171496f 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/Trade.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/Trade.java @@ -31,6 +31,8 @@ public class Trade implements Serializable { private long quantity = 0; private BigDecimal price = new BigDecimal(0); private String customer = ""; + private Long id; + private long version = 0; public Trade() { } @@ -42,7 +44,30 @@ public class Trade implements Serializable { this.customer = customer; } - public void setCustomer(String customer) { + /** + * @param id + */ + public Trade(long id) { + this.id = id; + } + + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + public long getVersion() { + return version; + } + + public void setVersion(long version) { + this.version = version; + } + + public void setCustomer(String customer) { this.customer = customer; } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/JdbcTradeDao.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/JdbcTradeDao.java index a900b8bc0..fd5569963 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/JdbcTradeDao.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/JdbcTradeDao.java @@ -16,6 +16,8 @@ package org.springframework.batch.sample.domain.trade.internal; +import javax.sql.DataSource; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.sample.domain.trade.Trade; @@ -23,8 +25,6 @@ import org.springframework.batch.sample.domain.trade.TradeDao; import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; -import javax.sql.DataSource; - /** * Writes a Trade object to a database @@ -36,7 +36,7 @@ public class JdbcTradeDao implements TradeDao { /** * template for inserting a row */ - private static final String INSERT_TRADE_RECORD = "INSERT INTO trade (id, isin, quantity, price, customer) VALUES (?, ?, ? ,?, ?)"; + private static final String INSERT_TRADE_RECORD = "INSERT INTO trade (id, version, isin, quantity, price, customer) VALUES (?, 0, ?, ? ,?, ?)"; /** * handles the processing of sql query @@ -44,7 +44,7 @@ public class JdbcTradeDao implements TradeDao { private SimpleJdbcTemplate simpleJdbcTemplate; /** - * database is not expected to be setup for autoincrementation + * database is not expected to be setup for autoincrement */ private DataFieldMaxValueIncrementer incrementer; diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeRowMapper.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeRowMapper.java index 9e372a8ca..8ca36055f 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeRowMapper.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeRowMapper.java @@ -28,14 +28,17 @@ public class TradeRowMapper implements RowMapper { public static final int QUANTITY_COLUMN = 2; public static final int PRICE_COLUMN = 3; public static final int CUSTOMER_COLUMN = 4; + public static final int ID_COLUMN = 5; + public static final int VERSION_COLUMN = 6; public Object mapRow(ResultSet rs, int rowNum) throws SQLException { - Trade trade = new Trade(); + Trade trade = new Trade(rs.getLong(ID_COLUMN)); trade.setIsin(rs.getString(ISIN_COLUMN)); trade.setQuantity(rs.getLong(QUANTITY_COLUMN)); trade.setPrice(rs.getBigDecimal(PRICE_COLUMN)); trade.setCustomer(rs.getString(CUSTOMER_COLUMN)); + trade.setVersion(rs.getLong(VERSION_COLUMN)); return trade; } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeWriter.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeWriter.java index 6f19dda94..d85a39bd6 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeWriter.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/TradeWriter.java @@ -40,7 +40,7 @@ public class TradeWriter extends ItemStreamSupport implements ItemWriter private static Log log = LogFactory.getLog(TradeWriter.class); - private static final String TOTAL_AMOUNT_KEY = "TOTAL_AMOUNT"; + public static final String TOTAL_AMOUNT_KEY = "TOTAL_AMOUNT"; private TradeDao dao; diff --git a/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml index 3d157b7a4..6aa5d0e87 100644 --- a/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml @@ -1,65 +1,53 @@ - + http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-2.0.xsd"> + - + - - + + - - + + - + - + + + + + + + + + + - + - + - + - - - - - - - - - - - - - - - - - - - - - - - - - + + + @@ -69,14 +57,17 @@ - + + + stepName + + + - - @@ -94,22 +85,49 @@ - - + + + + + + + + + + + + + + + + + + + + customer6 + + + + + + + + + - + - + - - - + + + diff --git a/spring-batch-samples/src/main/resources/skipSample-job-launcher-context.xml b/spring-batch-samples/src/main/resources/skipSample-job-launcher-context.xml index 94dc97796..e18d2513f 100644 --- a/spring-batch-samples/src/main/resources/skipSample-job-launcher-context.xml +++ b/spring-batch-samples/src/main/resources/skipSample-job-launcher-context.xml @@ -40,24 +40,4 @@ - - - - - - - - - - - - - - customer6 - - - - - - diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java index f51e8f586..e441e50fe 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java @@ -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 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 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 step1Execution = getStepExecutionAsMap(jobExecutionId, "step1"); + Map 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 getJobExecutionAsMap(long jobExecutionId) { - return simpleJdbcTemplate.queryForMap("SELECT * from BATCH_JOB_EXECUTION where JOB_EXECUTION_ID = ?", - jobExecutionId); } private Map 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); } } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java index 34c35abc3..039555beb 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java @@ -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"); } diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/SkipCheckingListener.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/SkipCheckingListener.java index ada3d8276..829fce048 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/SkipCheckingListener.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/common/SkipCheckingListener.java @@ -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()); diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/ItemTrackingTradeItemWriter.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/ItemTrackingTradeItemWriter.java index f1b6db8a9..78eb71cf5 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/ItemTrackingTradeItemWriter.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/domain/trade/internal/ItemTrackingTradeItemWriter.java @@ -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 { + private List items = new ArrayList(); + 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 { public List getItems() { return items; } - - public void clearItems(){ + + public void clearItems() { this.items.clear(); } public void write(List items) throws Exception { List newItems = new ArrayList(); - 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); }