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

IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)

Browse Source 
http://jira.springframework.org/browse/BATCH-7

Remove BatchTransactionSynchronizationManager
This commit is contained in:
dsyer
2008-01-31 22:55:54 +00:00
parent 22e3e64fbc
commit 68d0eacdc7
8 changed files with 51 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepScope.java
View File

@@ -72,10 +72,10 @@ public class StepScope implements Scope, BeanFactoryPostProcessor, Ordered {
scopedObject = context.getAttribute(name);
if (scopedObject == null) {
scopedObject = objectFactory.getObject();
context.setAttribute(name, scopedObject);
if (scopedObject instanceof StepContextAware) {
((StepContextAware) scopedObject).setStepContext(context);
}
context.setAttribute(name, scopedObject);
}
}
}

1
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/DrivingQueryItemReader.java
View File

@@ -22,7 +22,6 @@ import org.springframework.batch.io.support.AbstractTransactionalIoSource;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

186
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManager.java
View File

@@ -1,186 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.repeat.synch;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.core.AttributeAccessor;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* <p>
* Contains static methods for registering objects for transaction
* synchronization. Because there are many non-standard inputs that need to be
* aware of transactions, such as file input, this facade provides a hook into
* Spring TransactionSyncrhonization. The spring class
* TransactionSynchronizationManager has public static methods that are used by
* the AbstractPlatformTransactionManager to ensure other resources are
* synchronizaed with it's transaction. This means that any spring transaction
* manager which extends the afore mentioned abstract class will be notified of
* changes in a transaction. For more information on the type of transaction
* events that can be handled, please see the TransactionSynchronization
* interface.
* </p>
*
* <p>
* Spring's intended use for the TransactionSyncrhonizationManager is that any
* class that wishes can register for the current transaction only. When commit
* or rollback is called, this list will be used to notify interested classes.
* However, once a new transaction is obtained, this list will be cleared. This
* is problematic for batch processing, since input templates need to always be
* made aware of transaction events, without being forced to register every
* time. To solve this issue, classes should register with the
* BatchTransactionFacade, which will ensure that any time a new transaction is
* obtained, they are re-registered with spring's transaction synchronization.
* </p>
*
* @author Lucas Ward
* @author Dave Syer
*
* @see TransactionSynchronizationManager
* @see RepeatSynchronizationManager
* @see TransactionSynchronization
*/
public class BatchTransactionSynchronizationManager {
/**
* The key in the context attributes for the list of synchronizations.
*/
private static final String SYNCHS_ATTR_KEY = BatchTransactionSynchronizationManager.class.getName()
+ ".SYNCHRONIZATIONS";
/**
* Static method to register synchronizations. A TransactionSyncrhonization
* object will be added to the internal list within a threadLocal. After
* ensuring that there is a reference for later re-synchronization, the
* object is added to spring's TransactionSynchronizationManager.
*/
public static void registerSynchronization(TransactionSynchronization synchronization) {
List synchs = (List) getSynchronizations();
if (!synchs.contains(synchronization)) {
synchs.add(synchronization);
if (TransactionSynchronizationManager.isSynchronizationActive()
&& !TransactionSynchronizationManager.getSynchronizations().contains(synchronization)) {
TransactionSynchronizationManager.registerSynchronization(synchronization);
}
}
}
/**
* The internal list of synchronizations is iterated, and each
* synchronization object is registered with the
* TransactionSynchronizationManager again. This is necessary because any
* call to PlatformTransactionManager.getTransaction() will result in a
* clearing of the synchronizationManager's list.
*/
public static void resynchronize() {
List batchSynchs = (List) getSynchronizations();
if (batchSynchs != null) {
for (Iterator it = batchSynchs.iterator(); it.hasNext();) {
TransactionSynchronization synchronization = (TransactionSynchronization) it.next();
if (TransactionSynchronizationManager.isSynchronizationActive()
&& !TransactionSynchronizationManager.getSynchronizations().contains(synchronization)) {
TransactionSynchronizationManager.registerSynchronization(synchronization);
}
}
}
}
/**
* Set the synchronizations list to null. Usually called when the step is
* complete, to ensure no issues when the next step is called within the
* same thread, which should only happen when running out of container. Does
* not throw an exception if there is no batch context.
*/
public static void clearSynchronizations() {
AttributeAccessor context = getContext();
if (context == null) {
return; // Nothing to do
}
setSynchronizations(null);
}
/**
* Set the current synchronizations to the given list.
* @param synchs a list of {@link TransactionSynchronization} instances.
* @throws IllegalStateException if there is no batch context available.
*/
private static void setSynchronizations(List synchs) {
AttributeAccessor context = getContext();
if (context == null) {
return;
}
context.setAttribute(SYNCHS_ATTR_KEY, synchs);
}
/**
* Get the current list of synchronizations if there is one.
*
* @return a list of {@link TransactionSynchronization} instances or null.
*
* @throws IllegalStateException if there is no batch context available.
*/
private static List getSynchronizations() {
AttributeAccessor context = getContext();
if (context == null) {
// N.B. this returns a modifiable list on purpose - it is used
// internally to set up the list if there is no context available.
return new ArrayList();
}
List synchs = (List) context.getAttribute(SYNCHS_ATTR_KEY);
if (synchs == null) {
synchs = new ArrayList();
}
setSynchronizations(synchs);
return synchs;
}
/**
* @return the current context as an {@link AttributeAccessor}.
*/
private static AttributeAccessor getContext() {
RepeatContext context = RepeatSynchronizationManager.getContext();
return getSynchContext(context);
}
/**
* Locate the context that will contain the synchronisations by walking up
* the context hierarchy until the synchronisations are found or the top is
* reached.
*
* @param context
* @return
*/
private static AttributeAccessor getSynchContext(RepeatContext context) {
if (context == null || context.hasAttribute(SYNCHS_ATTR_KEY) || context.getParent() == null) {
return context;
}
return getSynchContext(context.getParent());
}
}

3
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/sql/AbstractJdbcItemReaderIntegrationTests.java
View File

@@ -7,7 +7,6 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.stream.GenericStreamContext;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
@@ -35,14 +34,12 @@ public abstract class AbstractJdbcItemReaderIntegrationTests extends AbstractTra
protected void onSetUp()throws Exception{
super.onSetUp();
BatchTransactionSynchronizationManager.clearSynchronizations();
source = createItemReader();
getAsInitializingBean(source).afterPropertiesSet();
}
protected void onTearDown()throws Exception {
getAsDisposableBean(source).destroy();
BatchTransactionSynchronizationManager.clearSynchronizations();
super.onTearDown();
}

3
spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/AbstractDataSourceItemReaderIntegrationTests.java
View File

@@ -8,7 +8,6 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.stream.GenericStreamContext;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
@@ -39,7 +38,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr
*/
protected void onSetUpInTransaction() throws Exception {
super.onSetUpInTransaction();
BatchTransactionSynchronizationManager.clearSynchronizations();
source = createItemReader();
}
@@ -48,7 +46,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Abstr
*/
protected void onTearDownAfterTransaction() throws Exception {
getAsDisposableBean(source).destroy();
BatchTransactionSynchronizationManager.clearSynchronizations();
super.onTearDownAfterTransaction();
}

143
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/synch/BatchTransactionSynchronizationManagerTests.java
View File

@@ -1,143 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.repeat.synch;
import java.util.List;
import junit.framework.TestCase;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
public class BatchTransactionSynchronizationManagerTests extends TestCase {
private TransactionSynchronization synchronization = new TransactionSynchronizationAdapter() {
};
protected void setUp() throws Exception {
super.setUp();
BatchTransactionSynchronizationManager.clearSynchronizations();
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.clearSynchronization();
}
TransactionSynchronizationManager.initSynchronization();
RepeatSynchronizationManager.register(new RepeatContextSupport(null));
}
/*
* (non-Javadoc)
* @see junit.framework.TestCase#tearDown()
*/
protected void tearDown() throws Exception {
super.tearDown();
RepeatSynchronizationManager.clear();
}
public void testRegisterWhenContextMissing() throws Exception {
RepeatSynchronizationManager.clear();
try {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(1, synchronizations.size());
}
catch (IllegalStateException e) {
fail("Unexpected IllegalStateException");
// Unexpected -
}
}
public void testRegisterSynchronization() {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
// There should be only one transaction synchronization object in the
// list.
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(1, synchronizations.size());
assertSame(synchronizations.get(0), synchronization);
if (RepeatSynchronizationManager.getContext() != null) {
assertEquals(1, RepeatSynchronizationManager.getContext().attributeNames().length);
}
}
public void testRegisterSynchronizationWithParentContext() {
RepeatSynchronizationManager.register(new RepeatContextSupport(RepeatSynchronizationManager.getContext()));
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
// There should be only one transaction synchronization object in the
// list.
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(1, synchronizations.size());
assertSame(synchronizations.get(0), synchronization);
assertEquals(0, RepeatSynchronizationManager.getContext().attributeNames().length);
assertEquals(1, RepeatSynchronizationManager.getContext().getParent().attributeNames().length);
}
public void testSynchronizeTwiceWithSameObject() throws Exception {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
testRegisterSynchronization();
}
public void testSynchronizeTwiceWithSameObjectAndNoContext() throws Exception {
RepeatSynchronizationManager.clear();
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
testRegisterSynchronization();
}
public void testReregisterSynchronization() {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.initSynchronization();
BatchTransactionSynchronizationManager.resynchronize();
// There should be only one transaction synchronization object in the
// list.
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(1, synchronizations.size());
assertSame(synchronizations.get(0), synchronization);
}
public void testResynchronizeWithNoSynchronizations() throws Exception {
BatchTransactionSynchronizationManager.resynchronize();
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(0, synchronizations.size());
}
public void testSynchronizeWhenNotInTransaction() throws Exception {
TransactionSynchronizationManager.clearSynchronization();
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
TransactionSynchronizationManager.initSynchronization();
BatchTransactionSynchronizationManager.resynchronize();
// There should be only one transaction synchronization object in the
// list.
List synchronizations = TransactionSynchronizationManager.getSynchronizations();
assertEquals(1, synchronizations.size());
assertSame(synchronizations.get(0), synchronization);
}
}

96
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java
View File

@@ -11,8 +11,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.execution.scope.StepContext;
import org.springframework.batch.execution.scope.StepContextAware;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.KeyedItemReader;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.sample.item.writer.StagingItemWriter;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.dao.OptimisticLockingFailureException;
@@ -20,12 +21,10 @@ import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader, DisposableBean,
public class StagingItemReader extends JdbcDaoSupport implements ItemStream, KeyedItemReader, DisposableBean,
StepContextAware {
// Key for buffer in transaction synchronization manager
@@ -43,8 +42,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader
private volatile Iterator keys;
private final TransactionSynchronization synchronization = new StagingInputTransactionSynchronization();
/**
*
* @see org.springframework.batch.io.driving.DrivingQueryItemReader#close()
@@ -151,8 +148,8 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader
getBuffer().add(next);
key = next;
logger.debug("Retrieved key from list: " + key);
Assert.state(TransactionSynchronizationManager.getSynchronizations().contains(synchronization),
"Appropriate transaction synchronization not registered for this thread.");
Assert.state(TransactionSynchronizationManager.isActualTransactionActive(),
"Transaction not active for this thread.");
}
}
}
@@ -166,7 +163,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader
private StagingBuffer getBuffer() {
if (!TransactionSynchronizationManager.hasResource(BUFFER_KEY)) {
TransactionSynchronizationManager.bindResource(BUFFER_KEY, new StagingBuffer());
registerSynchronization();
}
return (StagingBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY);
}
@@ -175,48 +171,6 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader
return false;
}
/**
* Register for Synchronization. This method is left protected because
* clients of this class should not be registering for synchronization, but
* rather only subclasses, at the appropriate time, i.e. when they are not
* initialized.
*/
protected void registerSynchronization() {
BatchTransactionSynchronizationManager.registerSynchronization(synchronization);
}
/*
* Called when a transaction has been committed.
*
* @see TransactionSynchronization#afterCompletion
*/
protected void transactionCommitted() {
getBuffer().commit();
}
/*
* Called when a transaction has been rolled back.
*
* @see TransactionSynchronization#afterCompletion
*/
protected void transactionRolledBack() {
getBuffer().rollback();
}
/**
* Encapsulates transaction events handling.
*/
private class StagingInputTransactionSynchronization extends TransactionSynchronizationAdapter {
public void afterCompletion(int status) {
if (status == TransactionSynchronization.STATUS_ROLLED_BACK) {
transactionRolledBack();
}
else if (status == TransactionSynchronization.STATUS_COMMITTED) {
transactionCommitted();
}
}
}
private class StagingBuffer {
private List list = new ArrayList();
@@ -250,4 +204,44 @@ public class StagingItemReader extends JdbcDaoSupport implements KeyedItemReader
}
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#isMarkSupported()
*/
public boolean isMarkSupported() {
return true;
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext)
*/
public void mark(StreamContext streamContext) {
getBuffer().commit();
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext)
*/
public void reset(StreamContext streamContext) {
getBuffer().rollback();
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.StreamContext)
*/
public void restoreFrom(StreamContext context) {
// no-op
}
/*
* (non-Javadoc)
* @see org.springframework.batch.item.StreamContextProvider#getStreamContext()
*/
public StreamContext getStreamContext() {
return new StreamContext();
}
}

13
spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/StagingItemReaderTests.java
View File

@@ -9,7 +9,6 @@ import org.springframework.batch.execution.scope.SimpleStepContext;
import org.springframework.batch.execution.scope.StepContext;
import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.batch.sample.item.writer.StagingItemWriter;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
@@ -87,12 +86,10 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
public void testProviderRollsBackMultipleTimes() throws Exception {
provider.mark(null);
setComplete();
endTransaction();
startNewTransaction();
// After a rollback we have to resynchronize the TX to simulate a real
// batch
BatchTransactionSynchronizationManager.resynchronize();
int count = getJdbcTemplate().queryForInt("SELECT COUNT(*) from BATCH_STAGING where JOB_ID=? AND PROCESSED=?",
new Object[] { jobId, StagingItemWriter.NEW });
@@ -103,9 +100,9 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
item = provider.read();
assertEquals("BAR", item);
provider.reset(null);
endTransaction();
startNewTransaction();
BatchTransactionSynchronizationManager.resynchronize();
item = provider.read();
assertEquals("FOO", item);
@@ -114,9 +111,9 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
item = provider.read();
assertEquals("SPAM", item);
provider.reset(null);
endTransaction();
startNewTransaction();
BatchTransactionSynchronizationManager.resynchronize();
item = provider.read();
assertEquals("FOO", item);
@@ -125,12 +122,12 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
public void testProviderRollsBackProcessIndicator() throws Exception {
provider.mark(null);
setComplete();
endTransaction();
startNewTransaction();
// After a rollback we have to resynchronize the TX to simulate a real
// batch
BatchTransactionSynchronizationManager.resynchronize();
long id = getJdbcTemplate().queryForLong("SELECT MIN(ID) from BATCH_STAGING where JOB_ID=?",
new Object[] { jobId });
@@ -141,11 +138,11 @@ public class StagingItemReaderTests extends AbstractTransactionalDataSourceSprin
Object item = provider.read();
assertEquals("FOO", item);
provider.reset(null);
endTransaction();
startNewTransaction();
// After a rollback we have to resynchronize the TX to simulate a real
// batch
BatchTransactionSynchronizationManager.resynchronize();
String after = (String) getJdbcTemplate().queryForObject("SELECT PROCESSED from BATCH_STAGING where ID=?",
new Object[] { new Long(id) }, String.class);