You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2013/06/03 19:00:38 UTC
svn commit: r1489062 - in
/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb:
KahaDBStore.java MessageDatabase.java
Author: tabish
Date: Mon Jun 3 17:00:38 2013
New Revision: 1489062
URL: http://svn.apache.org/r1489062
Log:
Additional AMQ-4563: Since the version of the OpenWire marshalers is configured, lets persist what it was when the store was created and log a warning should we start a broker with an old store configured to use a higher openwire version than originally set.
Modified:
activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
Modified: activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java?rev=1489062&r1=1489061&r2=1489062&view=diff
==============================================================================
--- activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java (original)
+++ activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java Mon Jun 3 17:00:38 2013
@@ -26,11 +26,20 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.Map.Entry;
-import java.util.concurrent.*;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.RegionBroker;
@@ -55,12 +64,14 @@ import org.apache.activemq.store.TopicMe
import org.apache.activemq.store.TransactionStore;
import org.apache.activemq.store.kahadb.data.KahaAddMessageCommand;
import org.apache.activemq.store.kahadb.data.KahaDestination;
+import org.apache.activemq.store.kahadb.data.KahaDestination.DestinationType;
import org.apache.activemq.store.kahadb.data.KahaLocation;
import org.apache.activemq.store.kahadb.data.KahaRemoveDestinationCommand;
import org.apache.activemq.store.kahadb.data.KahaRemoveMessageCommand;
import org.apache.activemq.store.kahadb.data.KahaSubscriptionCommand;
import org.apache.activemq.store.kahadb.data.KahaTransactionInfo;
-import org.apache.activemq.store.kahadb.data.KahaDestination.DestinationType;
+import org.apache.activemq.store.kahadb.disk.journal.Location;
+import org.apache.activemq.store.kahadb.disk.page.Transaction;
import org.apache.activemq.usage.MemoryUsage;
import org.apache.activemq.usage.SystemUsage;
import org.apache.activemq.util.ServiceStopper;
@@ -68,8 +79,6 @@ import org.apache.activemq.util.ThreadPo
import org.apache.activemq.wireformat.WireFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.activemq.store.kahadb.disk.journal.Location;
-import org.apache.activemq.store.kahadb.disk.page.Transaction;
public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
static final Logger LOG = LoggerFactory.getLogger(KahaDBStore.class);
@@ -96,7 +105,7 @@ public class KahaDBStore extends Message
// when true, message order may be compromised when cache is exhausted if store is out
// or order w.r.t cache
private boolean concurrentStoreAndDispatchTopics = false;
- private boolean concurrentStoreAndDispatchTransactions = false;
+ private final boolean concurrentStoreAndDispatchTransactions = false;
private int maxAsyncJobs = MAX_ASYNC_JOBS;
private final KahaDBTransactionStore transactionStore;
private TransactionIdTransformer transactionIdTransformer;
@@ -116,9 +125,11 @@ public class KahaDBStore extends Message
return "KahaDB:[" + directory.getAbsolutePath() + "]";
}
+ @Override
public void setBrokerName(String brokerName) {
}
+ @Override
public void setUsageManager(SystemUsage usageManager) {
this.usageManager = usageManager;
}
@@ -177,30 +188,49 @@ public class KahaDBStore extends Message
@Override
public void doStart() throws Exception {
- if( brokerService!=null ) {
- wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
+ if (brokerService != null) {
+ metadata.openwireVersion = brokerService.getStoreOpenWireVersion();
+ wireFormat.setVersion(metadata.openwireVersion);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Store OpenWire version configured as: {}", metadata.openwireVersion);
+ }
+
}
super.doStart();
+
+ if (brokerService != null) {
+ // In case the recovered store used a different OpenWire version log a warning
+ // to assist in determining why journal reads fail.
+ if (metadata.openwireVersion != brokerService.getStoreOpenWireVersion()) {
+ LOG.warn("Receovered Store uses a different OpenWire version[{}] " +
+ "than the version configured[{}].",
+ metadata.openwireVersion, brokerService.getStoreOpenWireVersion());
+ }
+ }
+
this.globalQueueSemaphore = new Semaphore(getMaxAsyncJobs());
this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs());
this.asyncQueueJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.asyncTopicJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.queueExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
- asyncQueueJobQueue, new ThreadFactory() {
- public Thread newThread(Runnable runnable) {
- Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch");
- thread.setDaemon(true);
- return thread;
- }
- });
+ asyncQueueJobQueue, new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable runnable) {
+ Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch");
+ thread.setDaemon(true);
+ return thread;
+ }
+ });
this.topicExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
- asyncTopicJobQueue, new ThreadFactory() {
- public Thread newThread(Runnable runnable) {
- Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch");
- thread.setDaemon(true);
- return thread;
- }
- });
+ asyncTopicJobQueue, new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable runnable) {
+ Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch");
+ thread.setDaemon(true);
+ return thread;
+ }
+ });
}
@Override
@@ -252,6 +282,7 @@ public class KahaDBStore extends Message
super.doStop(stopper);
}
+ @Override
void incrementRedeliveryAndReWrite(final String key, final KahaDestination destination) throws IOException {
Location location;
this.indexLock.writeLock().lock();
@@ -277,6 +308,7 @@ public class KahaDBStore extends Message
this.indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<IOException>() {
+ @Override
public void execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(destination, tx);
Long sequence = sd.messageIdIndex.get(tx, key);
@@ -306,6 +338,7 @@ public class KahaDBStore extends Message
private Location findMessageLocation(final String key, final KahaDestination destination) throws IOException {
return pageFile.tx().execute(new Transaction.CallableClosure<Location, IOException>() {
+ @Override
public Location execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(destination, tx);
Long sequence = sd.messageIdIndex.get(tx, key);
@@ -347,6 +380,7 @@ public class KahaDBStore extends Message
this.topicExecutor.execute(task);
}
+ @Override
public TransactionStore createTransactionStore() throws IOException {
return this.transactionStore;
}
@@ -423,6 +457,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void addMessage(ConnectionContext context, Message message) throws IOException {
KahaAddMessageCommand command = new KahaAddMessageCommand();
command.setDestination(dest);
@@ -436,6 +471,7 @@ public class KahaDBStore extends Message
}
+ @Override
public void removeMessage(ConnectionContext context, MessageAck ack) throws IOException {
KahaRemoveMessageCommand command = new KahaRemoveMessageCommand();
command.setDestination(dest);
@@ -447,12 +483,14 @@ public class KahaDBStore extends Message
store(command, isEnableJournalDiskSyncs() && ack.isResponseRequired(), null, null);
}
+ @Override
public void removeAllMessages(ConnectionContext context) throws IOException {
KahaRemoveDestinationCommand command = new KahaRemoveDestinationCommand();
command.setDestination(dest);
store(command, true, null, null);
}
+ @Override
public Message getMessage(MessageId identity) throws IOException {
final String key = identity.toProducerKey();
@@ -473,12 +511,14 @@ public class KahaDBStore extends Message
return loadMessage(location);
}
+ @Override
public int getMessageCount() throws IOException {
try {
lockAsyncJobQueue();
indexLock.writeLock().lock();
try {
return pageFile.tx().execute(new Transaction.CallableClosure<Integer, IOException>() {
+ @Override
public Integer execute(Transaction tx) throws IOException {
// Iterate through all index entries to get a count
// of
@@ -506,6 +546,7 @@ public class KahaDBStore extends Message
indexLock.writeLock().lock();
try {
return pageFile.tx().execute(new Transaction.CallableClosure<Boolean, IOException>() {
+ @Override
public Boolean execute(Transaction tx) throws IOException {
// Iterate through all index entries to get a count of
// messages in the destination.
@@ -518,11 +559,13 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void recover(final MessageRecoveryListener listener) throws Exception {
// recovery may involve expiry which will modify
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
public void execute(Transaction tx) throws Exception {
StoredDestination sd = getStoredDestination(dest, tx);
sd.orderIndex.resetCursorPosition();
@@ -543,10 +586,12 @@ public class KahaDBStore extends Message
}
+ @Override
public void recoverNextMessages(final int maxReturned, final MessageRecoveryListener listener) throws Exception {
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
public void execute(Transaction tx) throws Exception {
StoredDestination sd = getStoredDestination(dest, tx);
Entry<Long, MessageKeys> entry = null;
@@ -571,11 +616,13 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void resetBatching() {
if (pageFile.isLoaded()) {
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
public void execute(Transaction tx) throws Exception {
StoredDestination sd = getExistingStoredDestination(dest, tx);
if (sd != null) {
@@ -603,6 +650,7 @@ public class KahaDBStore extends Message
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<IOException>() {
+ @Override
public void execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(dest, tx);
Long location = sd.messageIdIndex.get(tx, key);
@@ -678,6 +726,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
MessageId messageId, MessageAck ack)
throws IOException {
@@ -721,6 +770,7 @@ public class KahaDBStore extends Message
store(command, false, null, null);
}
+ @Override
public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive) throws IOException {
String subscriptionKey = subscriptionKey(subscriptionInfo.getClientId(), subscriptionInfo
.getSubscriptionName());
@@ -734,6 +784,7 @@ public class KahaDBStore extends Message
this.subscriptionCount.incrementAndGet();
}
+ @Override
public void deleteSubscription(String clientId, String subscriptionName) throws IOException {
KahaSubscriptionCommand command = new KahaSubscriptionCommand();
command.setDestination(dest);
@@ -742,12 +793,14 @@ public class KahaDBStore extends Message
this.subscriptionCount.decrementAndGet();
}
+ @Override
public SubscriptionInfo[] getAllSubscriptions() throws IOException {
final ArrayList<SubscriptionInfo> subscriptions = new ArrayList<SubscriptionInfo>();
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<IOException>() {
+ @Override
public void execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(dest, tx);
for (Iterator<Entry<String, KahaSubscriptionCommand>> iterator = sd.subscriptions.iterator(tx); iterator
@@ -769,11 +822,13 @@ public class KahaDBStore extends Message
return rc;
}
+ @Override
public SubscriptionInfo lookupSubscription(String clientId, String subscriptionName) throws IOException {
final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
indexLock.writeLock().lock();
try {
return pageFile.tx().execute(new Transaction.CallableClosure<SubscriptionInfo, IOException>() {
+ @Override
public SubscriptionInfo execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(dest, tx);
KahaSubscriptionCommand command = sd.subscriptions.get(tx, subscriptionKey);
@@ -789,11 +844,13 @@ public class KahaDBStore extends Message
}
}
+ @Override
public int getMessageCount(String clientId, String subscriptionName) throws IOException {
final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
indexLock.writeLock().lock();
try {
return pageFile.tx().execute(new Transaction.CallableClosure<Integer, IOException>() {
+ @Override
public Integer execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(dest, tx);
LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
@@ -810,6 +867,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void recoverSubscription(String clientId, String subscriptionName, final MessageRecoveryListener listener)
throws Exception {
final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
@@ -818,6 +876,7 @@ public class KahaDBStore extends Message
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
public void execute(Transaction tx) throws Exception {
StoredDestination sd = getStoredDestination(dest, tx);
LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
@@ -838,6 +897,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void recoverNextMessages(String clientId, String subscriptionName, final int maxReturned,
final MessageRecoveryListener listener) throws Exception {
final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
@@ -846,6 +906,7 @@ public class KahaDBStore extends Message
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<Exception>() {
+ @Override
public void execute(Transaction tx) throws Exception {
StoredDestination sd = getStoredDestination(dest, tx);
sd.orderIndex.resetCursorPosition();
@@ -889,12 +950,14 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void resetBatching(String clientId, String subscriptionName) {
try {
final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<IOException>() {
+ @Override
public void execute(Transaction tx) throws IOException {
StoredDestination sd = getStoredDestination(dest, tx);
sd.subscriptionCursors.remove(subscriptionKey);
@@ -913,10 +976,12 @@ public class KahaDBStore extends Message
return clientId + ":" + subscriptionName;
}
+ @Override
public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException {
return this.transactionStore.proxy(new KahaDBMessageStore(destination));
}
+ @Override
public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException {
return this.transactionStore.proxy(new KahaDBTopicMessageStore(destination));
}
@@ -928,6 +993,7 @@ public class KahaDBStore extends Message
* @param destination
* Destination to forget
*/
+ @Override
public void removeQueueMessageStore(ActiveMQQueue destination) {
}
@@ -938,19 +1004,23 @@ public class KahaDBStore extends Message
* @param destination
* Destination to forget
*/
+ @Override
public void removeTopicMessageStore(ActiveMQTopic destination) {
}
+ @Override
public void deleteAllMessages() throws IOException {
deleteAllMessages = true;
}
+ @Override
public Set<ActiveMQDestination> getDestinations() {
try {
final HashSet<ActiveMQDestination> rc = new HashSet<ActiveMQDestination>();
indexLock.writeLock().lock();
try {
pageFile.tx().execute(new Transaction.Closure<IOException>() {
+ @Override
public void execute(Transaction tx) throws IOException {
for (Iterator<Entry<String, StoredDestination>> iterator = metadata.destinations.iterator(tx); iterator
.hasNext();) {
@@ -983,10 +1053,12 @@ public class KahaDBStore extends Message
}
}
+ @Override
public long getLastMessageBrokerSequenceId() throws IOException {
return 0;
}
+ @Override
public long getLastProducerSequenceId(ProducerId id) {
indexLock.readLock().lock();
try {
@@ -996,6 +1068,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public long size() {
try {
return journalSize.get() + getPageFile().getDiskSize();
@@ -1004,16 +1077,20 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void beginTransaction(ConnectionContext context) throws IOException {
throw new IOException("Not yet implemented.");
}
+ @Override
public void commitTransaction(ConnectionContext context) throws IOException {
throw new IOException("Not yet implemented.");
}
+ @Override
public void rollbackTransaction(ConnectionContext context) throws IOException {
throw new IOException("Not yet implemented.");
}
+ @Override
public void checkpoint(boolean sync) throws IOException {
super.checkpointCleanup(sync);
}
@@ -1158,6 +1235,7 @@ public class KahaDBStore extends Message
return this.future;
}
+ @Override
public boolean cancel() {
if (this.done.compareAndSet(false, true)) {
return this.future.cancel(false);
@@ -1165,6 +1243,7 @@ public class KahaDBStore extends Message
return false;
}
+ @Override
public void aquireLocks() {
if (this.locked.compareAndSet(false, true)) {
try {
@@ -1178,6 +1257,7 @@ public class KahaDBStore extends Message
}
+ @Override
public void releaseLocks() {
if (this.locked.compareAndSet(true, false)) {
store.releaseLocalAsyncLock();
@@ -1186,6 +1266,7 @@ public class KahaDBStore extends Message
}
}
+ @Override
public void run() {
this.store.doneTasks++;
try {
@@ -1304,6 +1385,7 @@ public class KahaDBStore extends Message
super(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, queue, threadFactory);
}
+ @Override
protected void afterExecute(Runnable runnable, Throwable throwable) {
super.afterExecute(runnable, throwable);
Modified: activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java?rev=1489062&r1=1489061&r2=1489062&view=diff
==============================================================================
--- activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java (original)
+++ activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java Mon Jun 3 17:00:38 2013
@@ -54,6 +54,7 @@ import org.apache.activemq.broker.Broker
import org.apache.activemq.broker.BrokerServiceAware;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.TransactionId;
+import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.protobuf.Buffer;
import org.apache.activemq.store.kahadb.data.KahaAckMessageFileMapCommand;
import org.apache.activemq.store.kahadb.data.KahaAddMessageCommand;
@@ -123,6 +124,8 @@ public abstract class MessageDatabase ex
protected transient ActiveMQMessageAuditNoSync producerSequenceIdTracker = new ActiveMQMessageAuditNoSync();
protected transient Map<Integer, Set<Integer>> ackMessageFileMap = new HashMap<Integer, Set<Integer>>();
protected int version = VERSION;
+ protected int openwireVersion = OpenWireFormat.DEFAULT_VERSION;
+
public void read(DataInput is) throws IOException {
state = is.readInt();
destinations = new BTreeIndex<String, StoredDestination>(pageFile, is.readLong());
@@ -145,15 +148,20 @@ public abstract class MessageDatabase ex
} catch (EOFException expectedOnUpgrade) {
}
try {
- version = is.readInt();
+ version = is.readInt();
} catch (EOFException expectedOnUpgrade) {
- version=1;
+ version = 1;
}
if (version >= 5 && is.readBoolean()) {
ackMessageFileMapLocation = LocationMarshaller.INSTANCE.readPayload(is);
} else {
ackMessageFileMapLocation = null;
}
+ try {
+ openwireVersion = is.readInt();
+ } catch (EOFException expectedOnUpgrade) {
+ openwireVersion = OpenWireFormat.DEFAULT_VERSION;
+ }
LOG.info("KahaDB is version " + version);
}
@@ -188,6 +196,7 @@ public abstract class MessageDatabase ex
} else {
os.writeBoolean(false);
}
+ os.writeInt(this.openwireVersion);
}
}