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

This closes #3567

This commit is contained in:
Justin Bertram 2021-05-24 11:49:38 -05:00
parent 0477fc1e77 3c0e14de58
commit a2db9d862b
14 changed files with 198 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/Validators.java
View File

@ -26,6 +26,7 @@ import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancing
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.DeletionPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
/**
* A Validators.
@ -193,6 +194,19 @@ public final class Validators {
}
};
public static final Validator SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT = new Validator() {
@Override
public void validate(final String name, final Object value) {
String val = (String) value;
if (val == null || !val.equals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_SECOND.toString()) &&
!val.equals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_MINUTE.toString()) &&
!val.equals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_HOUR.toString()) &&
!val.equals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_DAY.toString())) {
throw ActiveMQMessageBundle.BUNDLE.invalidSlowConsumerThresholdMeasurementUnit(val);
}
}
};
public static final Validator SLOW_CONSUMER_POLICY_TYPE = new Validator() {
@Override
public void validate(final String name, final Object value) {

8
artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java
View File

@ -94,6 +94,7 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.DeletionPolicy;
import org.apache.activemq.artemis.core.settings.impl.ResourceLimitSettings;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
import org.apache.activemq.artemis.utils.ByteUtil;
import org.apache.activemq.artemis.utils.ClassloadingUtil;
import org.apache.activemq.artemis.utils.DefaultSensitiveStringCodec;
@ -240,6 +241,8 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
private static final String SLOW_CONSUMER_THRESHOLD_NODE_NAME = "slow-consumer-threshold";
private static final String SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT_NODE_NAME = "slow-consumer-threshold-measurement-unit";
private static final String SLOW_CONSUMER_CHECK_PERIOD_NODE_NAME = "slow-consumer-check-period";
private static final String SLOW_CONSUMER_POLICY_NODE_NAME = "slow-consumer-policy";
@ -1247,6 +1250,11 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
Validators.MINUS_ONE_OR_GT_ZERO.validate(SLOW_CONSUMER_THRESHOLD_NODE_NAME, slowConsumerThreshold);
addressSettings.setSlowConsumerThreshold(slowConsumerThreshold);
} else if (SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT_NODE_NAME.equalsIgnoreCase(name)) {
String slowConsumerThresholdMeasurementUnit = getTrimmedTextContent(child);
Validators.SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT.validate(SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT_NODE_NAME, slowConsumerThresholdMeasurementUnit);
addressSettings.setSlowConsumerThresholdMeasurementUnit(SlowConsumerThresholdMeasurementUnit.valueOf(slowConsumerThresholdMeasurementUnit));
} else if (SLOW_CONSUMER_CHECK_PERIOD_NODE_NAME.equalsIgnoreCase(name)) {
long slowConsumerCheckPeriod = XMLUtil.parseLong(child);
Validators.GT_ZERO.validate(SLOW_CONSUMER_CHECK_PERIOD_NODE_NAME, slowConsumerCheckPeriod);

1
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
View File

@ -2934,6 +2934,7 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
.add("sendToDLAOnNoRoute", addressSettings.isSendToDLAOnNoRoute())
.add("addressFullMessagePolicy", policy)
.add("slowConsumerThreshold", addressSettings.getSlowConsumerThreshold())
.add("slowConsumerThresholdMeasurementUnit", addressSettings.getSlowConsumerThresholdMeasurementUnit().toString())
.add("slowConsumerCheckPeriod", addressSettings.getSlowConsumerCheckPeriod())
.add("slowConsumerPolicy", consumerPolicy)
.add("autoCreateJmsQueues", addressSettings.isAutoCreateJmsQueues())

3
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
View File

@ -498,4 +498,7 @@ public interface ActiveMQMessageBundle {
@Message(id = 229233, value = "Cannot set ActiveMQSecurityManager during startup or while started")
IllegalStateException cannotSetSecurityManager();
@Message(id = 229234, value = "Invalid slow consumer threshold measurement unit {0}", format = Message.Format.MESSAGE_FORMAT)
IllegalArgumentException invalidSlowConsumerThresholdMeasurementUnit(String val);
}

20
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -94,6 +94,7 @@ import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
@ -4344,7 +4345,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
} else {
if (slowConsumerReaperRunnable == null) {
scheduleSlowConsumerReaper(settings);
} else if (slowConsumerReaperRunnable.checkPeriod != settings.getSlowConsumerCheckPeriod() || slowConsumerReaperRunnable.threshold != settings.getSlowConsumerThreshold() || !slowConsumerReaperRunnable.policy.equals(settings.getSlowConsumerPolicy())) {
} else if (slowConsumerReaperRunnable.checkPeriod != settings.getSlowConsumerCheckPeriod() || slowConsumerReaperRunnable.thresholdInMsgPerSecond != settings.getSlowConsumerThreshold() || !slowConsumerReaperRunnable.policy.equals(settings.getSlowConsumerPolicy())) {
if (slowConsumerReaperFuture != null) {
slowConsumerReaperFuture.cancel(false);
slowConsumerReaperFuture = null;
@ -4355,12 +4356,12 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
void scheduleSlowConsumerReaper(AddressSettings settings) {
slowConsumerReaperRunnable = new SlowConsumerReaperRunnable(settings.getSlowConsumerCheckPeriod(), settings.getSlowConsumerThreshold(), settings.getSlowConsumerPolicy());
slowConsumerReaperRunnable = new SlowConsumerReaperRunnable(settings.getSlowConsumerCheckPeriod(), settings.getSlowConsumerThreshold(), settings.getSlowConsumerThresholdMeasurementUnit(), settings.getSlowConsumerPolicy());
slowConsumerReaperFuture = scheduledExecutor.scheduleWithFixedDelay(slowConsumerReaperRunnable, settings.getSlowConsumerCheckPeriod(), settings.getSlowConsumerCheckPeriod(), TimeUnit.SECONDS);
if (logger.isDebugEnabled()) {
logger.debug("Scheduled slow-consumer-reaper thread for queue \"" + getName() + "\"; slow-consumer-check-period=" + settings.getSlowConsumerCheckPeriod() + ", slow-consumer-threshold=" + settings.getSlowConsumerThreshold() + ", slow-consumer-policy=" + settings.getSlowConsumerPolicy());
logger.debug("Scheduled slow-consumer-reaper thread for queue \"" + getName() + "\"; slow-consumer-check-period=" + settings.getSlowConsumerCheckPeriod() + ", slow-consumer-threshold=" + settings.getSlowConsumerThreshold() + ", slow-consumer-threshold-measurement-unit=" + settings.getSlowConsumerThresholdMeasurementUnit().toString() + ", slow-consumer-policy=" + settings.getSlowConsumerPolicy());
}
}
@ -4424,13 +4425,13 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
private final class SlowConsumerReaperRunnable implements Runnable {
private final SlowConsumerPolicy policy;
private final float threshold;
private final float thresholdInMsgPerSecond;
private final long checkPeriod;
private SlowConsumerReaperRunnable(long checkPeriod, float threshold, SlowConsumerPolicy policy) {
private SlowConsumerReaperRunnable(long checkPeriod, float slowConsumerThreshold, SlowConsumerThresholdMeasurementUnit unit, SlowConsumerPolicy policy) {
this.checkPeriod = checkPeriod;
this.policy = policy;
this.threshold = threshold;
this.thresholdInMsgPerSecond = slowConsumerThreshold / unit.getValue();
}
@Override
@ -4447,7 +4448,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
logger.debug("There are no consumers, no need to check slow consumer's rate");
return;
} else {
float queueThreshold = threshold * consumers.size();
float queueThreshold = thresholdInMsgPerSecond * consumers.size();
if (queueRate < queueThreshold && queueMessages < queueThreshold) {
if (logger.isDebugEnabled()) {
@ -4462,7 +4463,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
if (consumer instanceof ServerConsumerImpl) {
ServerConsumerImpl serverConsumer = (ServerConsumerImpl) consumer;
float consumerRate = serverConsumer.getRate();
if (consumerRate < threshold) {
if (consumerRate < thresholdInMsgPerSecond) {
RemotingConnection connection = null;
ActiveMQServer server = ((PostOfficeImpl) postOffice).getServer();
RemotingService remotingService = server.getRemotingService();
@ -4476,7 +4477,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
serverConsumer.fireSlowConsumer();
if (connection != null) {
ActiveMQServerLogger.LOGGER.slowConsumerDetected(serverConsumer.getSessionID(), serverConsumer.getID(), getName().toString(), connection.getRemoteAddress(), threshold, consumerRate);
ActiveMQServerLogger.LOGGER.slowConsumerDetected(serverConsumer.getSessionID(), serverConsumer.getID(), getName().toString(), connection.getRemoteAddress(),
thresholdInMsgPerSecond, consumerRate);
if (policy.equals(SlowConsumerPolicy.KILL)) {
connection.killMessage(server.getNodeID());
remotingService.removeConnection(connection.getID());

34
artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
View File

@ -131,6 +131,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
public static final int MANAGEMENT_MESSAGE_ATTRIBUTE_SIZE_LIMIT = 256;
public static final SlowConsumerThresholdMeasurementUnit DEFAULT_SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT = SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_SECOND;
private AddressFullMessagePolicy addressFullMessagePolicy = null;
private Long maxSizeBytes = null;
@ -185,6 +187,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
private Long slowConsumerThreshold = null;
private SlowConsumerThresholdMeasurementUnit slowConsumerThresholdMeasurementUnit = DEFAULT_SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT;
private Long slowConsumerCheckPeriod = null;
private SlowConsumerPolicy slowConsumerPolicy = null;
@ -327,6 +331,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
this.defaultRingSize = other.defaultRingSize;
this.enableMetrics = other.enableMetrics;
this.managementMessageAttributeSizeLimit = other.managementMessageAttributeSizeLimit;
this.slowConsumerThresholdMeasurementUnit = other.slowConsumerThresholdMeasurementUnit;
}
public AddressSettings() {
@ -785,6 +790,15 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
return this;
}
public SlowConsumerThresholdMeasurementUnit getSlowConsumerThresholdMeasurementUnit() {
return slowConsumerThresholdMeasurementUnit != null ? slowConsumerThresholdMeasurementUnit : AddressSettings.DEFAULT_SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT;
}
public AddressSettings setSlowConsumerThresholdMeasurementUnit(final SlowConsumerThresholdMeasurementUnit slowConsumerThresholdMeasurementUnit) {
this.slowConsumerThresholdMeasurementUnit = slowConsumerThresholdMeasurementUnit;
return this;
}
public long getSlowConsumerCheckPeriod() {
return slowConsumerCheckPeriod != null ? slowConsumerCheckPeriod : AddressSettings.DEFAULT_SLOW_CONSUMER_CHECK_PERIOD;
}
@ -1005,6 +1019,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
if (slowConsumerThreshold == null) {
slowConsumerThreshold = merged.slowConsumerThreshold;
}
if (slowConsumerThresholdMeasurementUnit == null) {
slowConsumerThresholdMeasurementUnit = merged.slowConsumerThresholdMeasurementUnit;
}
if (slowConsumerCheckPeriod == null) {
slowConsumerCheckPeriod = merged.slowConsumerCheckPeriod;
}
@ -1354,6 +1371,12 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
managementMessageAttributeSizeLimit = BufferHelper.readNullableInteger(buffer);
}
if (buffer.readableBytes() > 0) {
Integer slowConsumerMeasurementUnitEnumValue = BufferHelper.readNullableInteger(buffer);
if (slowConsumerMeasurementUnitEnumValue != null) {
slowConsumerThresholdMeasurementUnit = SlowConsumerThresholdMeasurementUnit.valueOf(slowConsumerMeasurementUnitEnumValue);
}
}
}
@Override
@ -1418,7 +1441,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
SimpleString.sizeofNullableString(expiryQueueSuffix) +
BufferHelper.sizeOfNullableBoolean(enableMetrics) +
BufferHelper.sizeOfNullableBoolean(defaultGroupRebalancePauseDispatch) +
BufferHelper.sizeOfNullableInteger(managementMessageAttributeSizeLimit);
BufferHelper.sizeOfNullableInteger(managementMessageAttributeSizeLimit) +
BufferHelper.sizeOfNullableInteger(slowConsumerThresholdMeasurementUnit.getValue());
}
@Override
@ -1546,6 +1570,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
BufferHelper.writeNullableBoolean(buffer, defaultGroupRebalancePauseDispatch);
BufferHelper.writeNullableInteger(buffer, managementMessageAttributeSizeLimit);
BufferHelper.writeNullableInteger(buffer, slowConsumerThresholdMeasurementUnit == null ? null : slowConsumerThresholdMeasurementUnit.getValue());
}
/* (non-Javadoc)
@ -1619,6 +1645,7 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
result = prime * result + ((expiryQueueSuffix == null) ? 0 : expiryQueueSuffix.hashCode());
result = prime * result + ((enableMetrics == null) ? 0 : enableMetrics.hashCode());
result = prime * result + ((managementMessageAttributeSizeLimit == null) ? 0 : managementMessageAttributeSizeLimit.hashCode());
result = prime * result + ((slowConsumerThresholdMeasurementUnit == null) ? 0 : slowConsumerThresholdMeasurementUnit.hashCode());
return result;
}
@ -1976,6 +2003,9 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
} else if (!enableMetrics.equals(other.enableMetrics))
return false;
if (slowConsumerThresholdMeasurementUnit != other.slowConsumerThresholdMeasurementUnit)
return false;
return true;
}
@ -2031,6 +2061,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
sendToDLAOnNoRoute +
", slowConsumerThreshold=" +
slowConsumerThreshold +
", slowConsumerThresholdMeasurementUnit=" +
slowConsumerThresholdMeasurementUnit +
", slowConsumerCheckPeriod=" +
slowConsumerCheckPeriod +
", slowConsumerPolicy=" +

46
artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/SlowConsumerThresholdMeasurementUnit.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.artemis.core.settings.impl;
public enum SlowConsumerThresholdMeasurementUnit {
MESSAGES_PER_SECOND(1), MESSAGES_PER_MINUTE(60), MESSAGES_PER_HOUR(3600), MESSAGES_PER_DAY(3600 * 24);
private final int measurementUnitInSeconds;
SlowConsumerThresholdMeasurementUnit(int measurementUnitInSeconds) {
this.measurementUnitInSeconds = measurementUnitInSeconds;
}
public static SlowConsumerThresholdMeasurementUnit valueOf(int measurementUnitInSeconds) {
switch (measurementUnitInSeconds) {
case 1:
return MESSAGES_PER_SECOND;
case 60:
return MESSAGES_PER_MINUTE;
case 3600:
return MESSAGES_PER_HOUR;
case 3600 * 24:
return MESSAGES_PER_DAY;
default:
return null;
}
}
public int getValue() {
return measurementUnitInSeconds;
}
}

21
artemis-server/src/main/resources/schema/artemis-configuration.xsd
View File

@ -3685,12 +3685,29 @@
<xsd:element name="slow-consumer-threshold" type="xsd:long" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
The minimum rate of message consumption allowed before a consumer is considered "slow." Measured
in messages-per-second.
The minimum rate of message consumption allowed before a consumer is considered "slow." Measurement
unit is defined by the slow-consumer-threshold-measurement-unit parameter. By default this is
messages-per-seconds
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="slow-consumer-threshold-measurement-unit" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>
The units used to measure the slow consumer threshold. Default is messages-per-second.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="MESSAGES_PER_SECOND"/>
<xsd:enumeration value="MESSAGES_PER_MINUTE"/>
<xsd:enumeration value="MESSAGES_PER_HOUR"/>
<xsd:enumeration value="MESSAGES_PER_DAY"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:element>
<xsd:element name="slow-consumer-policy" default="NOTIFY" maxOccurs="1" minOccurs="0">
<xsd:annotation>
<xsd:documentation>

3
artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationTest.java
View File

@ -63,6 +63,7 @@ import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerBasePlugin;
import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerPlugin;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzerPolicy;
import org.junit.AfterClass;
@ -367,6 +368,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(10, conf.getAddressesSettings().get("a1").getPageCacheMaxSize());
assertEquals(4, conf.getAddressesSettings().get("a1").getMessageCounterHistoryDayLimit());
assertEquals(10, conf.getAddressesSettings().get("a1").getSlowConsumerThreshold());
assertEquals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_HOUR, conf.getAddressesSettings().get("a1").getSlowConsumerThresholdMeasurementUnit());
assertEquals(5, conf.getAddressesSettings().get("a1").getSlowConsumerCheckPeriod());
assertEquals(SlowConsumerPolicy.NOTIFY, conf.getAddressesSettings().get("a1").getSlowConsumerPolicy());
assertEquals(true, conf.getAddressesSettings().get("a1").isAutoCreateJmsQueues());
@ -401,6 +403,7 @@ public class FileConfigurationTest extends ConfigurationImplTest {
assertEquals(20, conf.getAddressesSettings().get("a2").getPageCacheMaxSize());
assertEquals(8, conf.getAddressesSettings().get("a2").getMessageCounterHistoryDayLimit());
assertEquals(20, conf.getAddressesSettings().get("a2").getSlowConsumerThreshold());
assertEquals(SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_DAY, conf.getAddressesSettings().get("a2").getSlowConsumerThresholdMeasurementUnit());
assertEquals(15, conf.getAddressesSettings().get("a2").getSlowConsumerCheckPeriod());
assertEquals(SlowConsumerPolicy.KILL, conf.getAddressesSettings().get("a2").getSlowConsumerPolicy());
assertEquals(false, conf.getAddressesSettings().get("a2").isAutoCreateJmsQueues());

1
artemis-server/src/test/java/org/apache/activemq/artemis/core/settings/AddressSettingsTest.java
View File

@ -38,6 +38,7 @@ public class AddressSettingsTest extends ActiveMQTestBase {
Assert.assertEquals(AddressSettings.DEFAULT_REDELIVER_DELAY, addressSettings.getRedeliveryDelay());
Assert.assertEquals(AddressSettings.DEFAULT_REDELIVER_MULTIPLIER, addressSettings.getRedeliveryMultiplier(), 0.000001);
Assert.assertEquals(AddressSettings.DEFAULT_SLOW_CONSUMER_THRESHOLD, addressSettings.getSlowConsumerThreshold());
Assert.assertEquals(AddressSettings.DEFAULT_SLOW_CONSUMER_THRESHOLD_MEASUREMENT_UNIT, addressSettings.getSlowConsumerThresholdMeasurementUnit());
Assert.assertEquals(AddressSettings.DEFAULT_SLOW_CONSUMER_CHECK_PERIOD, addressSettings.getSlowConsumerCheckPeriod());
Assert.assertEquals(AddressSettings.DEFAULT_SLOW_CONSUMER_POLICY, addressSettings.getSlowConsumerPolicy());
Assert.assertEquals(AddressSettings.DEFAULT_AUTO_CREATE_JMS_QUEUES, addressSettings.isAutoCreateJmsQueues());

2
artemis-server/src/test/resources/ConfigurationTest-full-config.xml
View File

@ -458,6 +458,7 @@
<slow-consumer-threshold>10</slow-consumer-threshold>
<slow-consumer-check-period>5</slow-consumer-check-period>
<slow-consumer-policy>NOTIFY</slow-consumer-policy>
<slow-consumer-threshold-measurement-unit>MESSAGES_PER_HOUR</slow-consumer-threshold-measurement-unit>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-delete-jms-queues>true</auto-delete-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
@ -488,6 +489,7 @@
<message-counter-history-day-limit>8</message-counter-history-day-limit>
<slow-consumer-threshold>20</slow-consumer-threshold>
<slow-consumer-check-period>15</slow-consumer-check-period>
<slow-consumer-threshold-measurement-unit>MESSAGES_PER_DAY</slow-consumer-threshold-measurement-unit>
<slow-consumer-policy>KILL</slow-consumer-policy>
<auto-create-jms-queues>false</auto-create-jms-queues>
<auto-delete-jms-queues>false</auto-delete-jms-queues>

2
artemis-server/src/test/resources/ConfigurationTest-xinclude-config-address-settings.xml
View File

@ -30,6 +30,7 @@
<slow-consumer-threshold>10</slow-consumer-threshold>
<slow-consumer-check-period>5</slow-consumer-check-period>
<slow-consumer-policy>NOTIFY</slow-consumer-policy>
<slow-consumer-threshold-measurement-unit>MESSAGES_PER_HOUR</slow-consumer-threshold-measurement-unit>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-delete-jms-queues>true</auto-delete-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
@ -61,6 +62,7 @@
<slow-consumer-threshold>20</slow-consumer-threshold>
<slow-consumer-check-period>15</slow-consumer-check-period>
<slow-consumer-policy>KILL</slow-consumer-policy>
<slow-consumer-threshold-measurement-unit>MESSAGES_PER_DAY</slow-consumer-threshold-measurement-unit>
<auto-create-jms-queues>false</auto-create-jms-queues>
<auto-delete-jms-queues>false</auto-delete-jms-queues>
<auto-create-jms-topics>false</auto-create-jms-topics>

24
docs/user-manual/en/address-model.md
View File

@ -660,6 +660,7 @@ that would be found in the `broker.xml` file.
<redistribution-delay>0</redistribution-delay>
<send-to-dla-on-no-route>true</send-to-dla-on-no-route>
<slow-consumer-threshold>-1</slow-consumer-threshold>
<slow-consumer-threshold-measurement-unit>MESSAGES_PER_SECOND</slow-consumer-threshold-measurement-unit>
<slow-consumer-policy>NOTIFY</slow-consumer-policy>
<slow-consumer-check-period>5</slow-consumer-check-period>
<auto-create-jms-queues>true</auto-create-jms-queues> <!-- deprecated! see auto-create-queues -->
@ -823,8 +824,19 @@ message will instead be sent to the `dead-letter-address` (DLA) for that
address, if it exists.
`slow-consumer-threshold`. The minimum rate of message consumption allowed
before a consumer is considered "slow." Measured in messages-per-second.
Default is `-1` (i.e. disabled); any other valid value must be greater than 0.
before a consumer is considered "slow." Measured in units specified by the
slow-consumer-threshold-measurement-unit configuration option. Default is `-1`
(i.e. disabled); any other valid value must be greater than 0.
Read more about [slow consumers](slow-consumers.md).
`slow-consumer-threshold-measurement-unit`. The units used to measure the
slow-consumer-threshold. Valid options are:
* MESSAGES_PER_SECOND
* MESSAGES_PER_MINUTE
* MESSAGES_PER_HOUR
* MESSAGES_PER_DAY
If no unit is specified the default MESSAGES_PER_SECOND will be used.
Read more about [slow consumers](slow-consumers.md).
`slow-consumer-policy`. What should happen when a slow consumer is detected.
@ -834,7 +846,13 @@ CONSUMER\_SLOW management notification which an application could receive and
take action with. Read more about [slow consumers](slow-consumers.md).
`slow-consumer-check-period`. How often to check for slow consumers on a
particular queue. Measured in *seconds*. Default is `5`. Read more about [slow
particular queue. Measured in *seconds*. Default is `5`.
* Note: This should be at least 2x the maximum time it takes a consumer to process
1 message. For example, if the slow-consumer-threshold is set to 1 and the
slow-consumer-threshold-measurement-unit is set to MESSAGES_PER_MINUTE then this
should be set to at least 2 x 60s i.e. 120s.
Read more about [slow
consumers](slow-consumers.md).
`auto-create-jms-queues` is **deprecated**. See `auto-create-queues`. Whether

47
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SlowConsumerTest.java
View File

@ -44,6 +44,7 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.TimeUtils;
@ -55,24 +56,42 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import static org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_MINUTE;
import static org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit.MESSAGES_PER_SECOND;
@RunWith(value = Parameterized.class)
public class SlowConsumerTest extends ActiveMQTestBase {
private static final Logger logger = Logger.getLogger(SlowConsumerTest.class);
private int threshold = 10;
private int threshold;
private float consumerRate; // The rate actual consumers will run in the test.
private SlowConsumerThresholdMeasurementUnit unit;
private long checkPeriod = 1;
private boolean isNetty = false;
private boolean isPaging = false;
// this will ensure that all tests in this class are run twice,
// once with "true" passed to the class' constructor and once with "false"
@Parameterized.Parameters(name = "netty={0}, paging={1}")
@Parameterized.Parameters(name = "netty={0}, paging={1}, threshold={2}, threshold_units={3}")
public static Collection getParameters() {
return Arrays.asList(new Object[][]{{true, false}, {false, false}, {true, true}, {false, true}});
return Arrays.asList(new Object[][]{
{true, false, 10, MESSAGES_PER_SECOND},
{false, false, 10, MESSAGES_PER_SECOND},
{true, true, 10, MESSAGES_PER_SECOND},
{false, true, 10, MESSAGES_PER_SECOND},
{true, false, 1, MESSAGES_PER_MINUTE},
{false, false, 1, MESSAGES_PER_MINUTE},
{true, true, 1, MESSAGES_PER_MINUTE},
{false, true, 1, MESSAGES_PER_MINUTE}
});
}
public SlowConsumerTest(boolean isNetty, boolean isPaging) {
public SlowConsumerTest(boolean isNetty, boolean isPaging, int threshold, SlowConsumerThresholdMeasurementUnit unit) {
this.threshold = threshold;
this.unit = unit;
this.consumerRate = threshold / unit.getValue();
this.isNetty = isNetty;
this.isPaging = isPaging;
}
@ -93,6 +112,7 @@ public class SlowConsumerTest extends ActiveMQTestBase {
AddressSettings addressSettings = new AddressSettings();
addressSettings.setSlowConsumerCheckPeriod(checkPeriod);
addressSettings.setSlowConsumerThreshold(threshold);
addressSettings.setSlowConsumerThresholdMeasurementUnit(unit);
addressSettings.setSlowConsumerPolicy(SlowConsumerPolicy.KILL);
if (isPaging) {
@ -442,14 +462,14 @@ public class SlowConsumerTest extends ActiveMQTestBase {
final int messages = 10 * threshold;
FixedRateProducer producer = new FixedRateProducer(threshold * 2, sf1, QUEUE, messages);
FixedRateProducer producer = new FixedRateProducer(threshold * 2, unit, sf1, QUEUE, messages);
final Set<FixedRateConsumer> consumers = new ConcurrentHashSet<>();
final Set<ClientMessage> receivedMessages = new ConcurrentHashSet<>();
consumers.add(new FixedRateConsumer(threshold, receivedMessages, sf2, QUEUE, 1));
consumers.add(new FixedRateConsumer(threshold, receivedMessages, sf3, QUEUE, 2));
consumers.add(new FixedRateConsumer(threshold, receivedMessages, sf4, QUEUE, 3));
consumers.add(new FixedRateConsumer(threshold, unit, receivedMessages, sf2, QUEUE, 1));
consumers.add(new FixedRateConsumer(threshold, unit, receivedMessages, sf3, QUEUE, 2));
consumers.add(new FixedRateConsumer(threshold, unit, receivedMessages, sf4, QUEUE, 3));
try {
producer.start();
@ -481,8 +501,8 @@ public class SlowConsumerTest extends ActiveMQTestBase {
int messages;
ClientProducer producer;
FixedRateProducer(int rate, ClientSessionFactory sf, SimpleString queue, int messages) throws ActiveMQException {
super(sf, queue, rate);
FixedRateProducer(int rate, SlowConsumerThresholdMeasurementUnit unit, ClientSessionFactory sf, SimpleString queue, int messages) throws ActiveMQException {
super(sf, queue, rate, unit);
this.messages = messages;
}
@ -518,11 +538,12 @@ public class SlowConsumerTest extends ActiveMQTestBase {
int id;
FixedRateConsumer(int rate,
SlowConsumerThresholdMeasurementUnit unit,
Set<ClientMessage> receivedMessages,
ClientSessionFactory sf,
SimpleString queue,
int id) throws ActiveMQException {
super(sf, queue, rate);
super(sf, queue, rate, unit);
this.id = id;
this.receivedMessages = receivedMessages;
}
@ -568,10 +589,10 @@ public class SlowConsumerTest extends ActiveMQTestBase {
protected volatile boolean working;
boolean failed;
FixedRateClient(ClientSessionFactory sf, SimpleString queue, int rate) throws ActiveMQException {
FixedRateClient(ClientSessionFactory sf, SimpleString queue, int rate, SlowConsumerThresholdMeasurementUnit unit) throws ActiveMQException {
this.sf = sf;
this.queue = queue;
this.sleepTime = 1000 / rate;
this.sleepTime = (int) (1000f / (unit.getValue() / rate));
}
protected void prepareWork() throws ActiveMQException {