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 2012/09/26 00:21:44 UTC
svn commit: r1390193 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
main/java/org/apache/activemq/broker/region/TopicRegion.java
test/java/org/apache/activemq/bugs/AMQ4062Test.java
Author: tabish
Date: Tue Sep 25 22:21:44 2012
New Revision: 1390193
URL: http://svn.apache.org/viewvc?rev=1390193&view=rev
Log:
fix for: https://issues.apache.org/jira/browse/AMQ-4062
Durable consumer should be configured with the policyEntry configuration on activate.
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java (with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java?rev=1390193&r1=1390192&r2=1390193&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java Tue Sep 25 22:21:44 2012
@@ -18,7 +18,6 @@ package org.apache.activemq.broker.regio
import java.io.IOException;
import java.util.Collections;
-import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -30,6 +29,7 @@ import org.apache.activemq.broker.Broker
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
import org.apache.activemq.broker.region.cursors.StoreDurableSubscriberCursor;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.Message;
@@ -55,9 +55,9 @@ public class DurableTopicSubscription ex
private AtomicLong offlineTimestamp = new AtomicLong(-1);
public DurableTopicSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info, boolean keepDurableSubsActive)
- throws JMSException {
- super(broker,usageManager, context, info);
- this.pending = new StoreDurableSubscriberCursor(broker,context.getClientId(), info.getSubscriptionName(), info.getPrefetchSize(), this);
+ throws JMSException {
+ super(broker, usageManager, context, info);
+ this.pending = new StoreDurableSubscriberCursor(broker, context.getClientId(), info.getSubscriptionName(), info.getPrefetchSize(), this);
this.pending.setSystemUsage(usageManager);
this.pending.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark());
this.keepDurableSubsActive = keepDurableSubsActive;
@@ -80,8 +80,8 @@ public class DurableTopicSubscription ex
}
/**
- * store will have a pending ack for all durables, irrespective of the selector
- * so we need to ack if node is un-matched
+ * store will have a pending ack for all durables, irrespective of the
+ * selector so we need to ack if node is un-matched
*/
public void unmatched(MessageReference node) throws IOException {
MessageAck ack = new MessageAck();
@@ -101,23 +101,23 @@ public class DurableTopicSubscription ex
}
// do it just once per destination
if (durableDestinations.containsKey(destination.getActiveMQDestination())) {
- return;
+ return;
}
durableDestinations.put(destination.getActiveMQDestination(), destination);
if (active.get() || keepDurableSubsActive) {
- Topic topic = (Topic)destination;
+ Topic topic = (Topic) destination;
topic.activate(context, this);
if (pending.isEmpty(topic)) {
topic.recoverRetroactiveMessages(context, this);
}
- this.enqueueCounter+=pending.size();
+ this.enqueueCounter += pending.size();
} else if (destination.getMessageStore() != null) {
- TopicMessageStore store = (TopicMessageStore)destination.getMessageStore();
+ TopicMessageStore store = (TopicMessageStore) destination.getMessageStore();
try {
- this.enqueueCounter+=store.getMessageCount(subscriptionKey.getClientId(),subscriptionKey.getSubscriptionName());
+ this.enqueueCounter += store.getMessageCount(subscriptionKey.getClientId(), subscriptionKey.getSubscriptionName());
} catch (IOException e) {
- JMSException jmsEx = new JMSException("Failed to retrieve enqueueCount from store "+ e);
+ JMSException jmsEx = new JMSException("Failed to retrieve enqueueCount from store " + e);
jmsEx.setLinkedException(e);
throw jmsEx;
}
@@ -125,16 +125,24 @@ public class DurableTopicSubscription ex
dispatchPending();
}
- public void activate(SystemUsage memoryManager, ConnectionContext context,
- ConsumerInfo info) throws Exception {
+ public void activate(SystemUsage memoryManager, ConnectionContext context, ConsumerInfo info, RegionBroker regionBroker) throws Exception {
if (!active.get()) {
this.context = context;
this.info = info;
+
+ // On Activation we should update the configuration based on our new consumer info.
+ ActiveMQDestination dest = this.info.getDestination();
+ if (dest != null && regionBroker.getDestinationPolicy() != null) {
+ PolicyEntry entry = regionBroker.getDestinationPolicy().getEntryFor(dest);
+ if (entry != null) {
+ entry.configure(broker, usageManager, this);
+ }
+ }
+
LOG.debug("Activating " + this);
if (!keepDurableSubsActive) {
- for (Iterator<Destination> iter = durableDestinations.values()
- .iterator(); iter.hasNext();) {
- Topic topic = (Topic) iter.next();
+ for (Destination destination : durableDestinations.values()) {
+ Topic topic = (Topic) destination;
add(context, topic);
topic.activate(context, this);
}
@@ -148,9 +156,8 @@ public class DurableTopicSubscription ex
// If nothing was in the persistent store, then try to use the
// recovery policy.
if (pending.isEmpty()) {
- for (Iterator<Destination> iter = durableDestinations.values()
- .iterator(); iter.hasNext();) {
- Topic topic = (Topic) iter.next();
+ for (Destination destination : durableDestinations.values()) {
+ Topic topic = (Topic) destination;
topic.recoverRetroactiveMessages(context, this);
}
}
@@ -171,8 +178,8 @@ public class DurableTopicSubscription ex
pending.stop();
synchronized (dispatchLock) {
- for (Iterator<Destination> iter = durableDestinations.values().iterator(); iter.hasNext();) {
- Topic topic = (Topic)iter.next();
+ for (Destination destination : durableDestinations.values()) {
+ Topic topic = (Topic) destination;
if (!keepDurableSubsActive) {
topic.deactivate(context, this);
} else {
@@ -247,7 +254,7 @@ public class DurableTopicSubscription ex
}
protected void doAddRecoveredMessage(MessageReference message) throws Exception {
- synchronized(pending) {
+ synchronized (pending) {
pending.addRecoveredMessage(message);
}
}
@@ -275,8 +282,9 @@ public class DurableTopicSubscription ex
}
public synchronized String toString() {
- return "DurableTopicSubscription-" + getSubscriptionKey() + ", id=" + info.getConsumerId() + ", active=" + isActive() + ", destinations=" + durableDestinations.size() + ", total=" + enqueueCounter + ", pending="
- + getPendingQueueSize() + ", dispatched=" + dispatchCounter + ", inflight=" + dispatched.size() + ", prefetchExtension=" + getPrefetchExtension();
+ return "DurableTopicSubscription-" + getSubscriptionKey() + ", id=" + info.getConsumerId() + ", active=" + isActive() + ", destinations="
+ + durableDestinations.size() + ", total=" + enqueueCounter + ", pending=" + getPendingQueueSize() + ", dispatched=" + dispatchCounter
+ + ", inflight=" + dispatched.size() + ", prefetchExtension=" + getPrefetchExtension();
}
public SubscriptionKey getSubscriptionKey() {
@@ -289,7 +297,6 @@ public class DurableTopicSubscription ex
public void destroy() {
synchronized (pendingLock) {
try {
-
pending.reset();
while (pending.hasNext()) {
MessageReference node = pending.next();
@@ -301,7 +308,7 @@ public class DurableTopicSubscription ex
pending.clear();
}
}
- synchronized (dispatchLock) {
+ synchronized (dispatchLock) {
for (MessageReference node : dispatched) {
node.decrementReferenceCount();
}
@@ -321,7 +328,7 @@ public class DurableTopicSubscription ex
}
protected boolean isDropped(MessageReference node) {
- return false;
+ return false;
}
public boolean isKeepDurableSubsActive() {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java?rev=1390193&r1=1390192&r2=1390193&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicRegion.java Tue Sep 25 22:21:44 2012
@@ -16,10 +16,18 @@
*/
package org.apache.activemq.broker.region;
-import java.util.*;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
+
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
+
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.policy.PolicyEntry;
@@ -147,7 +155,7 @@ public class TopicRegion extends Abstrac
+ " subscriberName: " + key.getSubscriptionName());
}
}
- sub.activate(usageManager, context, info);
+ sub.activate(usageManager, context, info, broker);
return sub;
} else {
return super.addConsumer(context, info);
@@ -332,8 +340,6 @@ public class TopicRegion extends Abstrac
}
}
- /**
- */
private boolean hasDurableSubChanged(ConsumerInfo info1, ConsumerInfo info2) {
if (info1.getSelector() != null ^ info2.getSelector() != null) {
return true;
@@ -367,5 +373,4 @@ public class TopicRegion extends Abstrac
public boolean durableSubscriptionExists(SubscriptionKey key) {
return this.durableSubscriptions.containsKey(key);
}
-
}
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java?rev=1390193&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java Tue Sep 25 22:21:44 2012
@@ -0,0 +1,273 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.DurableTopicSubscription;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.broker.region.TopicRegion;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.util.SubscriptionKey;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AMQ4062Test {
+
+ private BrokerService service;
+ private PolicyEntry policy;
+ private ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> durableSubscriptions;
+
+ private static final int PREFETCH_SIZE_5=5;
+ private String connectionUri;
+
+ @Before
+ public void startBroker() throws IOException, Exception {
+ service=new BrokerService();
+ service.setPersistent(true);
+ service.setDeleteAllMessagesOnStartup(true);
+ service.setUseJmx(false);
+
+ KahaDBPersistenceAdapter pa=new KahaDBPersistenceAdapter();
+ File dataFile=new File("createData");
+ pa.setDirectory(dataFile);
+ pa.setJournalMaxFileLength(1024*1024*32);
+
+ service.setPersistenceAdapter(pa);
+
+ policy = new PolicyEntry();
+ policy.setTopic(">");
+ policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+ PolicyMap pMap = new PolicyMap();
+ pMap.setDefaultEntry(policy);
+
+ service.setDestinationPolicy(pMap);
+
+ service.addConnector("tcp://localhost:0");
+
+ service.start();
+ service.waitUntilStarted();
+
+ connectionUri = service.getTransportConnectors().get(0).getPublishableConnectString();
+ }
+
+ public void restartBroker() throws IOException, Exception {
+ service=new BrokerService();
+ service.setPersistent(true);
+ service.setUseJmx(false);
+
+ KahaDBPersistenceAdapter pa=new KahaDBPersistenceAdapter();
+ File dataFile=new File("createData");
+ pa.setDirectory(dataFile);
+ pa.setJournalMaxFileLength(1024*1024*32);
+
+ service.setPersistenceAdapter(pa);
+
+ policy = new PolicyEntry();
+ policy.setTopic(">");
+ policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+ PolicyMap pMap = new PolicyMap();
+ pMap.setDefaultEntry(policy);
+
+ service.setDestinationPolicy(pMap);
+ service.addConnector("tcp://localhost:0");
+ service.start();
+ service.waitUntilStarted();
+
+ connectionUri = service.getTransportConnectors().get(0).getPublishableConnectString();
+ }
+
+ @After
+ public void stopBroker() throws Exception {
+ service.stop();
+ service.waitUntilStopped();
+ service = null;
+ }
+
+ @Test
+ public void testDirableSubPrefetchRecovered() throws Exception{
+
+ PrefetchConsumer consumer=new PrefetchConsumer(true, connectionUri);
+ consumer.recieve();
+ durableSubscriptions=getDurableSubscriptions();
+ ConsumerInfo info=getConsumerInfo(durableSubscriptions);
+
+ //check if the prefetchSize equals to the size we set in the PolicyEntry
+ assertEquals(PREFETCH_SIZE_5, info.getPrefetchSize());
+
+ consumer.a.countDown();
+ Producer p=new Producer(connectionUri);
+ p.send();
+ p = null;
+
+ service.stop();
+ service.waitUntilStopped();
+ durableSubscriptions=null;
+
+ consumer = null;
+ stopBroker();
+
+ restartBroker();
+
+ getDurableSubscriptions();
+ info=null;
+ info = getConsumerInfo(durableSubscriptions);
+
+ //check if the prefetchSize equals to 0 after persistent storage recovered
+ //assertEquals(0, info.getPrefetchSize());
+
+ consumer=new PrefetchConsumer(false, connectionUri);
+ consumer.recieve();
+ consumer.a.countDown();
+
+ info=null;
+ info = getConsumerInfo(durableSubscriptions);
+
+ //check if the prefetchSize is the default size for durable consumer and the PolicyEntry
+ //we set earlier take no effect
+ //assertEquals(100, info.getPrefetchSize());
+ //info.getPrefetchSize() is 100,it should be 5,because I set the PolicyEntry as follows,
+ //policy.setDurableTopicPrefetch(PREFETCH_SIZE_5);
+ assertEquals(5, info.getPrefetchSize());
+ }
+
+ private ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> getDurableSubscriptions() throws NoSuchFieldException, IllegalAccessException {
+ if(durableSubscriptions!=null) return durableSubscriptions;
+ RegionBroker regionBroker=(RegionBroker)service.getRegionBroker();
+ TopicRegion region=(TopicRegion)regionBroker.getTopicRegion();
+ Field field=TopicRegion.class.getDeclaredField("durableSubscriptions");
+ field.setAccessible(true);
+ durableSubscriptions=(ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription>)field.get(region);
+ return durableSubscriptions;
+ }
+
+ private ConsumerInfo getConsumerInfo(ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription> durableSubscriptions) {
+ ConsumerInfo info=null;
+ for(Iterator<DurableTopicSubscription> it=durableSubscriptions.values().iterator();it.hasNext();){
+ Subscription sub = it.next();
+ info=sub.getConsumerInfo();
+ if(info.getSubscriptionName().equals(PrefetchConsumer.SUBSCRIPTION_NAME)){
+ return info;
+ }
+ }
+ return null;
+ }
+
+ public class PrefetchConsumer implements MessageListener{
+ public static final String SUBSCRIPTION_NAME = "A_NAME_ABC_DEF";
+ private String user = ActiveMQConnection.DEFAULT_USER;
+ private String password = ActiveMQConnection.DEFAULT_PASSWORD;
+ private final String uri;
+ private boolean transacted;
+ ActiveMQConnection connection;
+ Session session;
+ MessageConsumer consumer;
+ private boolean needAck=false;
+ CountDownLatch a=new CountDownLatch(1);
+
+ public PrefetchConsumer(boolean needAck, String uri){
+ this.needAck=needAck;
+ this.uri = uri;
+ }
+
+ public void recieve() throws Exception{
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, uri);
+ connection = (ActiveMQConnection)connectionFactory.createConnection();
+ connection.setClientID("3");
+ connection.start();
+
+ session = connection.createSession(transacted, Session.CLIENT_ACKNOWLEDGE);
+ Destination destination = session.createTopic("topic2");
+ consumer = session.createDurableSubscriber((Topic)destination,SUBSCRIPTION_NAME);
+ consumer.setMessageListener(this);
+ }
+
+ public void onMessage(Message message) {
+ try {
+ a.await();
+ } catch (InterruptedException e1) {
+ }
+ if(needAck){
+ try {
+ message.acknowledge();
+ consumer.close();
+ session.close();
+ connection.close();
+ } catch (JMSException e) {
+ }
+ }
+ }
+ }
+
+ public class Producer {
+
+ protected final String user = ActiveMQConnection.DEFAULT_USER;
+
+ private String password = ActiveMQConnection.DEFAULT_PASSWORD;
+ private final String uri;
+ private boolean transacted;
+
+ public Producer(String uri) {
+ this.uri = uri;
+ }
+
+ public void send() throws Exception{
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, uri);
+ ActiveMQConnection connection = (ActiveMQConnection)connectionFactory.createConnection();
+ connection.start();
+
+ ActiveMQSession session = (ActiveMQSession)connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createTopic("topic2");
+ MessageProducer producer = session.createProducer(destination);
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ for(int i=0;i<100;i++){
+ TextMessage om=session.createTextMessage("hello from producer");
+ producer.send(om);
+ }
+ producer.close();
+ session.close();
+ connection.close();
+ }
+ }
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ4062Test.java
------------------------------------------------------------------------------
svn:eol-style = native