OPEN - issue BATCH-788: Remove flush/clear from ItemWriter

Removed from transaction buffering writers (Jpa etc.)

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractTransactionalResourceItemWriter.java

@@ -1,176 +0,0 @@
-/*
- * Copyright 2006-2007 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 java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.springframework.batch.item.ClearFailedException;
-import org.springframework.batch.item.FlushFailedException;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.repeat.RepeatContext;
-import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
-import org.springframework.util.Assert;
-
-/**
- * Stores items in transactional resource and flushes aggressively in case of
- * failure. This is useful for batch update writers which need to identify the
- * failed item after failed flush.
- * 
- * @see BatchSqlUpdateItemWriter
- * @see HibernateAwareItemWriter
- * 
- * @author Dave Syer
- * @author Robert Kasanicky
- */
-public abstract class AbstractTransactionalResourceItemWriter<T> implements ItemWriter<T> {
-
-	private Set<T> failed = new HashSet<T>();
-
-	/**
-	 * Flushing delegated to subclass surrounded by binding and unbinding of
-	 * transactional resources.
-	 * 
-	 * @see org.springframework.batch.item.ItemWriter#flush()
-	 */
-	public final void flush() throws FlushFailedException {
-		bindTransactionResources();
-		try {
-			doFlush();
-		}
-		catch (RuntimeException e) {
-			synchronized (failed) {
-				failed.addAll(getProcessed());
-			}
-			// This used to contain a call to onError, however, I think this
-			// should be handled within the step.
-			throw e;
-		}
-		finally {
-			unbindTransactionResources();
-		}
-	}
-
-	/**
-	 * Delegate to subclass to actually do the writing, but flushes aggressively
-	 * if the item was previously part of a failed chunk.
-	 * 
-	 * @throws Exception
-	 * 
-	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
-	 */
-	public final void write(List<? extends T> output) throws Exception {
-		bindTransactionResources();
-		getProcessed().addAll(output);
-		doWrite(output);
-		flushIfNecessary(output);
-	}
-
-	/**
-	 * Delegate to subclass and unbind transactional resources, effectively
-	 * clearing the item buffer.
-	 */
-	public final void clear() throws ClearFailedException {
-		try {
-			doClear();
-		}
-		finally {
-			unbindTransactionResources();
-		}
-	}
-
-	/**
-	 * Callback method of {@link #flush()}.
-	 */
-	protected abstract void doFlush() throws FlushFailedException;
-
-	/**
-	 * Callback method of {@link #clear()}.
-	 */
-	protected abstract void doClear() throws ClearFailedException;
-
-	/**
-	 * Callback method of {@link #write(List)}.
-	 */
-	protected abstract void doWrite(List<? extends T> output) throws Exception;
-
-	/**
-	 * @return Key for items processed in the current transaction
-	 * {@link RepeatContext}.
-	 */
-	protected abstract String getResourceKey();
-
-	private void flushIfNecessary(List<? extends T> outputs) {
-		Set<T> flush = new HashSet<T>();
-		synchronized (failed) {
-			for (T output : outputs) {
-				if (failed.contains(output)) {
-					flush.add(output);				
-				}
-			}
-		}
-		if (!flush.isEmpty()) {
-			// Force early completion to commit aggressively if we encounter a
-			// failed item (from a failed chunk but we don't know which one was
-			// the problem).
-			RepeatSynchronizationManager.setCompleteOnly();
-			// Remove the failed item from the cache, otherwise it could grow
-			// unnecessarily large.
-			failed.removeAll(flush);
-			// Flush now, so that if there is a failure this record can be
-			// skipped.
-			flush();
-		}
-
-	}
-
-	/**
-	 * Set up the {@link RepeatContext} as a transaction resource.
-	 * 
-	 * @param context the context to set
-	 */
-	private void bindTransactionResources() {
-		if (TransactionSynchronizationManager.hasResource(getResourceKey())) {
-			return;
-		}
-		TransactionSynchronizationManager.bindResource(getResourceKey(), new HashSet<Object>());
-	}
-
-	/**
-	 * Remove the transaction resource associated with this context.
-	 */
-	private void unbindTransactionResources() {
-		if (!TransactionSynchronizationManager.hasResource(getResourceKey())) {
-			return;
-		}
-		TransactionSynchronizationManager.unbindResource(getResourceKey());
-	}
-
-	/**
-	 * Accessor for the list of processed items in this transaction.
-	 * 
-	 * @return the processed
-	 */
-	@SuppressWarnings("unchecked")
-	protected Set<T> getProcessed() {
-		Assert.state(TransactionSynchronizationManager.hasResource(getResourceKey()),
-				"Processed items not bound to transaction.");
-		Set<T> processed = (Set<T>) TransactionSynchronizationManager.getResource(getResourceKey());
-		return processed;
-	}
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/BatchSqlUpdateItemWriter.java

@@ -17,12 +17,10 @@ package org.springframework.batch.item.database;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
-import java.util.ArrayList;
 import java.util.List;
 
-import org.springframework.batch.item.ClearFailedException;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.batch.item.support.AbstractItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.EmptyResultDataAccessException;
@@ -42,9 +40,7 @@ import org.springframework.util.Assert;
  * {@link ItemPreparedStatementSetter}, which is responsible for mapping the
  * item to a PreparedStatement.<br/>
  * 
- * It is expected that {@link #write(List)} is called inside a transaction,
- * and that {@link #flush()} is then subsequently called before the transaction
- * commits, or {@link #clear()} before it rolls back.<br/>
+ * It is expected that {@link #write(List)} is called inside a transaction.<br/>
  * 
  * The writer is thread safe after its properties are set (normal singleton
  * behaviour), so it can be used to write in multiple concurrent transactions.
@@ -56,12 +52,7 @@ import org.springframework.util.Assert;
  * @author Dave Syer
  * 
  */
-public class BatchSqlUpdateItemWriter<T> extends AbstractTransactionalResourceItemWriter<T> implements InitializingBean {
-
-	/**
-	 * Key for items processed in the current transaction {@link RepeatContext}.
-	 */
-	private static final String ITEMS_PROCESSED = BatchSqlUpdateItemWriter.class.getName() + ".ITEMS_PROCESSED";
+public class BatchSqlUpdateItemWriter<T> extends AbstractItemWriter<T> implements InitializingBean {
 
 	private JdbcOperations jdbcTemplate;
 
@@ -114,22 +105,18 @@ public class BatchSqlUpdateItemWriter<T> extends AbstractTransactionalResourceIt
 		Assert.notNull(jdbcTemplate, "BatchSqlUpdateItemWriter requires an data source.");
 		Assert.notNull(preparedStatementSetter, "BatchSqlUpdateItemWriter requires a ItemPreparedStatementSetter");
 	}
-
-	/**
-	 * Create and execute batch prepared statement.
-	 * @throws EmptyResultDataAccessException if any of the items does not cause
-	 * an update
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
 	 */
-	protected void doFlush() throws EmptyResultDataAccessException {
+	public void write(final List<? extends T> items) throws Exception {
 
-		final List<T> processed = new ArrayList<T>(getProcessed());
-
-		if (!processed.isEmpty()) {
+		if (!items.isEmpty()) {
 
 			int[] values = (int[]) jdbcTemplate.execute(sql, new PreparedStatementCallback() {
 				public Object doInPreparedStatement(PreparedStatement ps) throws SQLException, DataAccessException {
 
-					for (T item : processed) {
+					for (T item : items) {
 						preparedStatementSetter.setValues(item, ps);
 						ps.addBatch();
 					}
@@ -142,7 +129,7 @@ public class BatchSqlUpdateItemWriter<T> extends AbstractTransactionalResourceIt
 					int value = values[i];
 					if (value == 0) {
 						throw new EmptyResultDataAccessException("Item " + i + " of " + values.length
-								+ " did not update any rows: [" + processed.get(i) + "]", 1);
+								+ " did not update any rows: [" + items.get(i) + "]", 1);
 					}
 				}
 			}
@@ -151,20 +138,4 @@ public class BatchSqlUpdateItemWriter<T> extends AbstractTransactionalResourceIt
 
 	}
 
-	protected String getResourceKey() {
-		return ITEMS_PROCESSED;
-	}
-
-	/**
-	 * No-op.
-	 */
-	protected void doWrite(List<? extends T> item) {
-	}
-
-	/**
-	 * No-op.
-	 */
-	protected void doClear() throws ClearFailedException {
-	}
-
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateAwareItemWriter.java

@@ -18,9 +18,8 @@ package org.springframework.batch.item.database;
 import java.util.List;
 
 import org.hibernate.SessionFactory;
-import org.springframework.batch.item.ClearFailedException;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.batch.item.support.AbstractItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.orm.hibernate3.HibernateOperations;
 import org.springframework.orm.hibernate3.HibernateTemplate;
@@ -32,8 +31,8 @@ import org.springframework.util.Assert;
  * {@link ItemWriter} (the delegate). A delegate is required, and will be used
  * to do the actual writing of the item.<br/>
  * 
- * It is expected that {@link #write(List)} is called inside a transaction,
- * and that {@link #flush()} is then subsequently called before the transaction
+ * It is expected that {@link #write(List)} is called inside a transaction, and
+ * that {@link #flush()} is then subsequently called before the transaction
  * commits, or {@link #clear()} before it rolls back.<br/>
  * 
  * The writer is thread safe after its properties are set (normal singleton
@@ -46,12 +45,7 @@ import org.springframework.util.Assert;
  * @author Dave Syer
  * 
  */
-public class HibernateAwareItemWriter<T> extends AbstractTransactionalResourceItemWriter<T> implements InitializingBean {
-
-	/**
-	 * Key for items processed in the current transaction {@link RepeatContext}.
-	 */
-	private static final String ITEMS_PROCESSED = HibernateAwareItemWriter.class.getName() + ".ITEMS_PROCESSED";
+public class HibernateAwareItemWriter<T> extends AbstractItemWriter<T> implements InitializingBean {
 
 	private ItemWriter<? super T> delegate;
 
@@ -86,7 +80,8 @@ public class HibernateAwareItemWriter<T> extends AbstractTransactionalResourceIt
 	}
 
 	/**
-	 * Check mandatory properties - there must be a delegate and hibernateTemplate.
+	 * Check mandatory properties - there must be a delegate and
+	 * hibernateTemplate.
 	 */
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(delegate, "HibernateAwareItemWriter requires an ItemWriter as a delegate.");
@@ -94,30 +89,20 @@ public class HibernateAwareItemWriter<T> extends AbstractTransactionalResourceIt
 	}
 
 	/**
-	 * Delegate to subclass and flush the hibernate session.
+	 * Delegate the writing and then flush and clear te hibernate session.
+	 * 
+	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
 	 */
-	protected void doFlush() {
-		delegate.flush();
-		hibernateTemplate.flush();
-		// This should happen when the transaction commits anyway, but to be
-		// sure...
-		hibernateTemplate.clear();
-	}
-
-	/**
-	 * Call the delegate clear() method, and then clear the hibernate session.
-	 */
-	protected void doClear() throws ClearFailedException {
-		delegate.clear();
-		hibernateTemplate.clear();
-	}
-
-	protected String getResourceKey() {
-		return ITEMS_PROCESSED;
-	}
-
-	protected void doWrite(List<? extends T> item) throws Exception {
-		delegate.write(item);
+	public void write(List<? extends T> items) throws Exception {
+		delegate.write(items);
+		try {
+			hibernateTemplate.flush();
+		}
+		finally {
+			// This should happen when the transaction commits anyway, but to be
+			// sure...
+			hibernateTemplate.clear();
+		}
 	}
 
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaAwareItemWriter.java

@@ -5,25 +5,25 @@ import java.util.List;
 import javax.persistence.EntityManager;
 import javax.persistence.EntityManagerFactory;
 
-import org.springframework.batch.item.ClearFailedException;
 import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.support.AbstractItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.orm.jpa.EntityManagerFactoryUtils;
 import org.springframework.util.Assert;
 
 /**
- * {@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.<br/>
- *
- * It is required that {@link #write(List)} is called inside a transaction,
- * and that {@link #flush()} is then subsequently called before the transaction
- * commits, or {@link #clear()} before it rolls back.<br/>
- *
- * The reader must be configured with an {@link javax.persistence.EntityManagerFactory} that is capable
- * of participating in Spring managed transactions.
+ * {@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.<br/>
+ * 
+ * It is required that {@link #write(List)} is called inside a transaction.<br/>
+ * 
+ * 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.
@@ -31,25 +31,21 @@ import org.springframework.util.Assert;
  * 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<T> extends AbstractTransactionalResourceItemWriter<T> 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";
+public class JpaAwareItemWriter<T> extends AbstractItemWriter<T> implements InitializingBean {
 
 	private ItemWriter<? super T> delegate;
 
 	private EntityManagerFactory entityManagerFactory;
 
 	/**
-	 * Public setter for the {@link org.springframework.batch.item.ItemWriter} property.
-	 *
+	 * Public setter for the {@link org.springframework.batch.item.ItemWriter}
+	 * property.
+	 * 
 	 * @param delegate the delegate to set
 	 */
 	public void setDelegate(ItemWriter<? super T> delegate) {
@@ -58,7 +54,7 @@ public class JpaAwareItemWriter<T> extends AbstractTransactionalResourceItemWrit
 
 	/**
 	 * Set the EntityManager to be used internally.
-	 *
+	 * 
 	 * @param entityManagerFactory the entityManagerFactory to set
 	 */
 	public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
@@ -66,7 +62,8 @@ public class JpaAwareItemWriter<T> extends AbstractTransactionalResourceItemWrit
 	}
 
 	/**
-	 * Check mandatory properties - there must be a delegate and 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.");
@@ -74,38 +71,23 @@ public class JpaAwareItemWriter<T> extends AbstractTransactionalResourceItemWrit
 	}
 
 	/**
-	 * Delegate to subclass and flush the EntityManager.
+	 * Delegate the writing to the delegate writer and then flush and clear the
+	 * entity manager.
+	 * 
+	 * @see org.springframework.batch.item.ItemWriter#write(java.util.List)
 	 */
-	protected void doFlush() {
-		delegate.flush();
-		EntityManager entityManager =
-				EntityManagerFactoryUtils.getTransactionalEntityManager(entityManagerFactory);
+	public void write(List<? extends T> items) throws Exception {
+		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");
+		delegate.write(items);
+		try {
+			entityManager.flush();
+		}
+		finally {
+			entityManager.clear();
 		}
-		entityManager.clear();
-	}
-
-	protected String getResourceKey() {
-		return ITEMS_PROCESSED;
-	}
-
-	protected void doWrite(List<? extends T> item) throws Exception {
-		delegate.write(item);
 	}
 
 }