Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Changing memory limit flow control warning messages to be one time messages [http://issues.apache.org/activemq/browse/AMQ-2401] 

This avoids performance penalty of constant logging since under normal circumstance producers will be going in and out of flow control frequently. 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@817842 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Colin W Macnaughton 2009-09-22 21:12:49 +00:00
parent db3a021384
commit 4245c355dd
3 changed files with 21 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
View File

@ -51,6 +51,7 @@ public abstract class BaseDestination implements Destination {
protected SystemUsage systemUsage; protected SystemUsage systemUsage;
protected MemoryUsage memoryUsage; protected MemoryUsage memoryUsage;
private boolean producerFlowControl = false; private boolean producerFlowControl = false;
protected boolean warnOnProducerFlowControl = true;
private int maxProducersToAudit = 1024; private int maxProducersToAudit = 1024;
private int maxAuditDepth = 2048; private int maxAuditDepth = 2048;
private boolean enableAudit = true; private boolean enableAudit = true;

12
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -412,11 +412,15 @@ public class Queue extends BaseDestination implements Task, UsageListener {
isFull(context, memoryUsage); isFull(context, memoryUsage);
fastProducer(context, producerInfo); fastProducer(context, producerInfo);
if (isProducerFlowControl() && context.isProducerFlowControl()) { if (isProducerFlowControl() && context.isProducerFlowControl()) {
final String logMessage = "Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." + if(warnOnProducerFlowControl) {
" See http://activemq.apache.org/producer-flow-control.html for more info"; warnOnProducerFlowControl = false;
LOG.info(logMessage); LOG.info("Usage Manager memory limit reached on " +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()) { if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException("SystemUsage memory limit reached"); throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " +getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
} }
// We can avoid blocking due to low usage if the producer is sending // We can avoid blocking due to low usage if the producer is sending

18
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -280,14 +280,20 @@ public class Topic extends BaseDestination implements Task{
if(memoryUsage.isFull()) { if(memoryUsage.isFull()) {
isFull(context, memoryUsage); isFull(context, memoryUsage);
fastProducer(context, producerInfo); fastProducer(context, producerInfo);
if (isProducerFlowControl() && context.isProducerFlowControl()) { if (isProducerFlowControl() && context.isProducerFlowControl()) {
final String logMessage = "Usage Manager memory limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " +getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info"; if(warnOnProducerFlowControl) {
LOG.info(logMessage); warnOnProducerFlowControl = false;
if (systemUsage.isSendFailIfNoSpace()) { LOG.info("Usage Manager memory limit reached for " +getActiveMQDestination().getQualifiedName() + ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it." +
throw new javax.jms.ResourceAllocationException(logMessage); " 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 " +getActiveMQDestination().getQualifiedName() + "." +
" See http://activemq.apache.org/producer-flow-control.html for more info");
}
// We can avoid blocking due to low usage if the producer is sending // We can avoid blocking due to low usage if the producer is sending
// a sync message or // a sync message or
// if it is using a producer window // if it is using a producer window