RESOLVED - issue BATCH-1541, BATCH-1542: Thread safety for map daos
This commit is contained in:
dsyer
@@ -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<String> processed = new ArrayList<String>(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<String> {
|
||||
|
||||
private List<String> written = new ArrayList<String>();
|
||||
private List<String> written = new CopyOnWriteArrayList<String>();
|
||||
|
||||
private Collection<String> failures = Collections.emptySet();
|
||||
|
||||
@@ -243,7 +251,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private List<String> processed = new ArrayList<String>();
|
||||
private List<String> processed = new CopyOnWriteArrayList<String>();
|
||||
|
||||
private SimpleJdbcTemplate jdbcTemplate;
|
||||
|
||||
@@ -253,6 +261,13 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
|
||||
public SkipProcessorStub(DataSource dataSource) {
|
||||
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the processed
|
||||
*/
|
||||
public List<String> getProcessed() {
|
||||
return processed;
|
||||
}
|
||||
|
||||
public List<String> getCommitted() {
|
||||
return jdbcTemplate.query("SELECT MESSAGE from ERROR_LOG where STEP_NAME='processed'",
|
||||
|
||||
@@ -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<String, String> 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<String, String>();
|
||||
|
||||
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<String> processed = new ArrayList<String>(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<String> {
|
||||
|
||||
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<String> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private List<String> written = new CopyOnWriteArrayList<String>();
|
||||
|
||||
private Collection<String> failures = Collections.emptySet();
|
||||
|
||||
public void setFailures(String... failures) {
|
||||
this.failures = Arrays.asList(failures);
|
||||
}
|
||||
|
||||
public List<String> getWritten() {
|
||||
return written;
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
written.clear();
|
||||
}
|
||||
|
||||
public void write(List<? extends String> 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<String, String> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private List<String> processed = new CopyOnWriteArrayList<String>();
|
||||
|
||||
public List<String> 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<Class<? extends Throwable>, Boolean> getExceptionMap(Class<? extends Throwable>... args) {
|
||||
Map<Class<? extends Throwable>, Boolean> map = new HashMap<Class<? extends Throwable>, Boolean>();
|
||||
for (Class<? extends Throwable> arg : args) {
|
||||
map.put(arg, true);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, String> 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<String, String>();
|
||||
|
||||
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<String> committed = new ArrayList<String>(writer.getWritten());
|
||||
Collections.sort(committed);
|
||||
assertEquals("[1, 2, 3, 4, 5]", committed.toString());
|
||||
List<String> processed = new ArrayList<String>(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<String> {
|
||||
|
||||
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<String> {
|
||||
|
||||
private List<String> written = new CopyOnWriteArrayList<String>();
|
||||
|
||||
private Collection<String> failures = Collections.emptySet();
|
||||
|
||||
public List<String> getWritten() {
|
||||
return written;
|
||||
}
|
||||
|
||||
public void clear() {
|
||||
written.clear();
|
||||
}
|
||||
|
||||
public void write(List<? extends String> 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<String, String> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
private List<String> processed = new CopyOnWriteArrayList<String>();
|
||||
|
||||
public List<String> 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;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<StepExecution> completionService = new ExecutorCompletionService<StepExecution>(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<StepExecution>() {
|
||||
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<StepExecution> 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<StepExecution> 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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
|
||||
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">
|
||||
|
||||
<job id="job" xmlns="http://www.springframework.org/schema/batch">
|
||||
<split id="split" task-executor="taskExecutor">
|
||||
<flow>
|
||||
<step id="step1" parent="step" />
|
||||
</flow>
|
||||
<flow>
|
||||
<step id="step2" parent="step" />
|
||||
</flow>
|
||||
</split>
|
||||
</job>
|
||||
|
||||
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
|
||||
<property name="maxPoolSize" value="6"/>
|
||||
<property name="queueCapacity" value="0"/>
|
||||
</bean>
|
||||
|
||||
<step id="step" xmlns="http://www.springframework.org/schema/batch">
|
||||
<tasklet>
|
||||
<beans:bean class="org.springframework.batch.core.test.step.SplitJobMapRepositoryIntegrationTests$CountingTasklet" />
|
||||
</tasklet>
|
||||
</step>
|
||||
|
||||
<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
|
||||
<property name="jobRepository" ref="jobRepository" />
|
||||
</bean>
|
||||
|
||||
<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
|
||||
<property name="transactionManager" ref="transactionManager" />
|
||||
</bean>
|
||||
|
||||
<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
|
||||
|
||||
</beans>
|
||||
Reference in New Issue
Block a user