activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r1039108 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/store/jdbc/ main/java/org/apache/activemq/store/jdbc/adapter/ test/java/org/apache/activemq/usecases/
Date Thu, 25 Nov 2010 17:29:02 GMT
Author: gtully
Date: Thu Nov 25 17:29:02 2010
New Revision: 1039108

URL: http://svn.apache.org/viewvc?rev=1039108&view=rev
Log:
resolve regression with offline durable subs and jdbc, if there are no messages in the store
and some consumed messages, the durable is not reactivated and subsequent messages can get
deleted before they are consumed. Reduce the duplicate replays to a topic store cursor by
tracking the last recovered. additional test

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/Statements.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java?rev=1039108&r1=1039107&r2=1039108&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCTopicMessageStore.java
Thu Nov 25 17:29:02 2010
@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.sql.SQLException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.activemq.ActiveMQMessageAudit;
@@ -43,6 +44,7 @@ import org.apache.commons.logging.LogFac
 public class JDBCTopicMessageStore extends JDBCMessageStore implements TopicMessageStore
{
 
     private static final Log LOG = LogFactory.getLog(JDBCTopicMessageStore.class);
+    private Map<String, LastRecovered> subscriberLastRecoveredMap = new ConcurrentHashMap<String,
LastRecovered>();
 
     public JDBCTopicMessageStore(JDBCPersistenceAdapter persistenceAdapter, JDBCAdapter adapter,
WireFormat wireFormat, ActiveMQTopic topic, ActiveMQMessageAudit audit) {
         super(persistenceAdapter, adapter, wireFormat, topic, audit);
@@ -100,17 +102,37 @@ public class JDBCTopicMessageStore exten
         }
     }
 
+    private class LastRecovered {
+        long sequence = 0;
+        byte priority = 9;
+
+        public void update(long sequence, Message msg) {
+            this.sequence = sequence;
+            this.priority = msg.getPriority();
+        }
+
+        public String toString() {
+            return "" + sequence + ":" + priority;
+        }
+    }
+
     public synchronized void recoverNextMessages(final String clientId, final String subscriptionName,
final int maxReturned, final MessageRecoveryListener listener)
-        throws Exception {
+            throws Exception {
         TransactionContext c = persistenceAdapter.getTransactionContext();
-        try {
-             JDBCMessageRecoveryListener jdbcListener = new JDBCMessageRecoveryListener()
{
 
+        String key = getSubscriptionKey(clientId, subscriptionName);
+        if (!subscriberLastRecoveredMap.containsKey(key)) {
+           subscriberLastRecoveredMap.put(key, new LastRecovered());
+        }
+        final LastRecovered lastRecovered = subscriberLastRecoveredMap.get(key);        
+        try {
+            JDBCMessageRecoveryListener jdbcListener = new JDBCMessageRecoveryListener()
{
                 public boolean recoverMessage(long sequenceId, byte[] data) throws Exception
{
                     if (listener.hasSpace()) {
-                        Message msg = (Message)wireFormat.unmarshal(new ByteSequence(data));
+                        Message msg = (Message) wireFormat.unmarshal(new ByteSequence(data));
                         msg.getMessageId().setBrokerSequenceId(sequenceId);
                         if (listener.recoverMessage(msg)) {
+                            lastRecovered.update(sequenceId, msg);
                             return true;
                         }
                     }
@@ -124,10 +146,10 @@ public class JDBCTopicMessageStore exten
             };
             if (isPrioritizedMessages()) {
                 adapter.doRecoverNextMessagesWithPriority(c, destination, clientId, subscriptionName,
-                    0, 0, maxReturned, jdbcListener);
+                        lastRecovered.sequence, lastRecovered.priority, maxReturned, jdbcListener);
             } else {
                 adapter.doRecoverNextMessages(c, destination, clientId, subscriptionName,
-                    0, 0, maxReturned, jdbcListener);
+                        lastRecovered.sequence, 0, maxReturned, jdbcListener);
             }
         } catch (SQLException e) {
             JDBCPersistenceAdapter.log("JDBC Failure: ", e);
@@ -137,7 +159,7 @@ public class JDBCTopicMessageStore exten
     }
 
     public void resetBatching(String clientId, String subscriptionName) {
-        // DB always recovers from last ack
+        subscriberLastRecoveredMap.remove(getSubscriptionKey(clientId, subscriptionName));
     }
 
     public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive) throws
IOException {

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/Statements.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/Statements.java?rev=1039108&r1=1039107&r2=1039108&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/Statements.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/Statements.java
Thu Nov 25 17:29:02 2010
@@ -261,6 +261,7 @@ public class Statements {
                                               + getFullAckTableName() + " D "
                                               + " WHERE D.CONTAINER=? AND D.CLIENT_ID=? AND
D.SUB_NAME=?"
                                               + " AND M.CONTAINER=D.CONTAINER AND M.ID >
D.LAST_ACKED_ID"
+                                              + " AND M.ID > ?"
                                               + " ORDER BY M.ID";
         }
         return findDurableSubMessagesStatement;
@@ -273,6 +274,7 @@ public class Statements {
                                               + " WHERE D.CONTAINER=? AND D.CLIENT_ID=? AND
D.SUB_NAME=?"
                                               + " AND M.CONTAINER=D.CONTAINER"
                                               + " AND M.PRIORITY=D.PRIORITY AND M.ID >
D.LAST_ACKED_ID"
+                                              + " AND ( (M.ID > ?) OR (M.PRIORITY <
?) )" 
                                               + " ORDER BY M.PRIORITY DESC, M.ID";
         }
         return findDurableSubMessagesByPriorityStatement;
@@ -343,7 +345,7 @@ public class Statements {
 
     public String getFindAllDestinationsStatement() {
         if (findAllDestinationsStatement == null) {
-            findAllDestinationsStatement = "SELECT DISTINCT CONTAINER FROM " + getFullMessageTableName();
+            findAllDestinationsStatement = "SELECT DISTINCT CONTAINER FROM " + getFullAckTableName();
         }
         return findAllDestinationsStatement;
     }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java?rev=1039108&r1=1039107&r2=1039108&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/DefaultJDBCAdapter.java
Thu Nov 25 17:29:02 2010
@@ -169,10 +169,6 @@ public class DefaultJDBCAdapter implemen
             long seq2 = 0;
             if (rs.next()) {
                 seq2 = rs.getLong(1);
-                // if there is no such message, ignore the value
-                if (this.doGetMessageById(c, seq2) == null) {
-                    seq2 = 0;
-                }
             }
             long seq = Math.max(seq1, seq2);
             return seq;
@@ -512,6 +508,7 @@ public class DefaultJDBCAdapter implemen
             s.setString(1, destination.getQualifiedName());
             s.setString(2, clientId);
             s.setString(3, subscriptionName);
+            s.setLong(4, seq);
             rs = s.executeQuery();
             int count = 0;
             if (this.statements.isUseExternalMessageReferences()) {
@@ -542,12 +539,12 @@ public class DefaultJDBCAdapter implemen
         cleanupExclusiveLock.readLock().lock();
         try {
             s = c.getConnection().prepareStatement(this.statements.getFindDurableSubMessagesByPriorityStatement());
-            // maxRows needs to be twice prefetch as the db will replay all unacked, so inflight
messages will
-            // be returned and suppressed by the cursor audit. It is faster this way.
             s.setMaxRows(maxRows);
             s.setString(1, destination.getQualifiedName());
             s.setString(2, clientId);
             s.setString(3, subscriptionName);
+            s.setLong(4, seq);
+            s.setLong(5, priority);
             rs = s.executeQuery();
             int count = 0;
             if (this.statements.isUseExternalMessageReferences()) {

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java?rev=1039108&r1=1039107&r2=1039108&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
Thu Nov 25 17:29:02 2010
@@ -617,6 +617,11 @@ public class DurableSubscriptionOfflineT
         assertEquals(0, listener.count);
     }
 
+    public void initCombosForTestOfflineSubscriptionWithSelectorAfterRestart() throws Exception
{
+        this.addCombinationValues("defaultPersistenceAdapter",
+                new Object[]{ PersistenceAdapterChoice.KahaDB, PersistenceAdapterChoice.JDBC});
+    }
+    
     public void testOfflineSubscriptionWithSelectorAfterRestart() throws Exception {
         // create offline subs 1
         Connection con = createConnection("offCli1");
@@ -702,6 +707,72 @@ public class DurableSubscriptionOfflineT
         assertEquals(filtered, listener3.count);
     }
 
+    public void initCombosForTestOfflineAfterRestart() throws Exception {
+        this.addCombinationValues("defaultPersistenceAdapter",
+                new Object[]{ PersistenceAdapterChoice.KahaDB, PersistenceAdapterChoice.JDBC});
+    }
+
+    public void testOfflineSubscriptionAfterRestart() throws Exception {
+        // create offline subs 1
+        Connection con = createConnection("offCli1");
+        Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer consumer = session.createDurableSubscriber(topic, "SubsId", null,
false);
+        Listener listener = new Listener();
+        consumer.setMessageListener(listener);
+
+        // send messages
+        MessageProducer producer = session.createProducer(null);
+
+        int sent = 0;
+        for (int i = 0; i < 10; i++) {
+            sent++;
+            Message message = session.createMessage();
+            message.setStringProperty("filter", "false");
+            producer.send(topic, message);
+        }
+
+        LOG.info("sent: " + sent);
+        Thread.sleep(5 * 1000);
+        session.close();
+        con.close();
+
+        assertEquals(sent, listener.count);
+
+        // restart broker
+        Thread.sleep(3 * 1000);
+        broker.stop();
+        createBroker(false /*deleteAllMessages*/);
+
+        // send more messages
+        con = createConnection();
+        session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        producer = session.createProducer(null);
+
+        for (int i = 0; i < 10; i++) {
+            sent++;
+            Message message = session.createMessage();
+            message.setStringProperty("filter", "false");
+            producer.send(topic, message);
+        }
+
+        LOG.info("after restart, sent: " + sent);
+        Thread.sleep(1 * 1000);
+        session.close();
+        con.close();
+
+        // test offline subs
+        con = createConnection("offCli1");
+        session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        consumer = session.createDurableSubscriber(topic, "SubsId", null, true);
+        consumer.setMessageListener(listener);
+
+        Thread.sleep(3 * 1000);
+
+        session.close();
+        con.close();
+
+        assertEquals(sent, listener.count);
+    }
 
     public void testInterleavedOfflineSubscriptionCanConsumeAfterUnsub() throws Exception
{
         // create offline subs 1



Mime
View raw message