BATCH-1861: Concurrency support for the in-memory jobs, job repositories, etc.

spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java

@@ -26,6 +26,7 @@ import static org.junit.Assert.fail;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -100,8 +101,8 @@ public class SimpleJobOperatorTests {
 			}
 
 			@Override
-			public Collection<String> getJobNames() {
-				return Arrays.asList(new String[] { "foo", "bar" });
+			public Set<String> getJobNames() {
+				return new HashSet(Arrays.asList(new String[] { "foo", "bar" }));
 			}
 		});
 

spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java

@@ -3,6 +3,7 @@ package org.springframework.batch.core.repository.dao;
 import static org.junit.Assert.*;
 
 import org.junit.Test;
+import org.junit.Ignore;
 import org.junit.internal.runners.JUnit4ClassRunner;
 import org.junit.runner.RunWith;
 import org.springframework.batch.core.JobExecution;
@@ -34,7 +35,31 @@ public class MapExecutionContextDaoTests extends AbstractExecutionContextDaoTest
 	protected ExecutionContextDao getExecutionContextDao() {
 		return new MapExecutionContextDao();
 	}
+
+	@Test
+	public void testSaveBothJobAndStepContextWithSameId() throws Exception {
+		MapExecutionContextDao tested = new MapExecutionContextDao();
+		JobExecution jobExecution = new JobExecution(1L);
+		StepExecution stepExecution = new StepExecution("stepName", jobExecution, 1L);
+		
+		assertTrue(stepExecution.getId() == jobExecution.getId());
+		
+		jobExecution.getExecutionContext().put("type", "job");
+		stepExecution.getExecutionContext().put("type", "step");
+		assertTrue(!jobExecution.getExecutionContext().get("type").equals(stepExecution.getExecutionContext().get("type")));
+		assertEquals("job", jobExecution.getExecutionContext().get("type"));
+		assertEquals("step", stepExecution.getExecutionContext().get("type"));
+
+		tested.saveExecutionContext(jobExecution);
+		tested.saveExecutionContext(stepExecution);
 	
+		ExecutionContext jobCtx = tested.getExecutionContext(jobExecution);
+		ExecutionContext stepCtx = tested.getExecutionContext(stepExecution);
+
+		assertEquals("job", jobCtx.get("type"));
+		assertEquals("step", stepCtx.get("type"));
+	}
+
 	@Test
 	public void testPersistentCopy() throws Exception {
 		MapExecutionContextDao tested = new MapExecutionContextDao();

spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapJobExecutionDaoTests.java

@@ -1,6 +1,18 @@
 package org.springframework.batch.core.repository.dao;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
+
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -45,4 +57,63 @@ public class MapJobExecutionDaoTests extends AbstractJobExecutionDaoTests {
 		
 	}
 
+	/**
+	* Verify that the ids are properly generated even under heavy concurrent load
+	*/
+	@Test
+	public void testConcurrentSaveJobExecution() throws Exception {
+		final int iterations = 100;
+
+		// Object under test
+		final JobExecutionDao tested = new MapJobExecutionDao();
+
+		// Support objects for this testing
+		final CountDownLatch latch = new CountDownLatch(1);
+		final SortedSet<Long> ids = Collections.synchronizedSortedSet(new TreeSet<Long>()); // TODO Change to SkipList w/JDK6
+		final AtomicReference<Exception> exception = new AtomicReference<Exception>(null);
+
+		// Implementation of the high-concurrency code
+		final Runnable codeUnderTest = new Runnable() {
+			public void run() {
+				try {
+					JobExecution jobExecution = new JobExecution(new JobInstance((long) -1, new JobParameters(), "mapJob"));
+					latch.await();
+					tested.saveJobExecution(jobExecution);
+					ids.add(jobExecution.getId());
+				} catch(Exception e) {
+					exception.set(e);
+				}
+			}
+		};
+
+		// Create the threads
+		final Thread[] threads = new Thread[iterations];
+		for(int i = 0; i < iterations; i++) {
+			Thread t = new Thread(codeUnderTest, "Map Job Thread #" + (i+1));
+			t.setPriority(Thread.MAX_PRIORITY);
+			t.setDaemon(true);
+			t.start();
+			Thread.yield();
+			threads[i] = t;
+		}
+
+		// Let the high concurrency abuse begin!
+		do { latch.countDown(); } while(latch.getCount() > 0);
+		for(Thread t : threads) { t.join(); }
+
+		// Ensure no general exceptions arose
+		if(exception.get() != null) throw new RuntimeException("Excepion occurred under high concurrency usage", exception.get());
+
+		// Validate the ids: we'd expect one of these three things to fail
+		if(ids.size() < iterations) {
+			fail("Duplicate id generated during high concurrency usage");	
+		}
+		if(ids.first() < 0) {
+			fail("Generated an id less than zero during high concurrency usage: " + ids.first());
+		}
+		if(ids.last() > iterations) {
+			fail("Generated an id larger than expected during high concurrency usage: " + ids.last());
+		}
+	}
+
 }

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java

@@ -17,6 +17,7 @@ import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.batch.core.BatchStatus;
@@ -58,15 +59,13 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 
 	private JobRepository repository;
 
-	public FaultTolerantStepFactoryBeanRollbackTests() throws Exception {
-		reader = new SkipReaderStub<String>();
-		processor = new SkipProcessorStub<String>();
-		writer = new SkipWriterStub<String>();
-	}
-
 	@SuppressWarnings("unchecked")
 	@Before
 	public void setUp() throws Exception {
+		reader = new SkipReaderStub<String>();
+		processor = new SkipProcessorStub<String>();
+		writer = new SkipWriterStub<String>();
+
 		factory = new FaultTolerantStepFactoryBean<String, String>();
 
 		factory.setBeanName("stepName");
@@ -96,6 +95,14 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		stepExecution = jobExecution.createStepExecution(factory.getName());
 		repository.add(stepExecution);
 	}
+
+	@After
+	public void tearDown() throws Exception {
+		reader = null;
+		processor = null;
+		writer = null;
+		factory = null;
+	}
 	
 	@Test
 	public void testBeforeChunkListenerException() throws Exception{

spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java

@@ -222,6 +222,8 @@ public class StepExecutorInterruptionTests {
 				}
 			}
 		};
+		processingThread.setDaemon(true);
+		processingThread.setPriority(Thread.MIN_PRIORITY);
 		return processingThread;
 	}