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 2011/09/21 14:09:30 UTC
svn commit: r1173583 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
Author: gtully
Date: Wed Sep 21 12:09:30 2011
New Revision: 1173583
URL: http://svn.apache.org/viewvc?rev=1173583&view=rev
Log:
https://issues.apache.org/jira/browse/AMQ-3507 - Large number of expiring messages causing QueueSize to groww. have cursor delegate to the destination for expiry processing. additional test. This will also sort out the intermittent failure of this test.
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java?rev=1173583&r1=1173582&r2=1173583&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java Wed Sep 21 12:09:30 2011
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Destination;
+import org.apache.activemq.broker.region.IndirectMessageReference;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.QueueMessageReference;
import org.apache.activemq.command.Message;
@@ -233,7 +234,7 @@ public class FilePendingMessageCursor ex
throw new RuntimeException(e);
}
} else {
- discard(node);
+ discardExpiredMessage(node);
}
//message expired
return true;
@@ -279,7 +280,7 @@ public class FilePendingMessageCursor ex
throw new RuntimeException(e);
}
} else {
- discard(node);
+ discardExpiredMessage(node);
}
}
@@ -410,7 +411,8 @@ public class FilePendingMessageCursor ex
while (!tmpList.isEmpty()) {
MessageReference node = tmpList.removeFirst();
if (node.isExpired()) {
- discard(node);
+ node.decrementReferenceCount();
+ discardExpiredMessage(node);
} else {
memoryList.add(node);
}
@@ -463,14 +465,15 @@ public class FilePendingMessageCursor ex
return diskList;
}
- protected void discard(MessageReference message) {
- message.decrementReferenceCount();
+ private void discardExpiredMessage(MessageReference reference) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Discarding message " + message);
+ LOG.debug("Discarding expired message " + reference);
+ }
+ if (broker.isExpired(reference)) {
+ ConnectionContext context = new ConnectionContext(new NonCachedMessageEvaluationContext());
+ context.setBroker(broker);
+ reference.getRegionDestination().messageExpired(context, null, new IndirectMessageReference(reference.getMessage()));
}
- ConnectionContext ctx = new ConnectionContext(new NonCachedMessageEvaluationContext());
- ctx.setBroker(broker);
- broker.getRoot().sendToDeadLetterQueue(ctx, message, null);
}
protected ByteSequence getByteSequence(Message message) throws IOException {
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java?rev=1173583&r1=1173582&r2=1173583&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ExpiredMessagesWithNoConsumerTest.java Wed Sep 21 12:09:30 2011
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.jms.Connection;
+import javax.jms.DeliveryMode;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -72,14 +73,18 @@ public class ExpiredMessagesWithNoConsum
}
protected void createBrokerWithMemoryLimit() throws Exception {
- doCreateBroker(true);
+ createBrokerWithMemoryLimit(800);
+ }
+
+ protected void createBrokerWithMemoryLimit(int expireMessagesPeriod) throws Exception {
+ doCreateBroker(true, expireMessagesPeriod);
}
protected void createBroker() throws Exception {
- doCreateBroker(false);
+ doCreateBroker(false, 800);
}
- private void doCreateBroker(boolean memoryLimit) throws Exception {
+ private void doCreateBroker(boolean memoryLimit, int expireMessagesPeriod) throws Exception {
broker = new BrokerService();
broker.setBrokerName("localhost");
broker.setUseJmx(true);
@@ -89,7 +94,7 @@ public class ExpiredMessagesWithNoConsum
PolicyMap policyMap = new PolicyMap();
PolicyEntry defaultEntry = new PolicyEntry();
defaultEntry.setOptimizedDispatch(optimizedDispatch);
- defaultEntry.setExpireMessagesPeriod(800);
+ defaultEntry.setExpireMessagesPeriod(expireMessagesPeriod);
defaultEntry.setMaxExpirePageSize(800);
defaultEntry.setPendingQueuePolicy(pendingQueuePolicy);
@@ -109,6 +114,78 @@ public class ExpiredMessagesWithNoConsum
connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
}
+ public void testExpiredNonPersistentMessagesWithNoConsumer() throws Exception {
+
+ createBrokerWithMemoryLimit(2000);
+
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connectionUri);
+ connection = factory.createConnection();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ producer = session.createProducer(destination);
+ producer.setTimeToLive(1000);
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ connection.start();
+ final long sendCount = 2000;
+
+ final Thread producingThread = new Thread("Producing Thread") {
+ public void run() {
+ try {
+ int i = 0;
+ long tStamp = System.currentTimeMillis();
+ while (i++ < sendCount) {
+ producer.send(session.createTextMessage("test"));
+ if (i%100 == 0) {
+ LOG.info("sent: " + i + " @ " + ((System.currentTimeMillis() - tStamp) / 100) + "m/ms");
+ tStamp = System.currentTimeMillis() ;
+ }
+
+ if (135 == i) {
+ // allow pending messages to expire, before usage limit kicks in to flush them
+ TimeUnit.SECONDS.sleep(5);
+ }
+ }
+ } catch (Throwable ex) {
+ ex.printStackTrace();
+ }
+ }
+ };
+
+ producingThread.start();
+
+ assertTrue("producer failed to complete within allocated time", Wait.waitFor(new Wait.Condition() {
+ public boolean isSatisified() throws Exception {
+ producingThread.join(TimeUnit.SECONDS.toMillis(3000));
+ return !producingThread.isAlive();
+ }
+ }));
+
+ TimeUnit.SECONDS.sleep(5);
+ final DestinationViewMBean view = createView(destination);
+ Wait.waitFor(new Wait.Condition() {
+ public boolean isSatisified() throws Exception {
+ try {
+ LOG.info("enqueue=" + view.getEnqueueCount() + ", dequeue=" + view.getDequeueCount()
+ + ", inflight=" + view.getInFlightCount() + ", expired= " + view.getExpiredCount()
+ + ", size= " + view.getQueueSize());
+ return view.getDequeueCount() != 0
+ && view.getDequeueCount() == view.getExpiredCount()
+ && view.getDequeueCount() == view.getEnqueueCount()
+ && view.getQueueSize() == 0;
+ } catch (Exception ignored) {
+ LOG.info(ignored.toString());
+ }
+ return false;
+ }
+ }, Wait.MAX_WAIT_MILLIS * 10);
+ LOG.info("enqueue=" + view.getEnqueueCount() + ", dequeue=" + view.getDequeueCount()
+ + ", inflight=" + view.getInFlightCount() + ", expired= " + view.getExpiredCount()
+ + ", size= " + view.getQueueSize());
+
+ assertEquals("memory usage doesn't go to duck egg", 0, view.getMemoryPercentUsage());
+ assertEquals("0 queue", 0, view.getQueueSize());
+ }
+
+
public void initCombosForTestExpiredMessagesWithNoConsumer() {
addCombinationValues("optimizedDispatch", new Object[] {Boolean.TRUE, Boolean.FALSE});
addCombinationValues("pendingQueuePolicy", new Object[] {null, new VMPendingQueueMessageStoragePolicy(), new FilePendingQueueMessageStoragePolicy()});