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

ARTEMIS-3720 Improvements on Paging MaxMessages usage 

I "used" the broker a little bit around max-messages and found a few minor issues.
This commit is contained in:
Clebert Suconic 2022-03-21 10:08:58 -04:00 committed by clebertsuconic
parent 7bdb3dc176
commit 5f22a51926
11 changed files with 556 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
View File

@ -133,8 +133,14 @@ ${cluster-security.settings}${cluster.settings}${replicated.settings}${shared-st
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<!-- if max-size-bytes and max-size-messages were both enabled, the system will enter into paging
based on the first attribute to hits the maximum value -->
<!-- limit for the address in bytes, -1 means unlimited -->
<max-size-bytes>-1</max-size-bytes>
<!-- limit for the address in messages, -1 means unlimited -->
<max-size-messages>-1</max-size-messages>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>${full-policy}</address-full-policy>
<auto-create-queues>${auto-create}</auto-create-queues>

11
artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/global-max-default.txt
View File

@ -1,9 +1,6 @@
<!-- the system will enter into page mode once you hit this limit.
This is an estimate in bytes of how much the messages are using in memory
<!-- the system will enter into page mode once you hit this limit. This is an estimate in bytes of how much the messages are using in memory
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
<global-max-size>100Mb</global-max-size>
-->
<global-max-size>100Mb</global-max-size> -->

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

@ -452,10 +452,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
long globalMaxMessages = getLong(e, GLOBAL_MAX_MESSAGES, -1, Validators.MINUS_ONE_OR_GT_ZERO);
if (globalMaxSize > 0) {
config.setGlobalMaxMessages(globalMaxMessages);
}
config.setGlobalMaxMessages(globalMaxMessages);
config.setMaxDiskUsage(getInteger(e, MAX_DISK_USAGE, config.getMaxDiskUsage(), Validators.PERCENTAGE_OR_MINUS_ONE));

8
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingManager.java
View File

@ -131,6 +131,10 @@ public interface PagingManager extends ActiveMQComponent, HierarchicalRepository
return 0;
}
default long getGlobalMessages() {
return 0;
}
/**
* Use this when you have no refernce of an address. (anonymous AMQP Producers for example)
* @param runWhenAvailable
@ -149,4 +153,8 @@ public interface PagingManager extends ActiveMQComponent, HierarchicalRepository
default long getMaxSize() {
return 0;
}
default long getMaxMessages() {
return 0;
}
}

18
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
View File

@ -80,6 +80,8 @@ public final class PagingManagerImpl implements PagingManager {
private long maxSize;
private long maxMessages;
private volatile boolean cleanupEnabled = true;
private volatile boolean diskFull = false;
@ -117,6 +119,7 @@ public final class PagingManagerImpl implements PagingManager {
this.addressSettingsRepository = addressSettingsRepository;
addressSettingsRepository.registerListener(this);
this.maxSize = maxSize;
this.maxMessages = maxMessages;
this.globalSizeMetric = new SizeAwareMetric(maxSize, maxSize, maxMessages, maxMessages);
globalSizeMetric.setSizeEnabled(maxSize >= 0);
globalSizeMetric.setElementsEnabled(maxMessages >= 0);
@ -132,9 +135,10 @@ public final class PagingManagerImpl implements PagingManager {
/** To be used in tests only called through PagingManagerTestAccessor */
void resetMaxSize(long maxSize, long maxElements) {
void resetMaxSize(long maxSize, long maxMessages) {
this.maxSize = maxSize;
this.globalSizeMetric.setMax(maxSize, maxSize, maxElements, maxElements);
this.maxMessages = maxMessages;
this.globalSizeMetric.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
@Override
@ -142,6 +146,11 @@ public final class PagingManagerImpl implements PagingManager {
return maxSize;
}
@Override
public long getMaxMessages() {
return maxMessages;
}
public PagingManagerImpl(final PagingStoreFactory pagingSPI,
final HierarchicalRepository<AddressSettings> addressSettingsRepository) {
this(pagingSPI, addressSettingsRepository, -1, -1, null);
@ -186,6 +195,11 @@ public final class PagingManagerImpl implements PagingManager {
return globalSizeMetric.getSize();
}
@Override
public long getGlobalMessages() {
return globalSizeMetric.getElements();
}
protected void checkMemoryRelease() {
if (!diskFull && (maxSize < 0 || !globalFull) && !blockedStored.isEmpty()) {
if (!memoryCallback.isEmpty()) {

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

@ -530,7 +530,7 @@ public class PagingStoreImpl implements PagingStore {
final boolean isPaging = this.paging;
if (isPaging) {
paging = false;
ActiveMQServerLogger.LOGGER.pageStoreStop(storeName, size.getSize(), maxSize, pagingManager.getGlobalSize());
ActiveMQServerLogger.LOGGER.pageStoreStop(storeName, getPageInfo());
}
this.cursorProvider.onPageModeCleared();
} finally {
@ -538,6 +538,10 @@ public class PagingStoreImpl implements PagingStore {
}
}
private String getPageInfo() {
return String.format("size=%d bytes (%d messages); maxSize=%d bytes (%d messages); globalSize=%d bytes (%d messages); globalMaxSize=%d bytes (%d messages);", size.getSize(), size.getElements(), maxSize, maxMessages, pagingManager.getGlobalSize(), pagingManager.getGlobalMessages(), pagingManager.getMaxSize(), pagingManager.getMaxMessages());
}
@Override
public boolean startPaging() {
if (!running) {
@ -575,7 +579,7 @@ public class PagingStoreImpl implements PagingStore {
}
}
paging = true;
ActiveMQServerLogger.LOGGER.pageStoreStart(storeName, size.getSize(), maxSize, pagingManager.getGlobalSize());
ActiveMQServerLogger.LOGGER.pageStoreStart(storeName, getPageInfo());
return true;
} finally {
@ -764,7 +768,7 @@ public class PagingStoreImpl implements PagingStore {
if (pagingManager.isDiskFull()) {
ActiveMQServerLogger.LOGGER.blockingDiskFull(address);
} else {
ActiveMQServerLogger.LOGGER.blockingMessageProduction(address, size.getSize(), maxSize, pagingManager.getGlobalSize());
ActiveMQServerLogger.LOGGER.blockingMessageProduction(address, getPageInfo());
}
blocking = true;
}
@ -813,7 +817,7 @@ public class PagingStoreImpl implements PagingStore {
if (!onMemoryFreedRunnables.isEmpty()) {
executor.execute(this::memoryReleased);
if (blocking) {
ActiveMQServerLogger.LOGGER.unblockingMessageProduction(address, size.getSize(), maxSize);
ActiveMQServerLogger.LOGGER.unblockingMessageProduction(address, getPageInfo());
blocking = false;
return true;
}
@ -848,7 +852,7 @@ public class PagingStoreImpl implements PagingStore {
// Address is full, we just pretend we are paging, and drop the data
if (!printedDropMessagesWarning) {
printedDropMessagesWarning = true;
ActiveMQServerLogger.LOGGER.pageStoreDropMessages(storeName, size.getSize(), maxSize, pagingManager.getGlobalSize());
ActiveMQServerLogger.LOGGER.pageStoreDropMessages(storeName, getPageInfo());
}
return true;
} else {

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

@ -288,8 +288,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
void switchingNIO();
@LogMessage(level = Logger.Level.INFO)
@Message(id = 221046, value = "Unblocking message production on address ''{0}''; size is currently: {1} bytes; max-size-bytes: {2}", format = Message.Format.MESSAGE_FORMAT)
void unblockingMessageProduction(SimpleString addressName, long currentSize, long maxSize);
@Message(id = 221046, value = "Unblocking message production on address ''{0}''; {1}", format = Message.Format.MESSAGE_FORMAT)
void unblockingMessageProduction(SimpleString addressName, String sizeInfo);
@LogMessage(level = Logger.Level.INFO)
@Message(id = 221047, value = "Backup Server has scaled down to live server", format = Message.Format.MESSAGE_FORMAT)
@ -615,12 +615,12 @@ public interface ActiveMQServerLogger extends BasicLogger {
void pageStoreStartIOError(@Cause Exception e);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222038, value = "Starting paging on address ''{0}''; size is currently: {1} bytes; max-size-bytes: {2}; global-size-bytes: {3}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreStart(SimpleString storeName, long addressSize, long maxSize, long globalMaxSize);
@Message(id = 222038, value = "Starting paging on address ''{0}''; {1}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreStart(SimpleString storeName, String sizeInfo);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222039, value = "Messages sent to address ''{0}'' are being dropped; size is currently: {1} bytes; max-size-bytes: {2}; global-size-bytes: {3}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreDropMessages(SimpleString storeName, long addressSize, long maxSize, long globalMaxSize);
@Message(id = 222039, value = "Messages sent to address ''{0}'' are being dropped; {1}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreDropMessages(SimpleString storeName, String sizeInfo);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222040, value = "Server is stopped", format = Message.Format.MESSAGE_FORMAT)
@ -1218,8 +1218,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
void missingClusterConfigForScaleDown(String scaleDownCluster);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222183, value = "Blocking message production on address ''{0}''; size is currently: {1} bytes; max-size-bytes on address: {2}, global-max-size is {3}", format = Message.Format.MESSAGE_FORMAT)
void blockingMessageProduction(SimpleString addressName, long currentSize, long maxSize, long globalMaxSize);
@Message(id = 222183, value = "Blocking message production on address ''{0}''; {1}", format = Message.Format.MESSAGE_FORMAT)
void blockingMessageProduction(SimpleString addressName, String pageInfo);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222184,
@ -2181,8 +2181,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
void enableTraceForCriticalAnalyzer();
@LogMessage(level = Logger.Level.WARN)
@Message(id = 224108, value = "Stopped paging on address ''{0}''; size is currently: {1} bytes; max-size-bytes: {2}; global-size-bytes: {3}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreStop(SimpleString storeName, long addressSize, long maxSize, long globalMaxSize);
@Message(id = 224108, value = "Stopped paging on address ''{0}''; {1}", format = Message.Format.MESSAGE_FORMAT)
void pageStoreStop(SimpleString storeName, String pageInfo);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 224109, value = "ConnectionRouter {0} not found", format = Message.Format.MESSAGE_FORMAT)

41
tests/smoke-tests/pom.xml
View File

@ -861,7 +861,6 @@
</args>
</configuration>
</execution>
<!-- END JmxReplicatedMultipleFailbackTest -->
<execution>
<phase>test-compile</phase>
<id>create-paging</id>
@ -882,6 +881,46 @@
</args>
</configuration>
</execution>
<execution>
<phase>test-compile</phase>
<id>create-paging-address-messages</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<!-- this makes it easier in certain envs -->
<configuration>${basedir}/target/classes/servers/pagingAddressMaxMessages</configuration>
<allowAnonymous>true</allowAnonymous>
<user>admin</user>
<password>admin</password>
<instance>${basedir}/target/pagingAddressMaxMessages</instance>
<args>
<!-- this is needed to run the server remotely -->
<arg>--java-options</arg>
<arg>-Djava.rmi.server.hostname=localhost</arg>
</args>
</configuration>
</execution>
<execution>
<phase>test-compile</phase>
<id>create-paging-global-messages</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<!-- this makes it easier in certain envs -->
<configuration>${basedir}/target/classes/servers/pagingGlobalMaxMessages</configuration>
<allowAnonymous>true</allowAnonymous>
<user>admin</user>
<password>admin</password>
<instance>${basedir}/target/pagingGlobalMaxMessages</instance>
<args>
<!-- this is needed to run the server remotely -->
<arg>--java-options</arg>
<arg>-Djava.rmi.server.hostname=localhost</arg>
</args>
</configuration>
</execution>
<!-- used on TransferTest -->
<execution>
<phase>test-compile</phase>

183
tests/smoke-tests/src/main/resources/servers/pagingAddressMaxMessages/broker.xml Normal file
View File

@ -0,0 +1,183 @@
<?xml version='1.0'?>
<!--
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.
-->
<configuration xmlns="urn:activemq"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq:core ">
<name>0.0.0.0</name>
<persistence-enabled>true</persistence-enabled>
<!-- this could be ASYNCIO or NIO
-->
<journal-type>NIO</journal-type>
<paging-directory>./data/paging</paging-directory>
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/large-messages</large-messages-directory>
<journal-datasync>true</journal-datasync>
<journal-min-files>2</journal-min-files>
<journal-pool-files>-1</journal-pool-files>
<message-expiry-scan-period>1000</message-expiry-scan-period>
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
<!--
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<!-- <network-check-period>10000</network-check-period> -->
<!-- <network-check-timeout>1000</network-check-timeout> -->
<!-- this is a comma separated list, no spaces, just DNS or IPs
it should accept IPV6
Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.
You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
<!-- <network-check-list>10.0.0.1</network-check-list> -->
<!-- use this to customize the ping used for ipv4 addresses -->
<!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
<!-- use this to customize the ping used for ipv6 addresses -->
<!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
<!-- once the disk hits this limit the system will block, or close the connection in certain protocols
that won't support flow control. -->
<max-disk-usage>90</max-disk-usage>
<!-- the system will enter into page mode once you hit this limit.
This is an estimate in bytes of how much the messages are using in memory
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
<global-max-size>100Mb</global-max-size>
-->
<acceptors>
<!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
<!-- useKQueue means: it will use Netty kqueue if you are on a system (MacOS) that supports it -->
<!-- amqpCredits: The number of credits sent to AMQP producers -->
<!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
<!-- Acceptor for every supported protocol -->
<acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;useKQueue;amqpCredits=1000;amqpLowCredits=300;actorThresholdBytes=10000</acceptor>
<!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic.-->
<acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;useKQueue=true;amqpCredits=1000;amqpLowCredits=300</acceptor>
<!-- STOMP Acceptor. -->
<acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true;useKQueue=true</acceptor>
<!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
<acceptor name="hornetq">tcp://0.0.0.0:5445?protocols=HORNETQ,STOMP;useEpoll=true;useKQueue=true</acceptor>
<!-- MQTT Acceptor -->
<acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true;useKQueue=true</acceptor>
</acceptors>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="guest"/>
<permission type="deleteNonDurableQueue" roles="guest"/>
<permission type="createDurableQueue" roles="guest"/>
<permission type="deleteDurableQueue" roles="guest"/>
<permission type="createAddress" roles="guest"/>
<permission type="deleteAddress" roles="guest"/>
<permission type="consume" roles="guest"/>
<permission type="browse" roles="guest"/>
<permission type="send" roles="guest"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="guest"/>
</security-setting>
</security-settings>
<address-settings>
<!-- if you define auto-create on certain queues, management has to be auto-create -->
<address-setting match="activemq.management#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
</address-setting>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<max-size-messages>1000</max-size-messages>
<page-size-bytes>10000</page-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
</address-setting>
</address-settings>
<addresses>
<address name="DLQ">
<anycast>
<queue name="DLQ" />
</anycast>
</address>
<address name="ExpiryQueue">
<anycast>
<queue name="ExpiryQueue" />
</anycast>
</address>
</addresses>
</core>
</configuration>

184
tests/smoke-tests/src/main/resources/servers/pagingGlobalMaxMessages/broker.xml Normal file
View File

@ -0,0 +1,184 @@
<?xml version='1.0'?>
<!--
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.
-->
<configuration xmlns="urn:activemq"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq:core ">
<name>0.0.0.0</name>
<persistence-enabled>true</persistence-enabled>
<!-- this could be ASYNCIO or NIO
-->
<journal-type>NIO</journal-type>
<paging-directory>./data/paging</paging-directory>
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/large-messages</large-messages-directory>
<journal-datasync>true</journal-datasync>
<journal-min-files>2</journal-min-files>
<journal-pool-files>-1</journal-pool-files>
<message-expiry-scan-period>1000</message-expiry-scan-period>
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
<!--
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<!-- <network-check-period>10000</network-check-period> -->
<!-- <network-check-timeout>1000</network-check-timeout> -->
<!-- this is a comma separated list, no spaces, just DNS or IPs
it should accept IPV6
Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.
You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
<!-- <network-check-list>10.0.0.1</network-check-list> -->
<!-- use this to customize the ping used for ipv4 addresses -->
<!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
<!-- use this to customize the ping used for ipv6 addresses -->
<!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
<!-- once the disk hits this limit the system will block, or close the connection in certain protocols
that won't support flow control. -->
<max-disk-usage>90</max-disk-usage>
<global-max-messages>1000</global-max-messages>
<!-- the system will enter into page mode once you hit this limit.
This is an estimate in bytes of how much the messages are using in memory
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
<global-max-size>100Mb</global-max-size>
-->
<acceptors>
<!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
<!-- useKQueue means: it will use Netty kqueue if you are on a system (MacOS) that supports it -->
<!-- amqpCredits: The number of credits sent to AMQP producers -->
<!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
<!-- Acceptor for every supported protocol -->
<acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;useKQueue;amqpCredits=1000;amqpLowCredits=300;actorThresholdBytes=10000</acceptor>
<!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic.-->
<acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;useKQueue=true;amqpCredits=1000;amqpLowCredits=300</acceptor>
<!-- STOMP Acceptor. -->
<acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true;useKQueue=true</acceptor>
<!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
<acceptor name="hornetq">tcp://0.0.0.0:5445?protocols=HORNETQ,STOMP;useEpoll=true;useKQueue=true</acceptor>
<!-- MQTT Acceptor -->
<acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true;useKQueue=true</acceptor>
</acceptors>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="guest"/>
<permission type="deleteNonDurableQueue" roles="guest"/>
<permission type="createDurableQueue" roles="guest"/>
<permission type="deleteDurableQueue" roles="guest"/>
<permission type="createAddress" roles="guest"/>
<permission type="deleteAddress" roles="guest"/>
<permission type="consume" roles="guest"/>
<permission type="browse" roles="guest"/>
<permission type="send" roles="guest"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="guest"/>
</security-setting>
</security-settings>
<address-settings>
<!-- if you define auto-create on certain queues, management has to be auto-create -->
<address-setting match="activemq.management#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
</address-setting>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<page-size-bytes>10000</page-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
</address-setting>
</address-settings>
<addresses>
<address name="DLQ">
<anycast>
<queue name="DLQ" />
</anycast>
</address>
<address name="ExpiryQueue">
<anycast>
<queue name="ExpiryQueue" />
</anycast>
</address>
</addresses>
</core>
</configuration>

94
tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/paging/SmokeMaxMessagePagingTest.java Normal file
View File

@ -0,0 +1,94 @@
/*
* 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.tests.smoke.paging;
import java.io.File;
import org.apache.activemq.artemis.cli.commands.ActionContext;
import org.apache.activemq.artemis.cli.commands.messages.Producer;
import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class SmokeMaxMessagePagingTest extends SmokeTestBase {
public static final String SERVER_NAME_GLOBAL = "pagingGlobalMaxMessages";
public static final String SERVER_NAME_ADDRESS = "pagingAddressMaxMessages";
@Before
public void before() throws Exception {
}
@Test
public void testGlobalMaxSend() throws Exception {
internalTestSend(SERVER_NAME_GLOBAL);
}
@Test
public void testAddressMaxSend() throws Exception {
internalTestSend(SERVER_NAME_ADDRESS);
}
public void internalTestSend(String serverName) throws Exception {
cleanupData(serverName);
startServer(serverName, 0, 30000);
internalSend("core", 2000);
Assert.assertTrue("System did not page", isPaging(serverName));
}
boolean isPaging(String serverName) {
File location = new File(getServerLocation(serverName));
File paging = new File(location, "data/paging");
File[] pagingContents = paging.listFiles();
return pagingContents != null && pagingContents.length > 0;
}
@Test
public void testGlobalMaxSendRestart() throws Exception {
internalTestSendWithRestart(SERVER_NAME_GLOBAL);
}
@Test
public void testAddressMaxSendRestart() throws Exception {
internalTestSendWithRestart(SERVER_NAME_ADDRESS);
}
public void internalTestSendWithRestart(String serverName) throws Exception {
cleanupData(serverName);
Process process = startServer(serverName, 0, 30000);
internalSend("core", 500);
Assert.assertFalse(isPaging(serverName));
process.destroy();
process = startServer(serverName, 0, 30000);
internalSend("core", 1500);
Assert.assertTrue(isPaging(serverName));
}
private void internalSend(String protocol, int numberOfMessages) throws Exception {
Producer producer = (Producer)new Producer().setMessageSize(1).setMessageCount(numberOfMessages).setTxBatchSize(500);
producer.setProtocol(protocol);
producer.setSilentInput(true);
producer.execute(new ActionContext());
}
}