diff --git a/spring-batch-infrastructure/pom.xml b/spring-batch-infrastructure/pom.xml index de3bedb9b..8f8442648 100644 --- a/spring-batch-infrastructure/pom.xml +++ b/spring-batch-infrastructure/pom.xml @@ -80,6 +80,7 @@ org.springframework.xml.*;version="[1.0.0, 2.0.0)";resolution:=optional, org.springframework.batch.*, org.springframework.*;version="[2.0.0, 3.0.0)", + javax.persistence.*, javax.sql.*, javax.xml.namespace.*, javax.xml.transform.* diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaAwareItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaAwareItemWriter.java new file mode 100644 index 000000000..8ecadd2ae --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaAwareItemWriter.java @@ -0,0 +1,109 @@ +package org.springframework.batch.item.database; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.ClearFailedException; +import org.springframework.util.Assert; +import org.springframework.orm.jpa.EntityManagerFactoryUtils; +import org.springframework.dao.DataAccessResourceFailureException; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +/** + * {@link org.springframework.batch.item.ItemWriter} that is aware of the JPA EntityManagerFactory and can + * take some responsibilities to do with chunk boundaries away from a less smart + * {@link org.springframework.batch.item.ItemWriter} (the delegate). A delegate is required, and will be used + * to do the actual writing of the item.
+ * + * It is required that {@link #write(Object)} is called inside a transaction, + * and that {@link #flush()} is then subsequently called before the transaction + * commits, or {@link #clear()} before it rolls back.
+ * + * The reader must be configured with an {@link javax.persistence.EntityManagerFactory} that is capable + * of participating in Spring managed transactions. + * + * The writer is thread safe after its properties are set (normal singleton + * behaviour), so it can be used to write in multiple concurrent transactions. + * Note, however, that the set of failed items is stored in a collection + * internally, and this collection is never cleared, so it is not a great idea + * to go on using the writer indefinitely. Normally it would be used for the + * duration of a batch job and then discarded. + * + * @author Dave Syer + * @author Thomas Risberg + * + */ +public class JpaAwareItemWriter extends AbstractTransactionalResourceItemWriter implements InitializingBean { + + /** + * Key for items processed in the current transaction {@link org.springframework.batch.repeat.RepeatContext}. + */ + private static final String ITEMS_PROCESSED = JpaAwareItemWriter.class.getName() + ".ITEMS_PROCESSED"; + + private ItemWriter delegate; + + private EntityManagerFactory entityManagerFactory; + + /** + * Public setter for the {@link org.springframework.batch.item.ItemWriter} property. + * + * @param delegate the delegate to set + */ + public void setDelegate(ItemWriter delegate) { + this.delegate = delegate; + } + + /** + * Set the EntityManager to be used internally. + * + * @param entityManagerFactory the entityManagerFactory to set + */ + public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) { + this.entityManagerFactory = entityManagerFactory; + } + + /** + * Check mandatory properties - there must be a delegate and entityManagerFactory. + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(delegate, "An ItemWriter to be used as a delegate is required."); + Assert.notNull(entityManagerFactory, "An EntityManagerFactory is required"); + } + + /** + * Delegate to subclass and flush the EntityManager. + */ + protected void doFlush() { + delegate.flush(); + EntityManager entityManager = + EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory); + if (entityManager == null) { + throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager"); + } + entityManager.flush(); + entityManager.clear(); + } + + /** + * Call the delegate clear() method, and then clear the EntityManager. + */ + protected void doClear() throws ClearFailedException { + delegate.clear(); + EntityManager entityManager = + EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory); + if (entityManager == null) { + throw new DataAccessResourceFailureException("Unable to obtain a transactional EntityManager"); + } + entityManager.clear(); + } + + protected String getResourceKey() { + return ITEMS_PROCESSED; + } + + protected void doWrite(T item) throws Exception { + delegate.write(item); + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java index 6844c3629..bb82b5849 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java @@ -1,3 +1,19 @@ +/* + * Copyright 2006-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package org.springframework.batch.item.database; import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; @@ -24,8 +40,14 @@ import java.util.List; * are requested when needed as {@link #read()} method is called, returning * an object corresponding to current position. * + * The performance of the paging depends on the JPA implementation and its use of database specific + * features to limit the number of returned rows. + * + * Setting a fairly large page size and using a commit interval that matches the page size should provide + * better performance. + * * The reader must be configured with an {@link javax.persistence.EntityManagerFactory} that is capable - * of participating in SPring managed transactions. + * of participating in Spring managed transactions. * * The implementation is *not* thread-safe. * diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaAwareItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaAwareItemWriterTests.java new file mode 100644 index 000000000..cb89c162b --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaAwareItemWriterTests.java @@ -0,0 +1,178 @@ +/* + * Copyright 2006-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.item.database; + +import static org.easymock.EasyMock.*; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.orm.jpa.EntityManagerHolder; +import org.springframework.batch.item.ItemWriter; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import javax.persistence.EntityManagerFactory; +import javax.persistence.EntityManager; +import java.util.List; +import java.util.ArrayList; + +/** + * @author Thomas Risberg + * + */ +public class JpaAwareItemWriterTests { + + JpaAwareItemWriter writer = new JpaAwareItemWriter(); + + ItemWriter delegate; + + EntityManagerFactory emf; + + final List list = new ArrayList(); + + @Before + @SuppressWarnings({"unchecked"}) + public void setUp() throws Exception { + if (TransactionSynchronizationManager.isSynchronizationActive()) { + TransactionSynchronizationManager.clearSynchronization(); + } + delegate = createMock("delegate", ItemWriter.class); + writer.setDelegate(delegate); + emf = createMock("emf", EntityManagerFactory.class); + writer.setEntityManagerFactory(emf); + } + + @Test + public void testAfterPropertiesSet() throws Exception { + writer = new JpaAwareItemWriter(); + try { + writer.afterPropertiesSet(); + fail("Expected IllegalArgumentException"); + } catch (IllegalArgumentException e) { + // expected + assertTrue("Wrong message for exception: " + e.getMessage(), e + .getMessage().indexOf("delegate") >= 0); + } + writer.setDelegate(delegate); + try { + writer.afterPropertiesSet(); + fail("Expected IllegalArgumentException"); + } catch (IllegalArgumentException e) { + // expected + assertTrue("Wrong message for exception: " + e.getMessage(), e + .getMessage().indexOf("EntityManagerFactory") >= 0); + } + } + + @Test + public void testWrite() throws Exception { + delegate.write("foo"); + replay(delegate); + writer.write("foo"); + verify(delegate); + } + + @Test + public void testFlushWithFailure() throws Exception{ + final RuntimeException ex = new RuntimeException("bar"); + EntityManager em = createMock("em", EntityManager.class); + em.joinTransaction(); + em.flush(); + expectLastCall().andThrow(ex); + replay(em); + expect(emf.createEntityManager()).andReturn(em); + replay(emf); + TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em)); + delegate.flush(); + replay(delegate); + try { + writer.flush(); + fail("Expected RuntimeException"); + } catch (RuntimeException e) { + assertEquals("bar", e.getMessage()); + } + TransactionSynchronizationManager.unbindResource(emf); + } + + @Test + public void testWriteAndFlushWithFailure() throws Exception { + final RuntimeException ex = new RuntimeException("bar"); + EntityManager em = createMock("em", EntityManager.class); + em.flush(); + expectLastCall().andThrow(ex); + em.flush(); + em.clear(); + replay(em); + replay(emf); + TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em)); + delegate.write("foo"); + delegate.flush(); + delegate.write("spam"); + delegate.flush(); + replay(delegate); + + writer.write("foo"); + try { + writer.flush(); + fail("Expected RuntimeException"); + } catch (RuntimeException e) { + assertEquals("bar", e.getMessage()); + } + + writer.write("spam"); + writer.flush(); + + verify(delegate); + verify(em); + TransactionSynchronizationManager.unbindResource(emf); + } + + @Test + public void testFlush() throws Exception{ + EntityManager em = createMock("em", EntityManager.class); + em.flush(); + em.clear(); + replay(em); + replay(emf); + TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em)); + delegate.flush(); + replay(delegate); + + writer.flush(); + + verify(delegate); + verify(em); + TransactionSynchronizationManager.unbindResource(emf); + } + + @Test + public void testClear() throws Exception{ + EntityManager em = createMock("em", EntityManager.class); + em.clear(); + replay(em); + replay(emf); + TransactionSynchronizationManager.bindResource(emf, new EntityManagerHolder(em)); + delegate.clear(); + replay(delegate); + + writer.clear(); + + verify(delegate); + verify(em); + TransactionSynchronizationManager.unbindResource(emf); + } +}