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
