svn commit: r1038643 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb: KahaDBStore.java MessageDatabase.java

[gtully]

Author: gtully
Date: Wed Nov 24 15:15:39 2010
New Revision: 1038643

URL: http://svn.apache.org/viewvc?rev=1038643&view=rev
Log:
https://issues.apache.org/activemq/browse/AMQ-2985 - fix npe on remove consumer 
track new consumer last ack at lowest priority such that there can be no replay

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java?rev=1038643&r1=1038642&r2=1038643&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
 Wed Nov 24 15:15:39 2010
@@ -784,6 +784,10 @@ public class KahaDBStore extends Message
                         MessageOrderCursor moc = 
sd.subscriptionCursors.get(subscriptionKey);
                         if (moc == null) {
                             LastAck pos = sd.subscriptionAcks.get(tx, 
subscriptionKey);
+                            if (pos == null) {
+                                // sub deleted
+                                return;
+                            }
                             sd.orderIndex.setBatch(tx, pos);
                             moc = sd.orderIndex.cursor;
                         } else {

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java?rev=1038643&r1=1038642&r2=1038643&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
 Wed Nov 24 15:15:39 2010
@@ -1314,7 +1314,7 @@ public class MessageDatabase extends Ser
 
         public LastAck(long ackLocation) {
             this.lastAckedSequence = ackLocation;
-            this.priority = MessageOrderIndex.HI;
+            this.priority = MessageOrderIndex.LO;
         }
 
         public LastAck(long ackLocation, byte priority) {
@@ -2207,7 +2207,7 @@ public class MessageDatabase extends Ser
                     result = lowPriorityIndex.get(tx, key);
                     lastGetPriority = LO;
                 } else {
-                  lastGetPriority = HI;
+                    lastGetPriority = HI;
                 }
             } else {
                 lastGetPriority = DEF;