diff --git a/spring-batch-core-tests/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-core-tests/.settings/com.springsource.sts.config.flow.prefs new file mode 100644 index 000000000..bace70768 --- /dev/null +++ b/spring-batch-core-tests/.settings/com.springsource.sts.config.flow.prefs @@ -0,0 +1,3 @@ +#Sat Mar 13 13:37:06 GMT 2010 +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core-tests/src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n +eclipse.preferences.version=1 diff --git a/spring-batch-core-tests/.springBeans b/spring-batch-core-tests/.springBeans index 54dd032b6..5f141c431 100644 --- a/spring-batch-core-tests/.springBeans +++ b/spring-batch-core-tests/.springBeans @@ -1,239 +1,240 @@ - - - 1 - - - - - - - src/test/resources/data-source-context.xml - src/test/resources/simple-job-launcher-context.xml - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - - true - false - - - - - + + + 1 + + + + + + + src/test/resources/data-source-context.xml + src/test/resources/simple-job-launcher-context.xml + src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + + true + false + + + + + diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java index c886c279c..5af26034e 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java @@ -11,6 +11,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; import javax.sql.DataSource; @@ -121,6 +122,10 @@ public class FaultTolerantStepFactoryBeanRollbackTests { for (int i = 0; i < MAX_COUNT; i++) { + if (i%100==0) { + logger.info("Starting step: "+i); + } + SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource); assertEquals(0, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "ERROR_LOG")); @@ -145,6 +150,9 @@ public class FaultTolerantStepFactoryBeanRollbackTests { assertEquals("[]", writer.getCommitted().toString()); assertEquals("[]", processor.getCommitted().toString()); + List processed = new ArrayList(processor.getProcessed()); + Collections.sort(processed); + assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString()); assertEquals(5, stepExecution.getSkipCount()); } @@ -196,7 +204,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests { private static class SkipWriterStub implements ItemWriter { - private List written = new ArrayList(); + private List written = new CopyOnWriteArrayList(); private Collection failures = Collections.emptySet(); @@ -243,7 +251,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests { private final Log logger = LogFactory.getLog(getClass()); - private List processed = new ArrayList(); + private List processed = new CopyOnWriteArrayList(); private SimpleJdbcTemplate jdbcTemplate; @@ -253,6 +261,13 @@ public class FaultTolerantStepFactoryBeanRollbackTests { public SkipProcessorStub(DataSource dataSource) { jdbcTemplate = new SimpleJdbcTemplate(dataSource); } + + /** + * @return the processed + */ + public List getProcessed() { + return processed; + } public List getCommitted() { return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'", diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java new file mode 100644 index 000000000..0ec70a40d --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanRollbackTests.java @@ -0,0 +1,248 @@ +package org.springframework.batch.core.test.step; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * Tests for {@link FaultTolerantStepFactoryBean}. + */ +@ContextConfiguration(locations = "/simple-job-launcher-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class MapRepositoryFaultTolerantStepFactoryBeanRollbackTests { + + private static final int MAX_COUNT = 1000; + + private final Log logger = LogFactory.getLog(getClass()); + + private FaultTolerantStepFactoryBean factory; + + private SkipReaderStub reader; + + private SkipProcessorStub processor; + + private SkipWriterStub writer; + + private JobExecution jobExecution; + + private StepExecution stepExecution; + + private JobRepository repository; + + @Autowired + private PlatformTransactionManager transactionManager; + + @SuppressWarnings("unchecked") + @Before + public void setUp() throws Exception { + + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + + reader = new SkipReaderStub(); + writer = new SkipWriterStub(); + processor = new SkipProcessorStub(); + + factory = new FaultTolerantStepFactoryBean(); + + factory.setBeanName("stepName"); + factory.setTransactionManager(transactionManager); + factory.setJobRepository(repository); + factory.setCommitInterval(3); + ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + taskExecutor.setCorePoolSize(3); + taskExecutor.setMaxPoolSize(6); + taskExecutor.setQueueCapacity(0); + taskExecutor.afterPropertiesSet(); + factory.setTaskExecutor(taskExecutor); + + factory.setSkipLimit(10); + factory.setSkippableExceptionClasses(getExceptionMap(Exception.class)); + + } + + @Test + public void testUpdatesNoRollback() throws Exception { + + writer.write(Arrays.asList("foo", "bar")); + processor.process("spam"); + assertEquals(2, writer.getWritten().size()); + assertEquals(1, processor.getProcessed().size()); + + writer.clear(); + processor.clear(); + assertEquals(0, processor.getProcessed().size()); + + } + + @Test + public void testMultithreadedSkipInWrite() throws Throwable { + + for (int i = 0; i < MAX_COUNT; i++) { + + if (i%100==0) { + logger.info("Starting step: "+i); + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + factory.setJobRepository(repository); + jobExecution = repository.createJobExecution("vanillaJob", new JobParameters()); + } + + reader.clear(); + reader.setItems("1", "2", "3", "4", "5"); + factory.setItemReader(reader); + writer.clear(); + factory.setItemWriter(writer); + processor.clear(); + factory.setItemProcessor(processor); + + writer.setFailures("1", "2", "3", "4", "5"); + + try { + + Step step = (Step) factory.getObject(); + + stepExecution = jobExecution.createStepExecution(factory.getName()); + repository.add(stepExecution); + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + + assertEquals(5, stepExecution.getSkipCount()); + List processed = new ArrayList(processor.getProcessed()); + Collections.sort(processed); + assertEquals("[1, 1, 2, 2, 3, 3, 4, 4, 5, 5]", processed.toString()); + + } + catch (Throwable e) { + logger.info("Failed on iteration " + i + " of " + MAX_COUNT); + throw e; + } + + } + + } + + private static class SkipReaderStub implements ItemReader { + + private String[] items; + + private int counter = -1; + + public SkipReaderStub() throws Exception { + super(); + } + + public void setItems(String... items) { + Assert.isTrue(counter < 0, "Items cannot be set once reading has started"); + this.items = items; + } + + public void clear() { + counter = -1; + } + + public synchronized String read() throws Exception, UnexpectedInputException, ParseException { + counter++; + if (counter >= items.length) { + return null; + } + String item = items[counter]; + return item; + } + } + + private static class SkipWriterStub implements ItemWriter { + + private final Log logger = LogFactory.getLog(getClass()); + + private List written = new CopyOnWriteArrayList(); + + private Collection failures = Collections.emptySet(); + + public void setFailures(String... failures) { + this.failures = Arrays.asList(failures); + } + + public List getWritten() { + return written; + } + + public void clear() { + written.clear(); + } + + public void write(List items) throws Exception { + for (String item : items) { + logger.trace("Writing: "+item); + written.add(item); + checkFailure(item); + } + } + + private void checkFailure(String item) { + if (failures.contains(item)) { + throw new RuntimeException("Planned failure"); + } + } + } + + private static class SkipProcessorStub implements ItemProcessor { + + private final Log logger = LogFactory.getLog(getClass()); + + private List processed = new CopyOnWriteArrayList(); + + public List getProcessed() { + return processed; + } + + public void clear() { + processed.clear(); + } + + public String process(String item) throws Exception { + processed.add(item); + logger.debug("Processed item: "+item); + return item; + } + } + + private Map, Boolean> getExceptionMap(Class... args) { + Map, Boolean> map = new HashMap, Boolean>(); + for (Class arg : args) { + map.put(arg, true); + } + return map; + } + +} diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java new file mode 100644 index 000000000..7b4d167c4 --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/MapRepositoryFaultTolerantStepFactoryBeanTests.java @@ -0,0 +1,228 @@ +package org.springframework.batch.core.test.step; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; + +/** + * Tests for {@link FaultTolerantStepFactoryBean}. + */ +@ContextConfiguration(locations = "/simple-job-launcher-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class MapRepositoryFaultTolerantStepFactoryBeanTests { + + private static final int MAX_COUNT = 1000; + + private final Log logger = LogFactory.getLog(getClass()); + + private FaultTolerantStepFactoryBean factory; + + private SkipReaderStub reader; + + private SkipProcessorStub processor; + + private SkipWriterStub writer; + + private JobExecution jobExecution; + + private StepExecution stepExecution; + + private JobRepository repository; + + @Autowired + private PlatformTransactionManager transactionManager; + + @Before + public void setUp() throws Exception { + + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + + reader = new SkipReaderStub(); + writer = new SkipWriterStub(); + processor = new SkipProcessorStub(); + + factory = new FaultTolerantStepFactoryBean(); + + factory.setBeanName("stepName"); + factory.setTransactionManager(transactionManager); + factory.setJobRepository(repository); + factory.setCommitInterval(3); + ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + taskExecutor.setCorePoolSize(3); + taskExecutor.setMaxPoolSize(6); + taskExecutor.setQueueCapacity(0); + taskExecutor.afterPropertiesSet(); + factory.setTaskExecutor(taskExecutor); + + } + + @Test + public void testUpdatesNoRollback() throws Exception { + + writer.write(Arrays.asList("foo", "bar")); + processor.process("spam"); + assertEquals(2, writer.getWritten().size()); + assertEquals(1, processor.getProcessed().size()); + + writer.clear(); + processor.clear(); + assertEquals(0, processor.getProcessed().size()); + + } + + @Test + public void testMultithreadedSunnyDay() throws Throwable { + + for (int i = 0; i < MAX_COUNT; i++) { + + if (i%100==0) { + logger.info("Starting step: "+i); + repository = new MapJobRepositoryFactoryBean().getJobRepository(); + factory.setJobRepository(repository); + jobExecution = repository.createJobExecution("vanillaJob", new JobParameters()); + } + + reader.clear(); + reader.setItems("1", "2", "3", "4", "5"); + factory.setItemReader(reader); + writer.clear(); + factory.setItemWriter(writer); + processor.clear(); + factory.setItemProcessor(processor); + + try { + + Step step = (Step) factory.getObject(); + + stepExecution = jobExecution.createStepExecution(factory.getName()); + repository.add(stepExecution); + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + + List committed = new ArrayList(writer.getWritten()); + Collections.sort(committed); + assertEquals("[1, 2, 3, 4, 5]", committed.toString()); + List processed = new ArrayList(processor.getProcessed()); + Collections.sort(processed); + assertEquals("[1, 2, 3, 4, 5]", processed.toString()); + assertEquals(0, stepExecution.getSkipCount()); + + } + catch (Throwable e) { + logger.info("Failed on iteration " + i + " of " + MAX_COUNT); + throw e; + } + + } + + } + + private static class SkipReaderStub implements ItemReader { + + private String[] items; + + private int counter = -1; + + public SkipReaderStub() throws Exception { + super(); + } + + public void setItems(String... items) { + Assert.isTrue(counter < 0, "Items cannot be set once reading has started"); + this.items = items; + } + + public void clear() { + counter = -1; + } + + public synchronized String read() throws Exception, UnexpectedInputException, ParseException { + counter++; + if (counter >= items.length) { + return null; + } + String item = items[counter]; + return item; + } + } + + private static class SkipWriterStub implements ItemWriter { + + private List written = new CopyOnWriteArrayList(); + + private Collection failures = Collections.emptySet(); + + public List getWritten() { + return written; + } + + public void clear() { + written.clear(); + } + + public void write(List items) throws Exception { + for (String item : items) { + written.add(item); + checkFailure(item); + } + } + + private void checkFailure(String item) { + if (failures.contains(item)) { + throw new RuntimeException("Planned failure"); + } + } + } + + private static class SkipProcessorStub implements ItemProcessor { + + private final Log logger = LogFactory.getLog(getClass()); + + private List processed = new CopyOnWriteArrayList(); + + public List getProcessed() { + return processed; + } + + public void clear() { + processed.clear(); + } + + public String process(String item) throws Exception { + processed.add(item); + logger.debug("Processed item: "+item); + return item; + } + } + +} diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests.java new file mode 100644 index 000000000..4006b947d --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests.java @@ -0,0 +1,92 @@ +/* + * 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.step; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParametersBuilder; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Dave Syer + * + */ +public class SplitJobMapRepositoryIntegrationTests { + + private static final int MAX_COUNT = 1000; + + /** Logger */ + private final Log logger = LogFactory.getLog(getClass()); + + @Test + public void testMultithreadedSplit() throws Throwable { + + JobLauncher jobLauncher = null; + Job job = null; + + for (int i = 0; i < MAX_COUNT; i++) { + + if (i % 100 == 0) { + logger.info("Starting job: " + i); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(getClass().getSimpleName() + + "-context.xml", getClass()); + jobLauncher = (JobLauncher) context.getBean("jobLauncher", JobLauncher.class); + job = (Job) context.getBean("job", Job.class); + } + + try { + JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().addLong("count", new Long(i)) + .toJobParameters()); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + } + catch (Throwable e) { + logger.info("Failed on iteration " + i + " of " + MAX_COUNT); + throw e; + } + + } + + } + + public static class CountingTasklet implements Tasklet { + + private int maxCount = 10; + + private AtomicInteger count = new AtomicInteger(0); + + public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { + contribution.incrementReadCount(); + contribution.incrementWriteCount(1); + return RepeatStatus.continueIf(count.incrementAndGet() < maxCount); + } + + } + +} diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/StepExecutionSerializationUtilsTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/StepExecutionSerializationUtilsTests.java new file mode 100644 index 000000000..2c9d7de69 --- /dev/null +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/StepExecutionSerializationUtilsTests.java @@ -0,0 +1,100 @@ +/* + * Copyright 2006-2010 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.step; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.support.SerializationUtils; + +/** + * @author Dave Syer + * + */ +public class StepExecutionSerializationUtilsTests { + + @Test + public void testCycle() throws Exception { + StepExecution stepExecution = new StepExecution("step", new JobExecution(new JobInstance(123L, + new JobParameters(), "job"), 321L), 11L); + stepExecution.getExecutionContext().put("foo.bar.spam", 123); + StepExecution result = getCopy(stepExecution); + assertEquals(stepExecution, result); + } + + @Test + public void testMultipleCycles() throws Throwable { + + int count = 0; + int repeats = 100; + int threads = 10; + + Executor executor = Executors.newFixedThreadPool(threads); + CompletionService completionService = new ExecutorCompletionService(executor); + + for (int i = 0; i < repeats; i++) { + final JobExecution jobExecution = new JobExecution(new JobInstance(123L, new JobParameters(), "job"), 321L); + for (int j = 0; j < threads; j++) { + completionService.submit(new Callable() { + public StepExecution call() throws Exception { + final StepExecution stepExecution = jobExecution.createStepExecution("step"); + stepExecution.getExecutionContext().put("foo.bar.spam", 123); + StepExecution result = getCopy(stepExecution); + assertEquals(stepExecution.getExecutionContext(), result.getExecutionContext()); + return result; + } + }); + } + for (int j = 0; j < threads; j++) { + Future future = completionService.poll(repeats, TimeUnit.MILLISECONDS); + if (future != null) { + count++; + try { + future.get(); + } catch (Throwable e) { + throw new IllegalStateException("Failed on count="+count, e); + } + } + } + } + while (count < threads*repeats) { + Future future = completionService.poll(); + count++; + try { + future.get(); + } catch (Throwable e) { + throw new IllegalStateException("Failed on count="+count, e); + } + } + } + + private StepExecution getCopy(StepExecution stepExecution) { + return (StepExecution) SerializationUtils.deserialize(SerializationUtils.serialize(stepExecution)); + } + +} diff --git a/spring-batch-core-tests/src/test/resources/log4j.properties b/spring-batch-core-tests/src/test/resources/log4j.properties index b1071e86e..4f10f29e3 100644 --- a/spring-batch-core-tests/src/test/resources/log4j.properties +++ b/spring-batch-core-tests/src/test/resources/log4j.properties @@ -1,4 +1,4 @@ -log4j.rootCategory=INFO, stdout +log4j.rootCategory=WARN, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout @@ -7,8 +7,9 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{1}:%L - %m log4j.category.org.apache.activemq=ERROR # log4j.category.org.springframework=DEBUG log4j.category.org.springframework.jdbc=INFO +log4j.category.org.springframework.context=INFO log4j.category.org.springframework.jms=INFO -log4j.category.org.springframework.batch=INFO -# log4j.category.org.springframework.batch.core.test=DEBUG +# log4j.category.org.springframework.batch=INFO +log4j.category.org.springframework.batch.core.test=INFO log4j.category.org.springframework.retry=INFO # log4j.category.org.springframework.beans.factory.config=TRACE diff --git a/spring-batch-core-tests/src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml b/spring-batch-core-tests/src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml new file mode 100644 index 000000000..4c5e5789d --- /dev/null +++ b/spring-batch-core-tests/src/test/resources/org/springframework/batch/core/test/step/SplitJobMapRepositoryIntegrationTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs new file mode 100644 index 000000000..eb16c0f70 --- /dev/null +++ b/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs @@ -0,0 +1,3 @@ +#Wed Mar 24 07:48:49 GMT 2010 +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/JobRepositoryDefaultParserTests-context.xml=\n +eclipse.preferences.version=1 diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java index 99ed94b50..6ff30fdac 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java @@ -20,11 +20,13 @@ import java.io.IOException; import java.io.ObjectInputStream; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Date; import java.util.HashSet; -import java.util.LinkedHashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CopyOnWriteArraySet; import org.springframework.batch.item.ExecutionContext; @@ -38,7 +40,7 @@ public class JobExecution extends Entity { private JobInstance jobInstance; - private volatile Collection stepExecutions = new LinkedHashSet(); + private volatile Collection stepExecutions = new CopyOnWriteArraySet(); private volatile BatchStatus status = BatchStatus.STARTING; @@ -54,7 +56,7 @@ public class JobExecution extends Entity { private volatile ExecutionContext executionContext = new ExecutionContext(); - private transient volatile List failureExceptions = new ArrayList(); + private transient volatile List failureExceptions = new CopyOnWriteArrayList(); /** * Because a JobExecution isn't valid unless the job is set, this @@ -164,7 +166,7 @@ public class JobExecution extends Entity { * @return the step executions that were registered */ public Collection getStepExecutions() { - return stepExecutions; + return Collections.unmodifiableList(new ArrayList(stepExecutions)); } /** @@ -278,7 +280,7 @@ public class JobExecution extends Entity { * * @param t */ - public void addFailureException(Throwable t) { + public synchronized void addFailureException(Throwable t) { this.failureExceptions.add(t); } @@ -289,7 +291,7 @@ public class JobExecution extends Entity { * @return List containing all exceptions causing failure for * this JobExecution. */ - public List getAllFailureExceptions() { + public synchronized List getAllFailureExceptions() { Set allExceptions = new HashSet(failureExceptions); for (StepExecution stepExecution : stepExecutions) { @@ -320,8 +322,8 @@ public class JobExecution extends Entity { } /** - * Setter for the step executions. For internal use only. - * @param stepExecutions + * Add some step executions. For internal use only. + * @param stepExecutions step executions to add to the current list */ public void addStepExecutions(List stepExecutions) { if (stepExecutions!=null) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java index 9e787121d..bc1324fab 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java @@ -21,6 +21,7 @@ import java.io.ObjectInputStream; import java.util.ArrayList; import java.util.Date; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import org.springframework.batch.item.ExecutionContext; import org.springframework.util.Assert; @@ -70,7 +71,7 @@ public class StepExecution extends Entity { private volatile int filterCount; - private transient volatile List failureExceptions = new ArrayList(); + private transient volatile List failureExceptions = new CopyOnWriteArrayList(); /** * Constructor with mandatory properties. diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java index fcbf67c7f..2b4effdb5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java @@ -23,11 +23,11 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.support.SerializationUtils; -import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.util.Assert; @@ -36,7 +36,7 @@ import org.springframework.util.Assert; */ public class MapJobExecutionDao implements JobExecutionDao { - private Map executionsById = TransactionAwareProxyFactory.createTransactionalMap(); + private Map executionsById = new ConcurrentHashMap(); private long currentId = 0; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java index eec8dfd4e..97afb64fc 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java @@ -21,11 +21,11 @@ import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.concurrent.CopyOnWriteArraySet; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; -import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.util.Assert; /** @@ -33,7 +33,7 @@ import org.springframework.util.Assert; */ public class MapJobInstanceDao implements JobInstanceDao { - private Collection jobInstances = TransactionAwareProxyFactory.createTransactionalSet(); + private Collection jobInstances = new CopyOnWriteArraySet(); private long currentId = 0; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java index 4beab61a1..d29cc1d1e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java @@ -20,12 +20,13 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; import org.springframework.batch.core.Entity; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.support.SerializationUtils; -import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.util.Assert; @@ -34,13 +35,11 @@ import org.springframework.util.Assert; */ public class MapStepExecutionDao implements StepExecutionDao { - private Map> executionsByJobExecutionId = TransactionAwareProxyFactory - .createTransactionalMap(); + private Map> executionsByJobExecutionId = new ConcurrentHashMap>(); - private Map executionsByStepExecutionId = TransactionAwareProxyFactory - .createTransactionalMap(); + private Map executionsByStepExecutionId = new ConcurrentHashMap(); - private long currentId = 0; + private AtomicLong currentId = new AtomicLong(); public void clear() { executionsByJobExecutionId.clear(); @@ -59,11 +58,11 @@ public class MapStepExecutionDao implements StepExecutionDao { Map executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId()); if (executions == null) { - executions = TransactionAwareProxyFactory.createTransactionalMap(); + executions = new ConcurrentHashMap(); executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions); } - - stepExecution.setId(currentId++); + + stepExecution.setId(currentId.incrementAndGet()); stepExecution.incrementVersion(); StepExecution copy = copy(stepExecution); executions.put(stepExecution.getId(), copy); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java index 54c40ed37..5aa9a9976 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java @@ -179,6 +179,8 @@ public class SimpleJobRepository implements JobRepository { } public void updateExecutionContext(StepExecution stepExecution) { + validateStepExecution(stepExecution); + Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)"); ecDao.updateExecutionContext(stepExecution); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java index 584eb7f96..d48826d36 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java @@ -136,7 +136,7 @@ public class FlowJobTests { @Override public FlowExecutionStatus handle(FlowExecutor executor) throws Exception { JobExecution jobExecution = executor.getJobExecution(); - jobExecution.getStepExecutions().add(new StepExecution(getName(), jobExecution)); + jobExecution.createStepExecution(getName()); if (fail) { return FlowExecutionStatus.FAILED; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java index e79eda64b..7727bed7b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/StateSupport.java @@ -16,7 +16,6 @@ package org.springframework.batch.core.job.flow; import org.springframework.batch.core.JobExecution; -import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.flow.support.state.AbstractState; /** @@ -42,7 +41,7 @@ public class StateSupport extends AbstractState { public FlowExecutionStatus handle(FlowExecutor executor) throws Exception { JobExecution jobExecution = executor.getJobExecution(); if (jobExecution != null) { - jobExecution.getStepExecutions().add(new StepExecution(getName(), jobExecution)); + jobExecution.createStepExecution(getName()); } return this.status; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java index 920cfb25e..2d695917a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/SimpleStepExecutionSplitterTests.java @@ -9,7 +9,7 @@ import java.util.Set; import org.junit.Before; import org.junit.Test; -import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; @@ -23,13 +23,15 @@ public class SimpleStepExecutionSplitterTests { private JobRepository jobRepository; - private StepExecution stepExecution = new StepExecution("bar", new JobExecution(11L)); + private StepExecution stepExecution; @Before public void setUp() throws Exception { step = new TaskletStep("step"); MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); jobRepository = (JobRepository) factory.getObject(); + stepExecution = jobRepository.createJobExecution("job", new JobParameters()).createStepExecution("bar"); + jobRepository.add(stepExecution); } @Test diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java index 497ddc58e..7fd900d4c 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java @@ -39,7 +39,7 @@ public class MapExecutionContextDaoTests extends AbstractExecutionContextDaoTest public void testPersistentCopy() throws Exception { MapExecutionContextDao tested = new MapExecutionContextDao(); JobExecution jobExecution = new JobExecution((long)1); - StepExecution stepExecution = new StepExecution("stepName", jobExecution); + StepExecution stepExecution = new StepExecution("stepName", jobExecution, 123L); assertTrue(stepExecution.getExecutionContext().isEmpty()); tested.updateExecutionContext(stepExecution); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SerializationUtils.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SerializationUtils.java index 6c6473b56..ac4f5ac81 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SerializationUtils.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/SerializationUtils.java @@ -20,6 +20,7 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; +import java.io.OptionalDataException; /** @@ -66,6 +67,9 @@ public class SerializationUtils { try { return new ObjectInputStream(new ByteArrayInputStream(bytes)).readObject(); } + catch (OptionalDataException e) { + throw new IllegalArgumentException("Could not deserialize object: eof="+e.eof+ " at length="+e.length, e); + } catch (IOException e) { throw new IllegalArgumentException("Could not deserialize object", e); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java index 4bc720eb8..df069203a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java @@ -23,6 +23,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; @@ -74,7 +75,7 @@ public class TransactionAwareProxyFactory { return (T) new HashSet((Set) target); } else if (target instanceof Map) { - return (T) new HashMap((Map) target); + return (T) new ConcurrentHashMap((Map) target); } else { throw new UnsupportedOperationException("Cannot copy target for this type: " + target.getClass()); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/MapSerializationUtilsTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/MapSerializationUtilsTests.java new file mode 100644 index 000000000..e9d6e161c --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/MapSerializationUtilsTests.java @@ -0,0 +1,54 @@ +/* + * Copyright 2006-2010 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; + +import static org.junit.Assert.assertEquals; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.junit.Test; + +/** + * @author Dave Syer + * + */ +public class MapSerializationUtilsTests { + + private Map map = new ConcurrentHashMap(); + + @Test + public void testCycle() throws Exception { + map.put("foo.bar.spam", 123); + Map result = getCopy(map); + assertEquals(map, result); + } + + @Test + public void testMultipleCycles() throws Exception { + map.put("foo.bar.spam", 123); + for (int i = 0; i < 1000; i++) { + Map result = getCopy(map); + assertEquals(map, result); + } + } + + @SuppressWarnings("unchecked") + private Map getCopy(Map map) { + return (Map) SerializationUtils.deserialize(SerializationUtils.serialize(map)); + } + +}