Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

revert http://fisheye6.atlassian.com/changelog/activemq/?cs=958009 - concurrent dispatch with transaction completion breaks when the cache is disabled, either through memory limit or through durable subs with existing messages in the store. cannot see the use case though as the transaction semantic is broken if a message can get dispatched but a commit can still fail. With dispatch ocurring postCommit like before, the option for kahaDB to do concurrent transaction dispatch is disabled 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@982240 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2010-08-04 12:56:42 +00:00
parent 6f7e3fc1d5
commit 0af6b063e7
6 changed files with 4 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -679,7 +679,7 @@ public class Queue extends BaseDestination implements Task, UsageListener {
context.getTransaction().addSynchronization(new Synchronization() {
@Override
public void beforeCommit() throws Exception {
public void afterCommit() throws Exception {
sendLock.lockInterruptibly();
try {
// It could take while before we receive the commit

2
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -452,7 +452,7 @@ public class Topic extends BaseDestination implements Task {
if (context.isInTransaction()) {
context.getTransaction().addSynchronization(new Synchronization() {
@Override
public void beforeCommit() throws Exception {
public void afterCommit() throws Exception {
// It could take while before we receive the commit
// operration.. by that time the message could have
// expired..

15
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
View File

@ -460,21 +460,6 @@ public class KahaDBPersistenceAdapter implements PersistenceAdapter, BrokerServi
letter.setConcurrentStoreAndDispatchTopics(concurrentStoreAndDispatch);
}
/**
* @return the concurrentStoreAndDispatchTransactions
*/
public boolean isConcurrentStoreAndDispatchTransactions() {
return letter.isConcurrentStoreAndDispatchTransactions();
}
/**
* @param concurrentStoreAndDispatchTransactions
* the concurrentStoreAndDispatchTransactions to set
*/
public void setConcurrentStoreAndDispatchTransactions(boolean concurrentStoreAndDispatchTransactions) {
letter.setConcurrentStoreAndDispatchTransactions(concurrentStoreAndDispatchTransactions);
}
public int getMaxAsyncJobs() {
return letter.getMaxAsyncJobs();
}

13
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
View File

@ -106,7 +106,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
Semaphore globalTopicSemaphore;
private boolean concurrentStoreAndDispatchQueues = true;
private boolean concurrentStoreAndDispatchTopics = true;
private boolean concurrentStoreAndDispatchTransactions = true;
private boolean concurrentStoreAndDispatchTransactions = false;
private int maxAsyncJobs = MAX_ASYNC_JOBS;
private final KahaDBTransactionStore transactionStore;
@ -155,21 +155,10 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
this.concurrentStoreAndDispatchTopics = concurrentStoreAndDispatch;
}
/**
* @return the concurrentStoreAndDispatchTransactions
*/
public boolean isConcurrentStoreAndDispatchTransactions() {
return this.concurrentStoreAndDispatchTransactions;
}
/**
* @param concurrentStoreAndDispatchTransactions
* the concurrentStoreAndDispatchTransactions to set
*/
public void setConcurrentStoreAndDispatchTransactions(boolean concurrentStoreAndDispatchTransactions) {
this.concurrentStoreAndDispatchTransactions = concurrentStoreAndDispatchTransactions;
}
/**
* @return the maxAsyncJobs
*/

1
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java
View File

@ -217,7 +217,6 @@ public class KahaDBTransactionStore implements TransactionStore {
}
/**
* @throws XAException
* @see org.apache.activemq.store.TransactionStore#commit(org.apache.activemq.service.Transaction)
*/
public void commit(TransactionId txid, boolean wasPrepared, Runnable preCommit, Runnable postCommit)

2
activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java
View File

@ -42,7 +42,6 @@ public class TransactedTopicMasterSlaveTest extends JmsTopicTransactionTest {
File dir = new File ("target" + File.separator + "slave");
KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
adapter.setDirectory(dir);
adapter.setConcurrentStoreAndDispatchTransactions(false);
broker.start();
slave = new BrokerService();
slave.setBrokerName("slave");
@ -73,7 +72,6 @@ public class TransactedTopicMasterSlaveTest extends JmsTopicTransactionTest {
File dir = new File ("target" + File.separator + "master");
KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
adapter.setDirectory(dir);
adapter.setConcurrentStoreAndDispatchTransactions(false);
BrokerService broker = new BrokerService();
broker.setBrokerName("master");
broker.setPersistenceAdapter(adapter);