You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2010/02/24 13:53:38 UTC
svn commit: r915770 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/
main/java/org/apache/activemq/broker/region/cursors/
main/java/org/apache/activemq/broker/region/policy/
main/java/org/apache/activemq/store/kahadb...
Author: gtully
Date: Wed Feb 24 12:53:37 2010
New Revision: 915770
URL: http://svn.apache.org/viewvc?rev=915770&view=rev
Log:
resolve https://issues.apache.org/activemq/browse/AMQ-2626 - ensure reference count is decremented on non evicted messages
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java (with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/plist/PListStore.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ThreadTracker.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/TopicSubscriptionTest.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java Wed Feb 24 12:53:37 2010
@@ -284,13 +284,13 @@
if (warnOnProducerFlowControl) {
warnOnProducerFlowControl = false;
- LOG.info("Usage Manager memory limit reached for " + getActiveMQDestination().getQualifiedName()
+ LOG.info("Usage Manager memory limit ("+ memoryUsage.getLimit() + ") reached for " + getActiveMQDestination().getQualifiedName()
+ ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it."
+ " See http://activemq.apache.org/producer-flow-control.html for more info");
}
if (systemUsage.isSendFailIfNoSpace()) {
- throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding "
+ throw new javax.jms.ResourceAllocationException("Usage Manager memory limit ("+ memoryUsage.getLimit() + ") reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding "
+ getActiveMQDestination().getQualifiedName() + "." + " See http://activemq.apache.org/producer-flow-control.html for more info");
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java Wed Feb 24 12:53:37 2010
@@ -88,7 +88,7 @@
dispatch(node);
slowConsumer=false;
} else {
- //we are slow
+ //we are slow
if(!slowConsumer) {
slowConsumer=true;
for (Destination dest: destinations) {
@@ -124,8 +124,11 @@
LinkedList<MessageReference> list = null;
MessageReference[] oldMessages=null;
synchronized(matched){
- list = matched.pageInList(pageInSize);
+ list = matched.pageInList(pageInSize);
oldMessages = messageEvictionStrategy.evictMessages(list);
+ for (MessageReference ref : list) {
+ ref.decrementReferenceCount();
+ }
}
int messagesToEvict = 0;
if (oldMessages != null){
@@ -478,17 +481,5 @@
public int getPrefetchSize() {
return (int)info.getPrefetchSize();
}
-
- /**
- * Get the list of inflight messages
- * @return the list
- */
- public synchronized List<MessageReference> getInFlightMessages(){
- List<MessageReference> result = new ArrayList<MessageReference>();
- synchronized(matched) {
- result.addAll(matched.pageInList(1000));
- }
- return result;
- }
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java Wed Feb 24 12:53:37 2010
@@ -211,7 +211,7 @@
void destroy() throws Exception;
/**
- * Page in a restricted number of messages
+ * Page in a restricted number of messages and increment the reference count
*
* @param maxItems
* @return a list of paged in messages
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java Wed Feb 24 12:53:37 2010
@@ -147,6 +147,11 @@
}
public void configure(Broker broker, SystemUsage memoryManager, TopicSubscription subscription) {
+ //override prefetch size if not set by the Consumer
+ int prefetch=subscription.getConsumerInfo().getPrefetchSize();
+ if (prefetch == ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH){
+ subscription.getConsumerInfo().setPrefetchSize(getTopicPrefetch());
+ }
if (pendingMessageLimitStrategy != null) {
int value = pendingMessageLimitStrategy.getMaximumPendingMessageLimit(subscription);
int consumerLimit = subscription.getInfo().getMaximumPendingMessageLimit();
@@ -167,11 +172,6 @@
}
if (pendingSubscriberPolicy != null) {
String name = subscription.getContext().getClientId() + "_" + subscription.getConsumerInfo().getConsumerId();
- //override prefetch size if not set by the Consumer
- int prefetch=subscription.getConsumerInfo().getPrefetchSize();
- if (prefetch == ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH){
- subscription.getConsumerInfo().setPrefetchSize(getTopicPrefetch());
- }
int maxBatchSize = subscription.getConsumerInfo().getPrefetchSize();
subscription.setMatched(pendingSubscriberPolicy.getSubscriberPendingMessageCursor(broker,name, maxBatchSize));
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/plist/PListStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/plist/PListStore.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/plist/PListStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadb/plist/PListStore.java Wed Feb 24 12:53:37 2010
@@ -176,8 +176,10 @@
}
public long size() {
- if (!initialized) {
- return 0;
+ synchronized (this) {
+ if (!initialized) {
+ return 0;
+ }
}
try {
return journal.getDiskSize() + pageFile.getDiskSize();
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/usage/Usage.java Wed Feb 24 12:53:37 2010
@@ -240,7 +240,7 @@
private void fireEvent(final int oldPercentUsage, final int newPercentUsage) {
if (debug) {
- LOG.debug("Memory usage change from: " + oldPercentUsage + "% of available memory, to: "
+ LOG.debug(getName() + ": usage change from: " + oldPercentUsage + "% of available memory, to: "
+ newPercentUsage + "% of available memory");
}
if (started.get()) {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ThreadTracker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ThreadTracker.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ThreadTracker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ThreadTracker.java Wed Feb 24 12:53:37 2010
@@ -43,12 +43,14 @@
* output the result of stack trace capture to the log
*/
public static void result() {
- for (Entry<String, Tracker> t: trackers.entrySet()) {
- LOG.info("Tracker: " + t.getKey() + ", " + t.getValue().size() + " entry points...");
- for (Trace trace : t.getValue().values()) {
- LOG.info("count: " + trace.count, trace);
+ synchronized(trackers) {
+ for (Entry<String, Tracker> t: trackers.entrySet()) {
+ LOG.info("Tracker: " + t.getKey() + ", " + t.getValue().size() + " entry points...");
+ for (Trace trace : t.getValue().values()) {
+ LOG.info("count: " + trace.count, trace);
+ }
+ LOG.info("Tracker: " + t.getKey() + ", done.");
}
- LOG.info("Tracker: " + t.getKey() + ", done.");
}
}
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java?rev=915770&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java Wed Feb 24 12:53:37 2010
@@ -0,0 +1,199 @@
+/**
+ * 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;
+
+import static junit.framework.Assert.fail;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.broker.region.policy.ConstantPendingMessageLimitStrategy;
+import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy;
+import org.apache.activemq.broker.region.policy.PendingMessageLimitStrategy;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.broker.region.policy.PrefetchRatePendingMessageLimitStrategy;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.util.ThreadTracker;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MessageEvictionTest {
+ static final Log LOG = LogFactory.getLog(MessageEvictionTest.class);
+ private BrokerService broker;
+ private ConnectionFactory connectionFactory;
+ Connection connection;
+ private Session session;
+ private Topic destination;
+ protected int numMessages = 4000;
+ protected String payload = new String(new byte[1024*2]);
+
+ @Before
+ public void setUp() throws Exception {
+ broker = createBroker();
+ broker.start();
+ connectionFactory = createConnectionFactory();
+ connection = connectionFactory.createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ destination = session.createTopic("verifyEvection");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ ThreadTracker.result();
+ connection.stop();
+ broker.stop();
+ }
+
+ @Test
+ public void testMessageEvictionMemoryUsage() throws Exception {
+ ExecutorService executor = Executors.newCachedThreadPool();
+ final CountDownLatch doAck = new CountDownLatch(1);
+ final CountDownLatch consumerRegistered = new CountDownLatch(1);
+ executor.execute(new Runnable() {
+ public void run() {
+ try {
+ final MessageConsumer consumer = session.createConsumer(destination);
+ consumer.setMessageListener(new MessageListener() {
+ public void onMessage(Message message) {
+ try {
+ // very slow, only ack once
+ doAck.await(60, TimeUnit.SECONDS);
+ message.acknowledge();
+ } catch (Exception e) {
+ e.printStackTrace();
+ consumerRegistered.countDown();
+ fail(e.toString());
+ }
+ }
+ });
+ consumerRegistered.countDown();
+ doAck.await(60, TimeUnit.SECONDS);
+ consumer.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.toString());
+ }
+ }
+ });
+
+ assertTrue("we have a consumer", consumerRegistered.await(10, TimeUnit.SECONDS));
+
+ final AtomicInteger sent = new AtomicInteger(0);
+ final CountDownLatch sendDone = new CountDownLatch(1);
+ executor.execute(new Runnable() {
+ public void run() {
+ MessageProducer producer;
+ try {
+ producer = session.createProducer(destination);
+ for (int i=0; i< numMessages; i++) {
+ producer.send(session.createTextMessage(payload));
+ sent.incrementAndGet();
+ TimeUnit.MILLISECONDS.sleep(10);
+ }
+ producer.close();
+ sendDone.countDown();
+ } catch (Exception e) {
+ sendDone.countDown();
+ e.printStackTrace();
+ fail(e.toString());
+ }
+ }
+ });
+
+ assertTrue("messages sending done", sendDone.await(90, TimeUnit.SECONDS));
+ assertEquals("all message were sent", numMessages, sent.get());
+
+ doAck.countDown();
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
+
+ assertEquals("usage goes to 0", 0,
+ TestSupport.getDestination(broker,
+ ActiveMQDestination.transform(destination)).getMemoryUsage().getPercentUsage());
+
+ }
+
+ BrokerService createBroker() throws Exception {
+ BrokerService brokerService = new BrokerService();
+ brokerService.addConnector("tcp://localhost:0");
+ brokerService.setUseJmx(false);
+ brokerService.setDeleteAllMessagesOnStartup(true);
+
+ // spooling to disk early so topic memory limit is not reached
+ brokerService.getSystemUsage().getMemoryUsage().setLimit(500*1024);
+
+ final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
+ final PolicyEntry entry = new PolicyEntry();
+ entry.setTopic(">");
+
+ // so consumer does not get over run while blocked limit the prefetch
+ entry.setTopicPrefetch(50);
+
+ // limit the number of outstanding messages, large enough to use the file store
+ ConstantPendingMessageLimitStrategy pendingMessageLimitStrategy = new ConstantPendingMessageLimitStrategy();
+ pendingMessageLimitStrategy.setLimit(500);
+ entry.setPendingMessageLimitStrategy(pendingMessageLimitStrategy);
+
+ // to keep the limit in check and up to date rather than just the first few, evict some
+ OldestMessageEvictionStrategy messageEvictionStrategy = new OldestMessageEvictionStrategy();
+ messageEvictionStrategy.setEvictExpiredMessagesHighWatermark(100);
+ entry.setMessageEvictionStrategy(messageEvictionStrategy);
+
+ // let evicted messaged disappear
+ entry.setDeadLetterStrategy(null);
+ policyEntries.add(entry);
+
+ final PolicyMap policyMap = new PolicyMap();
+ policyMap.setPolicyEntries(policyEntries);
+ brokerService.setDestinationPolicy(policyMap);
+
+ brokerService.setAdvisorySupport(false);
+
+ return brokerService;
+ }
+
+ ConnectionFactory createConnectionFactory() throws Exception {
+ String url = ((TransportConnector) broker.getTransportConnectors().get(0)).getServer().getConnectURI().toString();
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
+ factory.setWatchTopicAdvisories(false);
+ return factory;
+ }
+
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/MessageEvictionTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/TopicSubscriptionTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/TopicSubscriptionTest.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/TopicSubscriptionTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/TopicSubscriptionTest.java Wed Feb 24 12:53:37 2010
@@ -16,8 +16,6 @@
*/
package org.apache.activemq.broker;
-import javax.jms.JMSException;
-
import org.apache.activemq.TestSupport;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.util.ThreadTracker;
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java?rev=915770&r1=915769&r2=915770&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerOutstandingCommitTest.java Wed Feb 24 12:53:37 2010
@@ -20,6 +20,7 @@
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
@@ -157,7 +158,19 @@
}
});
- produceMessage(producerSession, destination, prefetch * 2);
+ // may block if broker shutodwn happens quickly
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ LOG.info("producer started");
+ try {
+ produceMessage(producerSession, destination, prefetch * 2);
+ } catch (JMSException e) {
+ e.printStackTrace();
+ fail("unexpceted ex on producer: " + e);
+ }
+ LOG.info("producer done");
+ }
+ });
// will be stopped by the plugin
broker.waitUntilStopped();
@@ -245,7 +258,19 @@
}
});
- produceMessage(producerSession, destination, prefetch * 2);
+ // may block if broker shutdown happens quickly
+ Executors.newSingleThreadExecutor().execute(new Runnable() {
+ public void run() {
+ LOG.info("producer started");
+ try {
+ produceMessage(producerSession, destination, prefetch * 2);
+ } catch (JMSException e) {
+ e.printStackTrace();
+ fail("unexpceted ex on producer: " + e);
+ }
+ LOG.info("producer done");
+ }
+ });
// will be stopped by the plugin
broker.waitUntilStopped();