Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert to thread safe but less robust in the face of transaction exceptions MapStepExecutionDao

Browse Source
This commit is contained in:
dsyer
2010-09-11 08:04:58 +00:00
parent a53acbd6c9
commit c503a19992
8 changed files with 170 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/IgnoredTestSuite.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 2009-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;
import org.junit.Ignore;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import org.junit.runners.Suite.SuiteClasses;
import org.springframework.batch.core.test.step.FaultTolerantStepFactoryBeanIntegrationTests;
import org.springframework.batch.core.test.step.FaultTolerantStepFactoryBeanRollbackIntegrationTests;
/**
* A test suite that is ignored, but can be resurrected to help debug ordering
* issues in tests.
*
* @author Dave Syer
*
*/
@RunWith(Suite.class)
@SuiteClasses(value = { FaultTolerantStepFactoryBeanIntegrationTests.class, FaultTolerantStepFactoryBeanRollbackIntegrationTests.class })
@Ignore
public class IgnoredTestSuite {
}

4
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanTests.java → spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanIntegrationTests.java
View File

@@ -44,7 +44,7 @@ import org.springframework.util.Assert;
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanTests {
public class FaultTolerantStepFactoryBeanIntegrationTests {
private static final int MAX_COUNT = 1000;
@@ -90,6 +90,8 @@ public class FaultTolerantStepFactoryBeanTests {
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
SimpleJdbcTestUtils.deleteFromTables(new SimpleJdbcTemplate(dataSource), "ERROR_LOG");
}

4
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackTests.java → spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java
View File

@@ -47,7 +47,7 @@ import org.springframework.util.Assert;
*/
@ContextConfiguration(locations = "/simple-job-launcher-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanRollbackTests {
public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
private static final int MAX_COUNT = 1000;
@@ -98,6 +98,8 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
SimpleJdbcTestUtils.deleteFromTables(new SimpleJdbcTemplate(dataSource), "ERROR_LOG");
}
@Test

2
spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java
View File

@@ -34,7 +34,7 @@ public class Entity implements Serializable {
private Long id;
private Integer version;
private volatile Integer version;
public Entity() {
super();

8
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java
View File

@@ -21,13 +21,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;
import org.springframework.util.ReflectionUtils;
@@ -37,9 +37,9 @@ import org.springframework.util.ReflectionUtils;
*/
public class MapStepExecutionDao implements StepExecutionDao {
private Map<Long, Map<Long, StepExecution>> executionsByJobExecutionId = TransactionAwareProxyFactory.createAppendOnlyTransactionalMap();
private Map<Long, Map<Long, StepExecution>> executionsByJobExecutionId = new ConcurrentHashMap<Long, Map<Long,StepExecution>>();
private Map<Long, StepExecution> executionsByStepExecutionId = TransactionAwareProxyFactory.createAppendOnlyTransactionalMap();
private Map<Long, StepExecution> executionsByStepExecutionId = new ConcurrentHashMap<Long, StepExecution>();
private AtomicLong currentId = new AtomicLong();
@@ -71,7 +71,7 @@ public class MapStepExecutionDao implements StepExecutionDao {
Map<Long, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
if (executions == null) {
executions = TransactionAwareProxyFactory.createAppendOnlyTransactionalMap();
executions = new ConcurrentHashMap<Long, StepExecution>();
executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions);
}

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

@@ -23,13 +23,10 @@ import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.UnexpectedRollbackException;
import org.springframework.transaction.interceptor.RollbackRuleAttribute;
import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttributeEditor;
import org.springframework.transaction.support.DefaultTransactionStatus;
import org.springframework.util.StringUtils;
/**
@@ -473,40 +470,6 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
}
@Test
public void testTransactionException() throws Exception {
ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager() {
private boolean failed = false;
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
if (writer.getWritten().isEmpty() || failed || !isExistingTransaction(status.getTransaction())) {
super.doCommit(status);
return;
}
failed = true;
status.setRollbackOnly();
super.doRollback(status);
throw new UnexpectedRollbackException("Planned");
}
};
MapJobRepositoryFactoryBean repositoryFactory = new MapJobRepositoryFactoryBean();
repositoryFactory.setTransactionManager(transactionManager);
repositoryFactory.afterPropertiesSet();
repository = (JobRepository) repositoryFactory.getObject();
factory.setJobRepository(repository);
factory.setTransactionManager(transactionManager);
jobExecution = repository.createJobExecution("skipJob", new JobParameters());
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[]", writer.getCommitted().toString());
}
@SuppressWarnings("unchecked")
private Collection<Class<? extends Throwable>> getExceptionList(Class<? extends Throwable> arg) {
return Arrays.<Class<? extends Throwable>> asList(arg);

89
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanUnexpectedRollbackTests.java Normal file
View File

@@ -0,0 +1,89 @@
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertEquals;
import java.util.Arrays;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.UnexpectedRollbackException;
import org.springframework.transaction.support.DefaultTransactionStatus;
/**
* Tests for {@link FaultTolerantStepFactoryBean} with unexpected rollback.
*/
@ContextConfiguration(locations="classpath:/org/springframework/batch/core/repository/dao/data-source-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class FaultTolerantStepFactoryBeanUnexpectedRollbackTests {
protected final Log logger = LogFactory.getLog(getClass());
@Autowired
private DataSource dataSource;
@Test
public void testTransactionException() throws Exception {
final SkipWriterStub<String> writer = new SkipWriterStub<String>();
FaultTolerantStepFactoryBean<String, String> factory = new FaultTolerantStepFactoryBean<String, String>();
factory.setItemWriter(writer);
DataSourceTransactionManager transactionManager = new DataSourceTransactionManager(dataSource) {
private boolean failed = false;
protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
if (writer.getWritten().isEmpty() || failed || !isExistingTransaction(status.getTransaction())) {
super.doCommit(status);
return;
}
failed = true;
status.setRollbackOnly();
super.doRollback(status);
throw new UnexpectedRollbackException("Planned");
}
};
factory.setBeanName("stepName");
factory.setTransactionManager(transactionManager);
factory.setCommitInterval(2);
ItemReader<String> reader = new ListItemReader<String>(Arrays.asList("1", "2"));
factory.setItemReader(reader);
JobRepositoryFactoryBean repositoryFactory = new JobRepositoryFactoryBean();
repositoryFactory.setDataSource(dataSource);
repositoryFactory.setTransactionManager(transactionManager);
repositoryFactory.afterPropertiesSet();
JobRepository repository = (JobRepository) repositoryFactory.getObject();
factory.setJobRepository(repository);
JobExecution jobExecution = repository.createJobExecution("job", new JobParameters());
StepExecution stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
Step step = (Step) factory.getObject();
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[]", writer.getCommitted().toString());
}
}

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

@@ -87,29 +87,31 @@ public class TransactionAwareProxyFactory<T> {
* @return an independent copy
*/
@SuppressWarnings("unchecked")
protected synchronized final T begin(T target) {
protected 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();
synchronized (target) {
if (target instanceof List) {
if (appendOnly) {
return (T) new ArrayList();
}
return (T) new ArrayList((List) target);
}
return (T) new ArrayList((List) target);
}
else if (target instanceof Set) {
if (appendOnly) {
return (T) new HashSet();
else if (target instanceof Set) {
if (appendOnly) {
return (T) new HashSet();
}
return (T) new HashSet((Set) target);
}
return (T) new HashSet((Set) target);
}
else if (target instanceof Map) {
if (appendOnly) {
return (T) new HashMap();
else if (target instanceof Map) {
if (appendOnly) {
return (T) new HashMap();
}
return (T) new HashMap((Map) target);
}
else {
throw new UnsupportedOperationException("Cannot copy target for this type: " + target.getClass());
}
return (T) new HashMap((Map) target);
}
else {
throw new UnsupportedOperationException("Cannot copy target for this type: " + target.getClass());
}
}
@@ -122,20 +124,22 @@ public class TransactionAwareProxyFactory<T> {
* @param target the original target of the factory.
*/
@SuppressWarnings("unchecked")
protected synchronized void commit(T copy, T target) {
protected void commit(T copy, T target) {
// Unfortunately in Java 5 this method has to be synchronized
// (works OK without in Java 6).
if (target instanceof Collection) {
if (!appendOnly) {
((Collection) target).clear();
synchronized (target) {
if (target instanceof Collection) {
if (!appendOnly) {
((Collection) target).clear();
}
((Collection) target).addAll((Collection) copy);
}
((Collection) target).addAll((Collection) copy);
}
else {
if (!appendOnly) {
((Map) target).clear();
else {
if (!appendOnly) {
((Map) target).clear();
}
((Map) target).putAll((Map) copy);
}
((Map) target).putAll((Map) copy);
}
}