From 5d5ce07c8aae13ddb16424af36e23ef0f8a22d48 Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 28 Jan 2010 11:35:41 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1493: Step listeners detected and invoked twice RESOLVED - issue BATCH-1494: FetchSize not accessible in HibernateCursorItemReader --- .../core/step/item/SimpleStepFactoryBean.java | 20 +++++++--- .../step/item/SimpleStepFactoryBeanTests.java | 40 +++++++++++++++++++ .../database/AbstractCursorItemReader.java | 5 ++- .../database/HibernateCursorItemReader.java | 22 ++++++++-- .../database/HibernateItemReaderHelper.java | 31 +++++--------- .../database/HibernatePagingItemReader.java | 20 +++++++++- .../HibernateCursorItemReaderCommonTests.java | 1 + 7 files changed, 108 insertions(+), 31 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 18ebbc929..d674cc84a 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,6 +15,9 @@ */ package org.springframework.batch.core.step.item; +import java.util.Arrays; +import java.util.List; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ChunkListener; @@ -600,22 +603,29 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { */ private void registerItemListeners(SimpleChunkProvider chunkProvider, SimpleChunkProcessor chunkProcessor) { + StepListener[] listeners = getListeners(); + // explicitly set item listeners - chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), + chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, ItemReadListener.class)); - chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), + chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, SkipListener.class)); - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), + chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, ItemProcessListener.class)); - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), + chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, ItemWriteListener.class)); - chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), + chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(listeners, SkipListener.class)); + List listofListeners = Arrays.asList(listeners); // auto-register reader, processor and writer for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) { + if (listofListeners.contains(itemHandler)) { + continue; + } + if (StepListenerFactoryBean.isListener(itemHandler)) { StepListener listener = StepListenerFactoryBean.getListener(itemHandler); if (listener instanceof SkipListener) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java index 44bf34adb..b8c6b510a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java @@ -385,6 +385,46 @@ public class SimpleStepFactoryBeanTests { } } + @Test + public void testAutoRegisterItemListenersNoDoubleCounting() throws Exception { + + SimpleStepFactoryBean factory = getStepFactory(new String[] { "foo", "bar", "spam" }); + + final List listenerCalls = new ArrayList(); + + class TestItemListenerWriter implements ItemWriter, ItemWriteListener { + public void write(List items) throws Exception { + } + + public void afterWrite(List items) { + listenerCalls.add("write"); + } + + public void beforeWrite(List items) { + } + + public void onWriteError(Exception exception, List items) { + } + + } + + TestItemListenerWriter itemWriter = new TestItemListenerWriter(); + factory.setListeners(new StepListener[] {itemWriter}); + factory.setItemWriter(itemWriter); + + Step step = (Step) factory.getObject(); + + job.setSteps(Collections.singletonList(step)); + + JobExecution jobExecution = repository.createJobExecution(job.getName(), new JobParameters()); + + job.execute(jobExecution); + + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + assertEquals("[write, write, write]", listenerCalls.toString()); + + } + private SimpleStepFactoryBean getStepFactory(String... args) throws Exception { SimpleStepFactoryBean factory = new SimpleStepFactoryBean(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java index f07cd0ade..3da1779b7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractCursorItemReader.java @@ -391,9 +391,12 @@ public abstract class AbstractCursorItemReader extends AbstractItemCountingIt /** * Execute the statement to open the cursor. */ - protected void doOpen() throws Exception { + @Override + protected final void doOpen() throws Exception { + Assert.state(!initialized, "Stream is already initialized. Close before re-opening."); Assert.isNull(rs, "ResultSet still open! Close before re-opening."); + initializeConnection(); openCursor(con); initialized = true; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java index ad6c234dc..d7bb24124 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java @@ -68,7 +68,12 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream private boolean initialized = false; + private int fetchSize; + + private Map parameterValues; + public void afterPropertiesSet() throws Exception { + Assert.state(fetchSize >= 0, "fetchSize must not be negative"); helper.afterPropertiesSet(); } @@ -78,7 +83,7 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream * @param parameterValues the parameter values to set */ public void setParameterValues(Map parameterValues) { - helper.setParameterValues(parameterValues); + this.parameterValues = parameterValues; } /** @@ -93,6 +98,16 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream helper.setQueryName(queryName); } + /** + * Fetch size used internally by Hibernate to limit amount of data fetched + * from database per round trip. + * + * @param fetchSize the fetch size to pass down to Hibernate + */ + public void setFetchSize(int fetchSize) { + this.fetchSize = fetchSize; + } + /** * A query provider. Either this or the {{@link #setQueryString(String) * query string} or the {{@link #setQueryName(String) query name} should be @@ -165,7 +180,7 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream */ protected void doOpen() throws Exception { Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); - cursor = helper.getForwardOnlyCursor(); + cursor = helper.getForwardOnlyCursor(fetchSize, parameterValues); initialized = true; } @@ -193,7 +208,8 @@ public class HibernateCursorItemReader extends AbstractItemCountingItemStream */ @Override protected void jumpToItem(int itemIndex) throws Exception { - helper.jumpToItem(cursor, itemIndex); + int flushSize = Math.max(fetchSize, 100); + helper.jumpToItem(cursor, itemIndex, flushSize); } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java index 4132d8d08..97d4c4afe 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateItemReaderHelper.java @@ -46,14 +46,10 @@ public class HibernateItemReaderHelper implements InitializingBean { private String queryName = ""; - private Map parameterValues; - private HibernateQueryProvider queryProvider; private boolean useStatelessSession = true; - private int fetchSize = 0; - private StatelessSession statelessSession; private Session statefulSession; @@ -92,15 +88,6 @@ public class HibernateItemReaderHelper implements InitializingBean { this.useStatelessSession = useStatelessSession; } - /** - * The parameter values to apply to a query (map of name:value). - * - * @param parameterValues the parameter values to set - */ - public void setParameterValues(Map parameterValues) { - this.parameterValues = parameterValues; - } - /** * @param sessionFactory hibernate session factory */ @@ -111,7 +98,6 @@ public class HibernateItemReaderHelper implements InitializingBean { public void afterPropertiesSet() throws Exception { Assert.state(sessionFactory != null, "A SessionFactory must be provided"); - Assert.state(fetchSize >= 0, "fetchSize must not be negative"); if (queryProvider == null) { Assert.notNull(sessionFactory, "session factory must be set"); @@ -129,9 +115,12 @@ public class HibernateItemReaderHelper implements InitializingBean { /** * Get a cursor over all of the results, with the forward-only flag set. * + * @param fetchSize the fetch size to use retrieving the results + * @param parameterValues the parameter values to use (or null if none). + * * @return a forward-only {@link ScrollableResults} */ - public ScrollableResults getForwardOnlyCursor() { + public ScrollableResults getForwardOnlyCursor(int fetchSize, Map parameterValues) { Query query = createQuery(); if (parameterValues != null) { query.setProperties(parameterValues); @@ -142,7 +131,7 @@ public class HibernateItemReaderHelper implements InitializingBean { /** * Open appropriate type of hibernate session and create the query. */ - private Query createQuery() { + public Query createQuery() { if (useStatelessSession) { statelessSession = sessionFactory.openStatelessSession(); @@ -183,11 +172,10 @@ public class HibernateItemReaderHelper implements InitializingBean { * * @param cursor the results to scroll over */ - public void jumpToItem(ScrollableResults cursor, int itemIndex) { - int flushSize = Math.max(fetchSize, 100); + public void jumpToItem(ScrollableResults cursor, int itemIndex, int flushInterval) { for (int i = 0; i < itemIndex; i++) { cursor.next(); - if (i % flushSize == 0 && !useStatelessSession) { + if (i % flushInterval == 0 && !useStatelessSession) { statefulSession.clear(); // Clears in-memory cache } } @@ -211,9 +199,12 @@ public class HibernateItemReaderHelper implements InitializingBean { * * @param page the page to read (starting at 0) * @param pageSize the size of the page or maximum number of items to read + * @param fetchSize the fetch size to use + * @param parameterValues the parameter values to use (if any, otherwise + * null) * @return a collection of items */ - public Collection readPage(int page, int pageSize) { + public Collection readPage(int page, int pageSize, int fetchSize, Map parameterValues) { clear(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java index 31ca832a5..4e23457d7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernatePagingItemReader.java @@ -26,6 +26,7 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.database.orm.HibernateQueryProvider; import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; import org.springframework.util.ClassUtils; /** @@ -61,6 +62,10 @@ public class HibernatePagingItemReader extends AbstractPagingItemReader im private HibernateItemReaderHelper helper = new HibernateItemReaderHelper(); + private Map parameterValues; + + private int fetchSize; + public HibernatePagingItemReader() { setName(ClassUtils.getShortName(HibernatePagingItemReader.class)); } @@ -71,7 +76,7 @@ public class HibernatePagingItemReader extends AbstractPagingItemReader im * @param parameterValues the parameter values to set */ public void setParameterValues(Map parameterValues) { - helper.setParameterValues(parameterValues); + this.parameterValues = parameterValues; } /** @@ -86,6 +91,16 @@ public class HibernatePagingItemReader extends AbstractPagingItemReader im helper.setQueryName(queryName); } + /** + * Fetch size used internally by Hibernate to limit amount of data fetched + * from database per round trip. + * + * @param fetchSize the fetch size to pass down to Hibernate + */ + public void setFetchSize(int fetchSize) { + this.fetchSize = fetchSize; + } + /** * A query provider. Either this or the {{@link #setQueryString(String) * query string} or the {{@link #setQueryName(String) query name} should be @@ -130,6 +145,7 @@ public class HibernatePagingItemReader extends AbstractPagingItemReader im public void afterPropertiesSet() throws Exception { super.afterPropertiesSet(); + Assert.state(fetchSize >= 0, "fetchSize must not be negative"); helper.afterPropertiesSet(); } @@ -148,7 +164,7 @@ public class HibernatePagingItemReader extends AbstractPagingItemReader im results.clear(); } - results.addAll(helper.readPage(getPage(), getPageSize())); + results.addAll(helper.readPage(getPage(), getPageSize(), fetchSize, parameterValues)); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java index 8ffc6a488..a3fc1f197 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/HibernateCursorItemReaderCommonTests.java @@ -23,6 +23,7 @@ public class HibernateCursorItemReaderCommonTests extends AbstractDatabaseItemSt reader.setQueryString(hsqlQuery); reader.setSessionFactory(sessionFactory); reader.setUseStatelessSession(true); + reader.setFetchSize(10); reader.afterPropertiesSet(); reader.setSaveState(true);