Tweak concurrency support in transaction aware collections

spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/repository/ConcurrentMapExecutionContextDaoTests.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright 2006-2009 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.core.test.repository;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Executors;
+
+import org.junit.Test;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.repository.dao.MapExecutionContextDao;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class ConcurrentMapExecutionContextDaoTests {
+
+	private MapExecutionContextDao dao = new MapExecutionContextDao();
+
+	private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
+
+	@Test
+	public void testSaveUpdate() throws Exception {
+		StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
+		stepExecution.setId(123L);
+		stepExecution.getExecutionContext().put("foo", "bar");
+		dao.saveExecutionContext(stepExecution);
+		ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
+		assertEquals("bar", executionContext.get("foo"));
+	}
+
+	@Test
+	public void testTransactionalSaveUpdate() throws Exception {
+		final StepExecution stepExecution = new StepExecution("step", new JobExecution(11L));
+		stepExecution.setId(123L);
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				stepExecution.getExecutionContext().put("foo", "bar");
+				dao.saveExecutionContext(stepExecution);
+				return null;
+			}
+		});
+		ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
+		assertEquals("bar", executionContext.get("foo"));
+
+	}
+
+	@Test
+	public void testConcurrentTransactionalSaveUpdate() throws Exception {
+
+		CompletionService<StepExecution> completionService = new ExecutorCompletionService<StepExecution>(Executors
+				.newFixedThreadPool(3));
+
+		final int outerMax = 10;
+		final int innerMax = 100;
+
+		for (int i = 0; i < outerMax; i++) {
+
+			final StepExecution stepExecution1 = new StepExecution("step", new JobExecution(11L));
+			stepExecution1.setId(123L + i);
+			final StepExecution stepExecution2 = new StepExecution("step", new JobExecution(11L));
+			stepExecution2.setId(1234L + i);
+
+			completionService.submit(new Callable<StepExecution>() {
+				public StepExecution call() throws Exception {
+					for (int i = 0; i < innerMax; i++) {
+						String value = "bar" + i;
+						saveAndAssert(stepExecution1, value);
+					}
+					return stepExecution1;
+				}
+			});
+
+			completionService.submit(new Callable<StepExecution>() {
+				public StepExecution call() throws Exception {
+					for (int i = 0; i < innerMax; i++) {
+						String value = "spam" + i;
+						saveAndAssert(stepExecution2, value);
+					}
+					return stepExecution2;
+				}
+			});
+
+			completionService.take().get();
+			completionService.take().get();
+
+		}
+
+	}
+
+	private void saveAndAssert(final StepExecution stepExecution, final String value) {
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				stepExecution.getExecutionContext().put("foo", value);
+				dao.saveExecutionContext(stepExecution);
+				return null;
+			}
+		});
+
+		ExecutionContext executionContext = dao.getExecutionContext(stepExecution);
+		Assert.state(executionContext != null, "Lost insert: null executionContext at value=" + value);
+		String foo = executionContext.getString("foo");
+		Assert.state(value.equals(foo), "Lost update: wrong value=" + value + " (found " + foo + ") for id="
+				+ stepExecution.getId());
+
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java

@@ -25,8 +25,7 @@ import org.springframework.batch.support.SerializationUtils;
 import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
 
 /**
- * In-memory implementation of {@link ExecutionContextDao} backed by static
- * maps.
+ * In-memory implementation of {@link ExecutionContextDao} backed by maps.
  * 
  * @author Robert Kasanicky
  * @author Dave Syer
@@ -34,10 +33,10 @@ import org.springframework.batch.support.transaction.TransactionAwareProxyFactor
 public class MapExecutionContextDao implements ExecutionContextDao {
 
 	private Map<Long, ExecutionContext> contextsByStepExecutionId = TransactionAwareProxyFactory
-			.createTransactionalMap();
+			.createAppendOnlyTransactionalMap();
 
 	private Map<Long, ExecutionContext> contextsByJobExecutionId = TransactionAwareProxyFactory
-			.createTransactionalMap();
+			.createAppendOnlyTransactionalMap();
 
 	public void clear() {
 		contextsByJobExecutionId.clear();

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java

@@ -24,6 +24,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
@@ -33,28 +35,46 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 /**
+ * <p>
  * Factory for transaction aware objects (like lists, sets, maps). If a
  * transaction is active when a method is called on an instance created by the
  * factory, it makes a copy of the target object and carries out all operations
  * on the copy. Only when the transaction commits is the target re-initialised
- * with the copy.<br/>
+ * with the copy.
+ * </p>
  * 
+ * <p>
  * Works well with collections and maps for testing transactional behaviour
  * without needing a database. The base implementation handles lists, sets and
  * maps. Subclasses can implement {@link #begin(Object)} and
- * {@link #commit(Object, Object)} to provide support for other resources.<br/>
+ * {@link #commit(Object, Object)} to provide support for other resources.
+ * </p>
  * 
- * Not intended for multi-threaded use.
+ * <p>
+ * Generally not intended for multi-threaded use, but the
+ * {@link #createAppendOnlyTransactionalMap() append only version} of
+ * collections gives isolation between threads operating on different keys in a
+ * map, provided they only append to the map. (Threads are limited to removing
+ * entries that were created in the same transaction.)
+ * </p>
  * 
  * @author Dave Syer
  * 
  */
 public class TransactionAwareProxyFactory<T> {
 
-	private T target;
+	private final T target;
+
+	private final boolean appendOnly;
 
 	private TransactionAwareProxyFactory(T target) {
+		this(target, false);
+
+	}
+
+	private TransactionAwareProxyFactory(T target, boolean appendOnly) {
 		super();
+		this.appendOnly = appendOnly;
 		this.target = begin(target);
 	}
 
@@ -68,14 +88,25 @@ public class TransactionAwareProxyFactory<T> {
 	 */
 	@SuppressWarnings("unchecked")
 	protected synchronized final T begin(T target) {
+		// Unfortunately in Java 5 this method has to synchronized
+		// (works OK without in Java 6).
 		if (target instanceof List) {
+			if (appendOnly) {
+				return (T) new ArrayList();
+			}
 			return (T) new ArrayList((List) target);
 		}
 		else if (target instanceof Set) {
+			if (appendOnly) {
+				return (T) new HashSet();
+			}
 			return (T) new HashSet((Set) target);
 		}
 		else if (target instanceof Map) {
-			return (T) new ConcurrentHashMap((Map) target);
+			if (appendOnly) {
+				return (T) new HashMap();
+			}
+			return (T) new HashMap((Map) target);
 		}
 		else {
 			throw new UnsupportedOperationException("Cannot copy target for this type: " + target.getClass());
@@ -92,20 +123,27 @@ public class TransactionAwareProxyFactory<T> {
 	 */
 	@SuppressWarnings("unchecked")
 	protected synchronized void commit(T copy, T target) {
+		// Unfortunately in Java 5 this method has to synchronized
+		// (works OK without in Java 6).
 		if (target instanceof Collection) {
-			((Collection) target).clear();
+			if (!appendOnly) {
+				((Collection) target).clear();
+			}
 			((Collection) target).addAll((Collection) copy);
 		}
 		else {
-			((Map) target).clear();
+			if (!appendOnly) {
+				((Map) target).clear();
+			}
 			((Map) target).putAll((Map) copy);
 		}
 	}
 
 	private T createInstance() {
+
 		ProxyFactory factory = new ProxyFactory(target);
 		factory.addAdvice(new MethodInterceptor() {
-			
+
 			public Object invoke(MethodInvocation invocation) throws Throwable {
 
 				if (!TransactionSynchronizationManager.isActualTransactionActive()) {
@@ -125,7 +163,15 @@ public class TransactionAwareProxyFactory<T> {
 					cache = retrievedCache;
 				}
 
-				return invocation.getMethod().invoke(cache, invocation.getArguments());
+				Object result = invocation.getMethod().invoke(cache, invocation.getArguments());
+				
+				String methodName = invocation.getMethod().getName();
+				if (appendOnly && result==null && (methodName.equals("get") || methodName.equals("contains"))) {
+					// In appendOnly mode the result of a get might not be in the cache...
+					return invocation.proceed();
+				}
+
+				return result;
 
 			}
 		});
@@ -133,35 +179,50 @@ public class TransactionAwareProxyFactory<T> {
 		T instance = (T) factory.getProxy();
 		return instance;
 	}
-	
+
 	@SuppressWarnings("unchecked")
-	public static <K,V> Map<K,V> createTransactionalMap() {
-		return (Map<K,V>) new TransactionAwareProxyFactory(new HashMap<K,V>()).createInstance();
+	public static <K, V> Map<K, V> createTransactionalMap() {
+		return (Map<K, V>) new TransactionAwareProxyFactory(new ConcurrentHashMap<K, V>()).createInstance();
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <K,V> Map<K,V> createTransactionalMap(Map<K,V> map) {
-		return (Map<K,V>) new TransactionAwareProxyFactory(map).createInstance();
+	public static <K, V> Map<K, V> createTransactionalMap(Map<K, V> map) {
+		return (Map<K, V>) new TransactionAwareProxyFactory(new ConcurrentHashMap<K, V>(map)).createInstance();
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <K, V> Map<K, V> createAppendOnlyTransactionalMap() {
+		return (Map<K, V>) new TransactionAwareProxyFactory(new ConcurrentHashMap<K, V>(), true).createInstance();
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <T> Set<T> createAppendOnlyTransactionalSet() {
+		return (Set<T>) new TransactionAwareProxyFactory(new CopyOnWriteArraySet<T>(), true).createInstance();
 	}
 
 	@SuppressWarnings("unchecked")
 	public static <T> Set<T> createTransactionalSet() {
-		return (Set<T>) new TransactionAwareProxyFactory(new HashSet<T>()).createInstance();
+		return (Set<T>) new TransactionAwareProxyFactory(new CopyOnWriteArraySet<T>()).createInstance();
 	}
 
 	@SuppressWarnings("unchecked")
 	public static <T> Set<T> createTransactionalSet(Set<T> set) {
-		return (Set<T>) new TransactionAwareProxyFactory(set).createInstance();
+		return (Set<T>) new TransactionAwareProxyFactory(new CopyOnWriteArraySet<T>(set)).createInstance();
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <T> List<T> createAppendOnlyTransactionalList() {
+		return (List<T>) new TransactionAwareProxyFactory(new CopyOnWriteArrayList<T>(), true).createInstance();
 	}
 
 	@SuppressWarnings("unchecked")
 	public static <T> List<T> createTransactionalList() {
-		return (List<T>) new TransactionAwareProxyFactory(new ArrayList<T>()).createInstance();
+		return (List<T>) new TransactionAwareProxyFactory(new CopyOnWriteArrayList<T>()).createInstance();
 	}
 
 	@SuppressWarnings("unchecked")
 	public static <T> List<T> createTransactionalList(List<T> list) {
-		return (List<T>) new TransactionAwareProxyFactory(list).createInstance();
+		return (List<T>) new TransactionAwareProxyFactory(new CopyOnWriteArrayList<T>(list)).createInstance();
 	}
 
 	private class TargetSynchronization extends TransactionSynchronizationAdapter {

spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/ConcurrentTransactionAwareProxyTests.java

@@ -0,0 +1,232 @@
+/*
+ * Copyright 2006-2009 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.support.transaction;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Executors;
+
+import org.junit.Test;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class ConcurrentTransactionAwareProxyTests {
+
+	private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
+
+	int outerMax = 10;
+
+	int innerMax = 10;
+
+	@Test(expected=Throwable.class)
+	public void testConcurrentTransactionalSet() throws Exception {
+		 Set<String> set = TransactionAwareProxyFactory.createTransactionalSet();
+		 testSet(set);
+	}
+
+	@Test
+	public void testConcurrentTransactionalAppendOnlySet() throws Exception {
+		 Set<String> set = TransactionAwareProxyFactory.createAppendOnlyTransactionalSet();
+		 testSet(set);
+	}
+
+	@Test
+	public void testConcurrentTransactionalAppendOnlyList() throws Exception {
+		List<String> list = TransactionAwareProxyFactory.createAppendOnlyTransactionalList();
+		testList(list);
+	}
+	
+	@Test(expected=Throwable.class)
+	public void testConcurrentTransactionalList() throws Exception {
+		List<String> list = TransactionAwareProxyFactory.createTransactionalList();
+		testList(list);
+	}
+
+	@Test
+	public void testConcurrentTransactionalAppendOnlyMap() throws Exception {
+		Map<Long, Map<String, String>> map = TransactionAwareProxyFactory.createAppendOnlyTransactionalMap();
+		testMap(map);
+	}
+
+	@Test(expected = ExecutionException.class)
+	public void testConcurrentTransactionalMap() throws Exception {
+		Map<Long, Map<String, String>> map = TransactionAwareProxyFactory.createTransactionalMap();
+		testMap(map);
+	}
+	
+	private void testSet(final Set<String> set) throws Exception {
+
+		CompletionService<List<String>> completionService = new ExecutorCompletionService<List<String>>(Executors
+				.newFixedThreadPool(outerMax));
+
+		for (int i = 0; i < outerMax; i++) {
+
+			final int count = i;
+			completionService.submit(new Callable<List<String>>() {
+				public List<String> call() throws Exception {
+					List<String> list = new ArrayList<String>();
+					for (int i = 0; i < innerMax; i++) {
+						String value = count + "bar" + i;
+						saveInSetAndAssert(set, value);
+						list.add(value);
+					}
+					return list;
+				}
+			});
+
+		}
+
+		for (int i = 0; i < outerMax; i++) {
+			List<String> result = completionService.take().get();
+			assertEquals(innerMax, result.size());
+		}
+
+		assertEquals(innerMax * outerMax, set.size());
+
+	}
+
+	private void testList(final List<String> list) throws Exception {
+
+		CompletionService<List<String>> completionService = new ExecutorCompletionService<List<String>>(Executors
+				.newFixedThreadPool(outerMax));
+
+		for (int i = 0; i < outerMax; i++) {
+
+			completionService.submit(new Callable<List<String>>() {
+				public List<String> call() throws Exception {
+					List<String> result = new ArrayList<String>();
+					for (int i = 0; i < innerMax; i++) {
+						String value = "bar" + i;
+						saveInListAndAssert(list, value);
+						result.add(value);
+					}
+					return result;
+				}
+			});
+
+		}
+
+		for (int i = 0; i < outerMax; i++) {
+			List<String> result = completionService.take().get();
+			assertEquals(innerMax, result.size());
+		}
+
+		assertEquals(innerMax * outerMax, list.size());
+
+	}
+
+	private void testMap(final Map<Long, Map<String, String>> map) throws Exception {
+
+		CompletionService<List<String>> completionService = new ExecutorCompletionService<List<String>>(Executors
+				.newFixedThreadPool(outerMax));
+		
+		int numberOfKeys = outerMax;
+
+		for (int i = 0; i < outerMax; i++) {
+
+			for (int j=0; j<numberOfKeys; j++) { 
+			final long id = j*1000 + 123L + i;
+
+			completionService.submit(new Callable<List<String>>() {
+				public List<String> call() throws Exception {
+					List<String> list = new ArrayList<String>();
+					for (int i = 0; i < innerMax; i++) {
+						String value = "bar" + i;
+						list.add(saveInMapAndAssert(map, id, value).get("foo"));
+					}
+					return list;
+				}
+			});
+			}
+
+			for (int j=0; j<numberOfKeys; j++) { 
+				completionService.take().get();
+			}
+
+		}
+
+	}
+
+	private String saveInSetAndAssert(final Set<String> set, final String value) {
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				set.add(value);
+				return null;
+			}
+		});
+
+		Assert.state(set.contains(value), "Lost update: value=" + value);
+
+		return value;
+
+	}
+
+	private String saveInListAndAssert(final List<String> list, final String value) {
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				list.add(value);
+				return null;
+			}
+		});
+
+		Assert.state(list.contains(value), "Lost update: value=" + value);
+
+		return value;
+
+	}
+
+	private Map<String, String> saveInMapAndAssert(final Map<Long, Map<String, String>> map, final Long id,
+			final String value) {
+
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				if (!map.containsKey(id)) {
+					map.put(id, new HashMap<String, String>());
+				}
+				map.get(id).put("foo", value);
+				return null;
+			}
+		});
+
+		Map<String, String> result = map.get(id);
+		Assert.state(result != null, "Lost insert: null String at value=" + value);
+		String foo = result.get("foo");
+		Assert.state(value.equals(foo), "Lost update: wrong value=" + value + " (found " + foo + ") for id=" + id);
+
+		return result;
+
+	}
+
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareListFactoryTests.java

@@ -16,31 +16,39 @@
 
 package org.springframework.batch.support.transaction;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
 import java.util.Arrays;
 import java.util.List;
 
-import junit.framework.TestCase;
-
+import org.junit.Before;
+import org.junit.Test;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 
-public class TransactionAwareListFactoryTests extends TestCase {
+public class TransactionAwareListFactoryTests {
 
-	TransactionTemplate transactionTemplate = new TransactionTemplate(new ResourcelessTransactionManager());
+	private TransactionTemplate transactionTemplate = new TransactionTemplate(new ResourcelessTransactionManager());
 
-	List<String> list;
+	private List<String> list;
 
-	protected void setUp() throws Exception {
+	@Before
+	public void setUp() throws Exception {
 		list = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList("foo", "bar", "spam"));
 	}
 
+	@Test
 	public void testAdd() {
 		assertEquals(3, list.size());
 		list.add("bucket");
 		assertTrue(list.contains("bucket"));
 	}
 
+	@Test
 	public void testRemove() {
 		assertEquals(3, list.size());
 		assertTrue(list.contains("spam"));
@@ -48,12 +56,14 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertFalse(list.contains("spam"));
 	}
 
+	@Test
 	public void testClear() {
 		assertEquals(3, list.size());
 		list.clear();
 		assertEquals(0, list.size());
 	}
 
+	@Test
 	public void testTransactionalAdd() throws Exception {
 		transactionTemplate.execute(new TransactionCallback() {
 			public Object doInTransaction(TransactionStatus status) {
@@ -64,6 +74,7 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertEquals(4, list.size());
 	}
 
+	@Test
 	public void testTransactionalRemove() throws Exception {
 		transactionTemplate.execute(new TransactionCallback() {
 			public Object doInTransaction(TransactionStatus status) {
@@ -74,6 +85,7 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertEquals(2, list.size());
 	}
 
+	@Test
 	public void testTransactionalClear() throws Exception {
 		transactionTemplate.execute(new TransactionCallback() {
 			public Object doInTransaction(TransactionStatus status) {
@@ -84,6 +96,7 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertEquals(0, list.size());
 	}
 
+	@Test
 	public void testTransactionalAddWithRollback() throws Exception {
 		try {
 			transactionTemplate.execute(new TransactionCallback() {
@@ -100,6 +113,7 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertEquals(3, list.size());
 	}
 
+	@Test
 	public void testTransactionalRemoveWithRollback() throws Exception {
 		try {
 			transactionTemplate.execute(new TransactionCallback() {
@@ -116,6 +130,7 @@ public class TransactionAwareListFactoryTests extends TestCase {
 		assertEquals(3, list.size());
 	}
 
+	@Test
 	public void testTransactionalClearWithRollback() throws Exception {
 		try {
 			transactionTemplate.execute(new TransactionCallback() {

spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareSetFactoryTests.java

@@ -0,0 +1,150 @@
+/*
+ * 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.support.transaction;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+public class TransactionAwareSetFactoryTests {
+
+	private TransactionTemplate transactionTemplate = new TransactionTemplate(new ResourcelessTransactionManager());
+
+	private Set<String> set;
+
+	@Before
+	public void setUp() throws Exception {
+		set = TransactionAwareProxyFactory.createTransactionalSet(new HashSet<String>(Arrays.asList("foo", "bar", "spam")));
+	}
+
+	@Test
+	public void testAdd() {
+		assertEquals(3, set.size());
+		set.add("bucket");
+		assertTrue(set.contains("bucket"));
+	}
+
+	@Test
+	public void testRemove() {
+		assertEquals(3, set.size());
+		assertTrue(set.contains("spam"));
+		set.remove("spam");
+		assertFalse(set.contains("spam"));
+	}
+
+	@Test
+	public void testClear() {
+		assertEquals(3, set.size());
+		set.clear();
+		assertEquals(0, set.size());
+	}
+
+	@Test
+	public void testTransactionalAdd() throws Exception {
+		transactionTemplate.execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				testAdd();
+				return null;
+			}
+		});
+		assertEquals(4, set.size());
+	}
+
+	@Test
+	public void testTransactionalRemove() throws Exception {
+		transactionTemplate.execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				testRemove();
+				return null;
+			}
+		});
+		assertEquals(2, set.size());
+	}
+
+	@Test
+	public void testTransactionalClear() throws Exception {
+		transactionTemplate.execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				testClear();
+				return null;
+			}
+		});
+		assertEquals(0, set.size());
+	}
+
+	@Test
+	public void testTransactionalAddWithRollback() throws Exception {
+		try {
+			transactionTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					testAdd();
+					throw new RuntimeException("Rollback!");
+				}
+			});
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Rollback!", e.getMessage());
+		}
+		assertEquals(3, set.size());
+	}
+
+	@Test
+	public void testTransactionalRemoveWithRollback() throws Exception {
+		try {
+			transactionTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					testRemove();
+					throw new RuntimeException("Rollback!");
+				}
+			});
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Rollback!", e.getMessage());
+		}
+		assertEquals(3, set.size());
+	}
+
+	@Test
+	public void testTransactionalClearWithRollback() throws Exception {
+		try {
+			transactionTemplate.execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					testClear();
+					throw new RuntimeException("Rollback!");
+				}
+			});
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Rollback!", e.getMessage());
+		}
+		assertEquals(3, set.size());
+	}
+}