You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2013/03/11 20:18:24 UTC
svn commit: r1455290 - in /activemq/trunk:
activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java
Author: chirino
Date: Mon Mar 11 19:18:23 2013
New Revision: 1455290
URL: http://svn.apache.org/r1455290
Log:
Added at test case and fix for AMQ-4351.
Avoids deadlock by not holding on to the pending and dispatch locks when we try to detach a durable sub from a topic.
Added:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java
Modified:
activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
Modified: activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java?rev=1455290&r1=1455289&r2=1455290&view=diff
==============================================================================
--- activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java (original)
+++ activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java Mon Mar 11 19:18:23 2013
@@ -17,6 +17,7 @@
package org.apache.activemq.broker.region;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -188,6 +189,8 @@ public class DurableTopicSubscription ex
active.set(false);
offlineTimestamp.set(System.currentTimeMillis());
this.usageManager.getMemoryUsage().removeUsageListener(this);
+
+ ArrayList<Topic> topicsToDeactivate = new ArrayList<Topic>();
synchronized (pendingLock) {
pending.stop();
@@ -195,7 +198,7 @@ public class DurableTopicSubscription ex
for (Destination destination : durableDestinations.values()) {
Topic topic = (Topic) destination;
if (!keepDurableSubsActive) {
- topic.deactivate(context, this);
+ topicsToDeactivate.add(topic);
} else {
topic.getDestinationStatistics().getInflight().subtract(dispatched.size());
}
@@ -236,6 +239,9 @@ public class DurableTopicSubscription ex
}
}
}
+ for(Topic topic: topicsToDeactivate) {
+ topic.deactivate(context, this);
+ }
prefetchExtension.set(0);
}
Added: activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java?rev=1455290&view=auto
==============================================================================
--- activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java (added)
+++ activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/broker/AMQ4351Test.java Mon Mar 11 19:18:23 2013
@@ -0,0 +1,233 @@
+package org.apache.activemq.broker;
+
+import junit.framework.Test;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter;
+import org.apache.derby.jdbc.EmbeddedDataSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Implements the test case attached to:
+ * https://issues.apache.org/jira/browse/AMQ-4351
+ *
+ * This version avoids the spring deps.
+ */
+public class AMQ4351Test extends BrokerTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AMQ4351Test.class);
+
+ public static Test suite() {
+ return suite(AMQ4351Test.class);
+ }
+
+ public static void main(String[] args) {
+ junit.textui.TestRunner.run(suite());
+ }
+
+ protected BrokerService createBroker() throws Exception {
+ BrokerService broker = new BrokerService();
+
+ // Lets clean up often.
+ broker.setOfflineDurableSubscriberTaskSchedule(500);
+ broker.setOfflineDurableSubscriberTimeout(2000); // lets delete durable subs much faster.
+
+ JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
+ EmbeddedDataSource dataSource = new EmbeddedDataSource();
+ dataSource.setDatabaseName("derbyDb");
+ dataSource.setCreateDatabase("create");
+ jdbc.setDataSource(dataSource);
+
+ jdbc.deleteAllMessages();
+ broker.setPersistenceAdapter(jdbc);
+ return broker;
+ }
+
+ ActiveMQConnectionFactory connectionFactory;
+ ActiveMQTopic destination = new ActiveMQTopic("TEST");
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ connectionFactory = new ActiveMQConnectionFactory(broker.getVmConnectorURI());
+ }
+
+ class ProducingClient implements Runnable {
+ final AtomicLong size = new AtomicLong();
+ final AtomicBoolean done = new AtomicBoolean();
+ CountDownLatch doneLatch = new CountDownLatch(1);
+
+ Connection connection;
+ Session session;
+ MessageProducer producer;
+
+ ProducingClient() throws JMSException {
+ connection = connectionFactory.createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ producer = session.createProducer(destination);
+ }
+
+ private void sendMessage() {
+ try {
+ producer.send(session.createTextMessage("Test"));
+ long i = size.incrementAndGet();
+ if( (i % 1000) == 0 ) {
+ LOG.info("produced " + i + ".");
+ }
+ } catch (JMSException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void start() {
+ new Thread(this, "ProducingClient").start();
+ }
+
+ public void stop() throws InterruptedException {
+ done.set(true);
+ if( !doneLatch.await(20, TimeUnit.MILLISECONDS) ) {
+ try {
+ connection.close();
+ doneLatch.await();
+ } catch (JMSException e) {
+ }
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ try {
+ while (!done.get()) {
+ sendMessage();
+ Thread.sleep(10);
+ }
+ } finally {
+ connection.close();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ done.set(true);
+ } finally {
+ doneLatch.countDown();
+ }
+ }
+ }
+
+ class ConsumingClient implements Runnable {
+ final String name;
+ final AtomicLong size = new AtomicLong();
+ final AtomicBoolean done = new AtomicBoolean();
+ CountDownLatch doneLatch = new CountDownLatch(1);
+
+ public ConsumingClient(String name) {
+ this.name = name;
+ }
+
+ public void start() {
+ LOG.info("Starting JMS listener " + name);
+ new Thread(this, "ConsumingClient: "+name).start();
+ }
+
+ public void stopAsync() {
+ done.set(true);
+ }
+
+ public void stop() throws InterruptedException {
+ stopAsync();
+ doneLatch.await();
+ }
+
+ @Override
+ public void run() {
+ try {
+ Connection connection = connectionFactory.createConnection();
+ connection.setClientID(name);
+ connection.start();
+ try {
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageConsumer consumer = session.createDurableSubscriber(destination, name, null, false);
+ while( !done.get() ) {
+ Message msg = consumer.receive(100);
+ if(msg!=null ) {
+ size.incrementAndGet();
+ session.commit();
+ }
+ }
+ } finally {
+ connection.close();
+ LOG.info("Stopped JMS listener " + name);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ done.set(true);
+ } finally {
+ doneLatch.countDown();
+ }
+ }
+
+ }
+
+ public void testAMQ4351() throws InterruptedException, JMSException {
+ LOG.info("Start test.");
+
+ ProducingClient producer = new ProducingClient();
+ ConsumingClient listener1 = new ConsumingClient("subscriber-1");
+ ConsumingClient listener2 = new ConsumingClient("subscriber-2");
+ ConsumingClient listener3 = new ConsumingClient("subscriber-3");
+ try {
+
+ listener1.start();
+ listener2.start();
+ listener3.start();
+ int subs = 100;
+
+ List<ConsumingClient> subscribers = new ArrayList<ConsumingClient>(subs);
+ for (int i = 4; i < subs; i++) {
+ ConsumingClient client = new ConsumingClient("subscriber-" + i);
+ subscribers.add(client);
+ client.start();
+ }
+
+ LOG.info("All subscribers started.");
+ producer.sendMessage();
+
+ LOG.info("Stopping 97 subscribers....");
+ for (ConsumingClient client : subscribers) {
+ client.stopAsync();
+ }
+
+ // Start producing messages for 10 minutes, at high rate
+ LOG.info("Starting mass message producer...");
+ producer.start();
+
+
+ long lastSize = listener1.size.get();
+ for( int i=0 ; i < 10; i++ ) {
+ Thread.sleep(1000);
+ long size = listener1.size.get();
+ LOG.info("Listener 1: consumed: "+(size - lastSize));
+ assertTrue( size > lastSize );
+ lastSize = size;
+ }
+ } finally {
+ LOG.info("Stopping clients");
+ listener1.stop();
+ listener2.stop();
+ listener3.stop();
+ producer.stop();
+ }
+ }
+
+}