You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by "gaohoward (via GitHub)" <gi...@apache.org> on 2024/08/05 14:53:56 UTC
[PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
gaohoward opened a new pull request, #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705696846
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
hmm, that's ugly. I think we can do
1. make sure pageLimitBytes is greater than or equal pageSize (i.e. at least one page file should be allowed)
2. if page store is not running (running flag is false), check the number of pages in the store and compare the new max estimated pages, and return false if the number of the new estimated pages is smaller than the number of existing page files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705762102
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
The limit > page size requirement would remove any issue with e.g 0 page files at the low end. Though essentially the same thing occurs with `n` files at the top end, unless the limit is always a multiple of the page size.
The main issue here is that the limit is tracked by a file count based on current limit/size, and so any existing 'old' files of the 'wrong' size can over/underutilise the effective limit if you adjust the page size with outstanding page files. There doesnt seem to be a way around that without byte-level tracking, which is likely to be awkward.
I think we should perhaps just better document that the effective limit is a file count based on current limit/page size, and so changing page size whilst there are existing outstanding pages can lead to such behaviour and if you want to influence things to e.g unblock earlier it might be necessary to adjust the limit at the same time whilst those old files are present.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1718212763
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
Review Comment:
Since at no point has it been validated these getters are returning the values we expect, this isnt strictly checking its working.
We should either use the initial PageSizeBytes and PageLimitBytes values we provided in the settings to calculate the expected pages limit, or else check that the getters are returning those values (or do both) as e.g the previous test did for PageLimitBytes.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
Review Comment:
procuder -> producer
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // current pages not exceeds the max pages
+ Long currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
Review Comment:
Again can verify against pre-known value, and/or verify the getters are accurate.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // current pages not exceeds the max pages
+ Long currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // however messages reaches limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(maxMessages, queuePagingStore.getPageLimitMessages());
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes((long) (size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless of current page count > (pageLimitBytes / pageSizeBytes)
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ // verify current situation
+ queue = server.locateQueue(queueAddr);
+
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging() && queuePagingStore.isPageFull());
+
+ long currentPages = queuePagingStore.getNumberOfPages();
+ assertEquals(existingPages, currentPages);
+
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
Review Comment:
Again can verify against pre-known value, and/or verify the getters are accurate.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -232,7 +233,7 @@ private void configureSizeMetric() {
* @param addressSettings
*/
@Override
- public void applySetting(final AddressSettings addressSettings) {
+ public void applySetting(final AddressSettings addressSettings, final boolean firstTime) {
Review Comment:
Since it is really an implementation detail that the constructor calls the applySetting method, I think this interface method should probably just go back to the way it was, and a private internal second method be created with the firstTime boolean that both this methods implementation and the constructor can then call. That way the interface and none of the tests etc need to be updated to support calling with false (and they shouldnt be able to call true)
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
Review Comment:
Similarly here too
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
Review Comment:
Similarly here, it should validate against a pre-known value and/or check the getters are correct, rather than trusting that they are.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
Review Comment:
queuePagingStore != null is superfluous her, since it already checked it wasnt null, and havent changed the variable since (but have used it again without checking it isnt null)
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
Review Comment:
Would be clearer to use assertNotNull rather than the combined check
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
Review Comment:
Similarly here also
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // current pages not exceeds the max pages
+ Long currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // however messages reaches limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(maxMessages, queuePagingStore.getPageLimitMessages());
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes((long) (size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless of current page count > (pageLimitBytes / pageSizeBytes)
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ // verify current situation
+ queue = server.locateQueue(queueAddr);
+
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging() && queuePagingStore.isPageFull());
+
+ long currentPages = queuePagingStore.getNumberOfPages();
+ assertEquals(existingPages, currentPages);
+
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ //consume messages until current pages goes down to below maxPage
+ locator = createFactory(isNetty());
+ final int numMessages = 25;
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ session = csf.createSession(false, true);
+ session.start();
+ ClientConsumer consumer = session.createConsumer(queueName);
+ for (int i = 0; i < numMessages; i++) {
+ ClientMessage message = consumer.receive(5000);
+ assertNotNull(message);
+ message.acknowledge();
+ session.commit();
+ }
+
+ currentPages = queuePagingStore.getNumberOfPages();
+ assertTrue(currentPages < maxPages);
+ // check page store not page full
+ assertTrue(queuePagingStore.isPaging() && !queuePagingStore.isPageFull());
+
+ //send messages one by one until page full
+ ClientProducer producer = session.createProducer(queueName);
+ boolean isFull = false;
+ try {
+ for (int i = 0; i < numMessages; i++) {
+ ClientMessage message = createMessage(session, messageSize, i, null);
+ producer.send(message);
+ session.commit();
+ }
+ } catch (ActiveMQAddressFullException e) {
+ isFull = true;
+ session.close();
Review Comment:
session close should move after the try-catch as we always want to do it, and it would otherwise fail to happen if the loop somehow succeeded.
EDIT: actually, since the wider try-catch with the ClientSessionFactory csf is there, that means it always closes right...so can this just be omitted rather than look wrong?
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // current pages not exceeds the max pages
+ Long currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // however messages reaches limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(maxMessages, queuePagingStore.getPageLimitMessages());
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes((long) (size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless of current page count > (pageLimitBytes / pageSizeBytes)
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ // verify current situation
+ queue = server.locateQueue(queueAddr);
+
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging() && queuePagingStore.isPageFull());
Review Comment:
assertNotNull
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
Review Comment:
No need for null check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1715547459
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -268,7 +269,12 @@ public void applySetting(final AddressSettings addressSettings) {
pageLimitBytes = null;
}
- pageLimitMessages = addressSettings.getPageLimitMessages();
+ if (!Objects.equals(this.pageLimitMessages, addressSettings.getPageLimitMessages())) {
+ pageLimitMessages = addressSettings.getPageLimitMessages();
+ if (this.cursorProvider != null) {
+ this.cursorProvider.checkClearPageLimit();
+ }
+ }
Review Comment:
Feels like this should only be actioned after the null-replacement being done immediately below here. I'd take a variable copy of the existing value here, and then move this overall 'do if changed' block below the null replacement below...much as you did for the later bits.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +295,290 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes(Long.valueOf(size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidation() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // now pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ // but the messages reach the limit
+ PageCursorProviderImpl cursorProvider = (PageCursorProviderImpl) queuePagingStore.getCursorProvider();
+ assertTrue(cursorProvider.getNumberOfMessagesOnSubscriptions() == maxMessages);
Review Comment:
Expand the use of PageCursorProviderTestAccessor rather than making the method public so you can do this
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -343,6 +353,10 @@ private void checkNumberOfPages() {
if (!isBelowPageLimitBytes()) {
this.pageFull = true;
ActiveMQServerLogger.LOGGER.pageFullMaxBytes(storeName, numberOfPages, estimatedMaxPages, pageLimitBytes, pageSize);
+ } else {
+ if (cursorProvider != null) {
+ cursorProvider.checkClearPageLimit();
+ }
Review Comment:
It feels wrong for the wider method called checkNumberOfPages() to also be doing this, when its more aimed at the message count stuff. Especially as you already call this directly if the message limit changes.
EDIT: see suggestion from earlier comment.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -290,7 +296,11 @@ public void applySetting(final AddressSettings addressSettings) {
}
if (pageLimitBytes != null && pageSize > 0) {
+ Long originalEstimatedMaxPages = this.estimatedMaxPages;
estimatedMaxPages = pageLimitBytes / pageSize;
+ if (!Objects.equals(estimatedMaxPages, originalEstimatedMaxPages)) {
+ checkNumberOfPages();
+ }
Review Comment:
I would move this below the log statement, otherwise it has possibly gone away and done stuff based on the change, before logging what it was.
I also dont think it should ever be doing this for the constructor case, as it hasnt finished creating the object yet and is pre-empting start() doing this later. The earlier change for message count kind of gets away with it due to the null check on cursorProvider (which is actually final and so only could be null during the constructor), but its not clear this one does. It might be time to pass a boolean to indicate whether this shared method is being called from the constructor or later, so it simply cant do these things at construction (allowing removal of the null check)
I think this bit, and the pageLimitMessages related check earlier, should perhaps just set booleans to indicate if they changed, which you then action at the end after both have been updated, by calling checkNumberOfPages() and cursorProvider.checkClearPageLimit() respectively (again, if not acting for the constructor)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2285249462
> To be honest, given the uglyness from not applying the settings but then only logging about it, and the other uglyness from throwing (potentially preventing other valid settings being applied)...
>
> I find myself wondering if this should either just be doc-only and left working the way it was...or else, adding the doc changes and just a log message to indicate setting pageLimit < pageSize will cause immediate block, but no 'dont apply these settings' step so there is no unexpected knock on effects.
Yeah make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2285910308
@gemmellr PR updated. Please take a look. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2288744331
@gemmellr Hi Robbie I update the PR. Please take a look again see if there is any issues. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic merged PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1706678398
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
I've added test to cover the outstanding page files ( > 1). Internally we use number of page files to calculate the limit against the limit which is specified by the bytes. I guess that is fine as long as the user understands and use them properly. I'll add some document for that.
Regarding jdbc I haven't noticed the limit also apply to database store. I'll take a look and see what impact would it be.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705515013
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java:
##########
@@ -559,4 +560,6 @@ IllegalStateException invalidRoutingTypeUpdate(String queueName,
@Message(id = 229255, value = "Bridge {} cannot be {}. Current state: {}")
ActiveMQIllegalStateException bridgeOperationCannotBeExecuted(String bridgeName, String failedOp, BridgeImpl.State currentState);
+ @Message(id = 229256, value = "Invalid address settings {} for page store {}. Details: {}")
+ ActiveMQException invalidPageSettings(String pageStoreName, AddressSettings addressSettings, String details);
Review Comment:
good catch. I forgot to delete it. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708994571
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
Yep, the JDBC bit was based on a mis-read.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708487795
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +291,103 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
Review Comment:
That makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1704871499
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +229,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private void validateNewSettings(final AddressSettings addressSettings) throws ActiveMQException {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ throw ActiveMQMessageBundle.BUNDLE.invalidPageSettings(this.storeName.toString(), addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
Review Comment:
Make sense. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1718225568
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +296,408 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+ final String addressName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of(addressName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) size);
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ assertFalse(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to be smaller than pageSizeBytes
+ addressSettings.setPageLimitBytes((long) (size - 1));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // check the settings applied
+ assertEquals(size - 1, queuePagingStore.getPageLimitBytes());
+
+ // check pageFull is true
+ assertTrue(queuePagingStore.isPageFull());
+
+ // send a messages should be immediately blocked (in our case FAIL)
+ try {
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ fail("should be immediate blocked on paging");
+ } catch (ActiveMQAddressFullException ex) {
+ //ok
+ }
+
+ assertTrue(queuePagingStore.isPageFull());
+
+ // set pageLimitBytes to bigger value to unblock paging again
+ addressSettings.setPageLimitBytes((long) (size * 2));
+ server.getAddressSettingsRepository().addMatch(addressName, addressSettings);
+
+ // now page is enabled again
+ assertFalse(queuePagingStore.isPageFull());
+
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitMessagesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ final int size = 1024 * 50;
+ final Long maxMessages = 10L;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes((long) (size * 10));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the messages reach the limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider), maxMessages);
+
+ // but pages still under limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages <= queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes());
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but current pages still under limit
+ long currentPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages <= maxPages);
+
+ // now increase the max messages to unblock procuder
+ final Long bigLimitMessages = 1000L;
+ addressSettings.setPageLimitMessages(bigLimitMessages);
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertFalse(queuePagingStore.isPageFull());
+
+ // now send more messages until pagefull
+ locator = createFactory(isNetty());
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ stop = false;
+ session = csf.createSession(true, true);
+ producer = session.createProducer(queueAddr);
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ assertTrue(totalMessages <= bigLimitMessages, "test is broken");
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+ }
+
+ // check the page full
+ assertTrue(queuePagingStore.isPageFull());
+ // because it reaches pageLimitBytes
+ currentPages = queuePagingStore.getNumberOfPages();
+ maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ // and messages still below limit messages
+ cursorProvider = queuePagingStore.getCursorProvider();
+ assertEquals(bigLimitMessages, queuePagingStore.getPageLimitMessages());
+ assertTrue(PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider) < queuePagingStore.getPageLimitMessages());
+ } finally {
+ server.stop(true);
+ }
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesAndPageLimitMessagesValidationBlockOnLimitBytesFirst() throws Exception {
+
+ final String queueName = getTestMethodName();
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(true, true);
+
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ // the numbers should make sure page files reach the pageLimitBytes before pageLimitMessages
+ final int size = 1024 * 50;
+ final Long maxMessages = 200L;
+
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 5));
+ addressSettings.setMaxSizeBytes(size);
+ addressSettings.setPageLimitMessages(maxMessages);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 0;
+ int messageSize = 1024;
+ ClientProducer producer = session.createProducer(queueAddr);
+ boolean stop = false;
+ while (!stop) {
+ try {
+ ClientMessage message = createMessage(session, messageSize, totalMessages, null);
+ producer.send(message);
+ totalMessages++;
+ session.commit();
+ } catch (ActiveMQAddressFullException ex) {
+ stop = true;
+ }
+ }
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // the pages reach the limit
+ long existingPages = queuePagingStore.getNumberOfPages();
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+
+ assertTrue(existingPages > maxPages);
+
+ // but messages still under limit
+ PageCursorProvider cursorProvider = queuePagingStore.getCursorProvider();
+ Long existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be less than max " + maxMessages);
+
+ server.stop(true);
+ waitForServerToStop(server);
+
+ // restart the server the pageFull is still true
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPageFull());
+
+ // but messages still under limit
+ cursorProvider = queuePagingStore.getCursorProvider();
+ existingMessages = PageCursorProviderTestAccessor.getNumberOfMessagesOnSubscriptions(cursorProvider);
+ assertTrue(existingMessages < maxMessages, "existing " + existingMessages + " should be still less than max " + maxMessages);
+
+ // now increase the pageLimitBytes to unblock procuder
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 20));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // no longer page full
+ assertTrue(queuePagingStore != null && !queuePagingStore.isPageFull());
Review Comment:
queuePagingStore != null is superfluous here, since it already checked it wasnt null, and havent changed the variable since (but have used it again without checking it isnt null)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2292621656
@gemmellr Thanks Robbie!
@clebertsuconic Hi Clebert can you take a look? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2291035710
got 2 test failures but seems not relevant. They all passed in my local (1000 times in a row).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1717017555
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -289,9 +290,23 @@ public void applySetting(final AddressSettings addressSettings) {
this.pageLimitBytes = null;
}
+ this.pageLimitMessages = addressSettings.getPageLimitMessages();
Review Comment:
This is in the wrong place, it should be _before_ the matching 'null replacement check' for the value, up on line 273, i.e where it was to begin with (originally as line 271), otherwise it may think the value has changed when it hasnt (but the comparison fouled because the null replacement has been 'bypassed' here).
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -289,9 +290,23 @@ public void applySetting(final AddressSettings addressSettings) {
this.pageLimitBytes = null;
}
+ this.pageLimitMessages = addressSettings.getPageLimitMessages();
+ if (firstTime || !Objects.equals(this.pageLimitMessages, originalLimitMessages)) {
+ if (this.cursorProvider != null) {
+ this.cursorProvider.checkClearPageLimit();
+ }
+ }
Review Comment:
The cursorProvider hasnt been created at the point this is called during the constructor, so its never going to be non-null if 'firstTime' is true and it cant be non-null in real usage if it is false. My suggestion on the prior round was actually that it should _not_ do this for the constructor, since this is handled via the start etc processes later after construction and so would basically be duplicate.
This is also going to check with the _previous_ estimated max pages rather than any new value, because that value hasnt been updated yet (its done later below). I dont think that makes sense, as makes it possible this may want to clear the limit, only for the very next bit to reinstate it because it should never have been cleared with the updated setting.
As I said, while applying the settings values, I would use a flag for each of pageLimitMessages and the estimated max pages, to remember whether they have changed....then at end, after application, check things when both are effective together.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -289,9 +290,23 @@ public void applySetting(final AddressSettings addressSettings) {
this.pageLimitBytes = null;
}
+ this.pageLimitMessages = addressSettings.getPageLimitMessages();
+ if (firstTime || !Objects.equals(this.pageLimitMessages, originalLimitMessages)) {
+ if (this.cursorProvider != null) {
+ this.cursorProvider.checkClearPageLimit();
+ }
+ }
+
if (pageLimitBytes != null && pageSize > 0) {
+ Long originalEstimatedMaxPages = this.estimatedMaxPages;
estimatedMaxPages = pageLimitBytes / pageSize;
logger.debug("Address {} should not allow more than {} pages", storeName, estimatedMaxPages);
+ if (firstTime || !Objects.equals(estimatedMaxPages, originalEstimatedMaxPages)) {
+ checkNumberOfPages();
+ if (cursorProvider != null) {
+ cursorProvider.checkClearPageLimit();
+ }
+ }
Review Comment:
Similarly, since start() checks at least the page count, and any use of the address checks the message limits, the suggestion was that it not do this for the constructor., but only later if the settings have changed. This instead always does it for the constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708515262
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
+
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+
+ long existingNumberOfPages;
+ if (!running) {
+ try {
+ checkFileFactory();
+ List<String> files = this.fileFactory.listFiles("page");
+ existingNumberOfPages = files.size();
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "failed to get existing page files with exception: " + e);
+ return false;
+ }
+ } else {
+ existingNumberOfPages = this.numberOfPages;
+ }
+
+ if (existingNumberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " (pageLimitBytes / pageSize) is less than current number of pages " + existingNumberOfPages);
+ return false;
+ }
Review Comment:
I think your idea is more friendly to the users. But that'll change the current implementation of it. According to current impl, the practical usage of pageLimitBytes would be a hard rule, i.e. in that case the user has to consume the messages until the current pages is below the expected new limit, then apply the new limit. Based on your idea the new limit can be set at anytime during broker running. something worth to do. Anyway we need to document that behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2289111952
The current changes also look to cause failures, in several tests in the full test suite, and the examples.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2290463735
@gemmellr PR updated. I'll run the test suite and see if it breaks anything. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2284314913
To be honest, given the uglyness from not applying the settings but then only logging about it, and the other uglyness from throwing (potentially preventing other valid settings being applied)...
I find myself wondering if this should either just be doc-only and left working the way it was...or else, adding the doc changes and just a log message to indicate setting pageLimit < pageSize will cause immediate block, but no 'dont apply these settings' step so there is no unexpected knock on effects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1706683089
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java:
##########
@@ -559,4 +560,6 @@ IllegalStateException invalidRoutingTypeUpdate(String queueName,
@Message(id = 229255, value = "Bridge {} cannot be {}. Current state: {}")
ActiveMQIllegalStateException bridgeOperationCannotBeExecuted(String bridgeName, String failedOp, BridgeImpl.State currentState);
+ @Message(id = 229256, value = "Invalid address settings {} for page store {}. Details: {}")
+ ActiveMQException invalidPageSettings(String pageStoreName, AddressSettings addressSettings, String details);
Review Comment:
ah you mean in constructor scenario we can throw an exception instead of log a warning? It seems makes sense.
I'll add it back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705664759
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java:
##########
@@ -559,4 +560,6 @@ IllegalStateException invalidRoutingTypeUpdate(String queueName,
@Message(id = 229255, value = "Bridge {} cannot be {}. Current state: {}")
ActiveMQIllegalStateException bridgeOperationCannotBeExecuted(String bridgeName, String failedOp, BridgeImpl.State currentState);
+ @Message(id = 229256, value = "Invalid address settings {} for page store {}. Details: {}")
+ ActiveMQException invalidPageSettings(String pageStoreName, AddressSettings addressSettings, String details);
Review Comment:
Perhaps, but it seemed more like it should be getting used hehe, e.g in the constructor scenario from my other comment :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1709002549
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +291,103 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server and the invalid settings are not applied too.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+ queuePagingStore = queue.getPagingStore();
+
+ // check settings not applied
+ assertEquals(0, queuePagingStore.getPageSizeBytes());
+ assertNull(queuePagingStore.getPageFullMessagePolicy());
+ assertNull(queuePagingStore.getPageLimitBytes());
+ assertEquals(0, queuePagingStore.getMaxSize());
Review Comment:
Would be good for the test to assert the expected behaviour if it can. Though that seems like crummy behaviour, I don't see why it wouldn't just act as if it was at the limit until stuff was consumed and it was below the limit. By all means it could log an info/warning message about being over (as it would here, but without mentioning that this was breaking the address entirely and they need to fix+restart), but stopping the address working seems weird.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1714599261
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java:
##########
@@ -196,7 +197,11 @@ public void onChange() {
private void reapplySettings() {
for (PagingStore store : stores.values()) {
AddressSettings settings = this.addressSettingsRepository.getMatch(store.getAddress().toString());
- store.applySetting(settings);
+ try {
+ store.applySetting(settings);
+ } catch (ActiveMQException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
That's true.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1715108779
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -291,6 +291,7 @@ public void applySetting(final AddressSettings addressSettings) {
if (pageLimitBytes != null && pageSize > 0) {
estimatedMaxPages = pageLimitBytes / pageSize;
+ checkNumberOfPages();
Review Comment:
Calling this here would have it act before start() in the constructor case, which seems like it could lead to duplicate logging.
In the non-constructor case it seems like it could cause logging every time any [possibly-unrelated] setting changes, which may also not be desirable. Maybe it could check if any of those settings changed?
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -343,6 +344,8 @@ private void checkNumberOfPages() {
if (!isBelowPageLimitBytes()) {
this.pageFull = true;
ActiveMQServerLogger.LOGGER.pageFullMaxBytes(storeName, numberOfPages, estimatedMaxPages, pageLimitBytes, pageSize);
+ } else {
+ pageLimitReleased();
Review Comment:
This seems wrong. Just because it isnt over the file-based paging limit, doesnt mean it isnt still 'full'; there is a separate message-based limit which this would appear to essentially disregard, effectively bypassing it any time this method is called...see related use of `checkPageLimit(long numberOfMessages)` (next method right below this).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705518732
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
Review Comment:
I agree. Will insert some more to improve the spacing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708515262
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
+
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+
+ long existingNumberOfPages;
+ if (!running) {
+ try {
+ checkFileFactory();
+ List<String> files = this.fileFactory.listFiles("page");
+ existingNumberOfPages = files.size();
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "failed to get existing page files with exception: " + e);
+ return false;
+ }
+ } else {
+ existingNumberOfPages = this.numberOfPages;
+ }
+
+ if (existingNumberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " (pageLimitBytes / pageSize) is less than current number of pages " + existingNumberOfPages);
+ return false;
+ }
Review Comment:
I think your idea is more friendly to the users. But that'll change the current implementation of it. According to current impl, the practical usage of pageLimitBytes would be a hard rule, i.e. in that case the user has to consume the messages until the current pages is below the expected new limit, that apply the new limit. Based on your idea the new limit can be set at anytime during broker running. something worth to do. Anyway we need to document that behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2291343015
@gemmellr PR updated. Please review. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2291506169
LGTM but I'd like someone else to review.
I'd also note the latest versions have removed the specific warning log message for pageSize > pageLimitBytes...though personally, since it does log a warn when it starts blocking paging, and that mentions the file counts, I think thats already implicit and fine as-is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705764377
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
Another thing I realised is that the page size is adjusted further down this method for the JDBC use case. So validating the value at the start of this method, before that later change occurs, is likely to give the wrong behaviour.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708493050
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
Review Comment:
I'm fine with either options. probably throw can mean more serious about that and user gets more attention to correct their settings.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705323890
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
Review Comment:
The .longValue() call seems redundant
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
Review Comment:
"estimated max pages \<foo\> is less than current number of pages \<bar\>" doesnt do a great job of conveying it is the configured page size and page limit bytes that are responsible for the 'estimated max pages'.
Perhaps add an indicator that the estimate is calculated from them, e.g "estimated max pages \<foo\> (page limit / page size) is less than current number of pages \<bar\>" ?
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
The containing applySetting method is also called from the constructor, but this could make it 'silently' (not counting the logging, which is out of band) abort and yet successfully return without apply _any_ of the settings, and so perhaps just continue on with entirely the wrong settings. That feels off.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
Review Comment:
If we can manage 2 newline spacers here, seems like we could manage some below here in the actual logic for improved readability.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java:
##########
@@ -559,4 +560,6 @@ IllegalStateException invalidRoutingTypeUpdate(String queueName,
@Message(id = 229255, value = "Bridge {} cannot be {}. Current state: {}")
ActiveMQIllegalStateException bridgeOperationCannotBeExecuted(String bridgeName, String failedOp, BridgeImpl.State currentState);
+ @Message(id = 229256, value = "Invalid address settings {} for page store {}. Details: {}")
+ ActiveMQException invalidPageSettings(String pageStoreName, AddressSettings addressSettings, String details);
Review Comment:
This new exception doesnt appear to be getting thrown anywhere
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705521745
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
Review Comment:
yeah that's better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1713165121
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
+
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+
+ long existingNumberOfPages;
+ if (!running) {
+ try {
+ checkFileFactory();
+ List<String> files = this.fileFactory.listFiles("page");
+ existingNumberOfPages = files.size();
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "failed to get existing page files with exception: " + e);
+ return false;
+ }
+ } else {
+ existingNumberOfPages = this.numberOfPages;
+ }
+
+ if (existingNumberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " (pageLimitBytes / pageSize) is less than current number of pages " + existingNumberOfPages);
+ return false;
+ }
Review Comment:
I've updated the PR and a bit document. Please review see if I missed anything. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2286513460
@gemmellr Just updated the PR to address the problems and added one more test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1713944655
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
Review Comment:
Rather than using "FOO" literals here and various points below, assigning a variable would allow following where all the name is used. Picking a more specific name could also assist with following things later (e.g debugging logs etc).
For example, perhaps:
final String addressName = getTestMethodName();
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java:
##########
@@ -196,7 +197,11 @@ public void onChange() {
private void reapplySettings() {
for (PagingStore store : stores.values()) {
AddressSettings settings = this.addressSettingsRepository.getMatch(store.getAddress().toString());
- store.applySetting(settings);
+ try {
+ store.applySetting(settings);
+ } catch (ActiveMQException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
One aspect of this change is that if it fails the validation and throws applying for one address, this bit here looping over all the addresses means that it wont even try for the addresses following one that fails. The feels potentially problematic as it could leave them with unexpected settings, and yet its entirely possible _their_ settings would have applied ok (i.e theirs may have changed but yet not be invalid).
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ final String queueName = "FOO";
Review Comment:
Similarly here
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ final String queueName = "FOO";
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless current pages > pageLimitBytes
Review Comment:
```suggestion
// server will start regardless of current page count > (pageLimitBytes / pageSizeBytes)
```
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ final String queueName = "FOO";
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
Review Comment:
Setting the address full policy to PAGE explicitly would be good as in other test.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ final String queueName = "FOO";
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless current pages > pageLimitBytes
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ // verify current situation
+ queue = server.locateQueue(queueAddr);
+
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging() && queuePagingStore.isPageFull());
+
+ long currentPages = queuePagingStore.getNumberOfPages();
+ assertEquals(existingPages, currentPages);
+
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ //consume messages until current pages goes down to below maxPage
+ locator = createFactory(isNetty());
+ final int numMessages = 25;
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ session = csf.createSession(false, true);
+ session.start();
+ ClientConsumer consumer = session.createConsumer(queueName);
+ for (int i = 0; i < numMessages; i++) {
+ ClientMessage message = consumer.receive(5000);
+ assertNotNull(message);
+ message.acknowledge();
+ session.commit();
+ }
+
+ currentPages = queuePagingStore.getNumberOfPages();
+ assertTrue(currentPages < maxPages);
Review Comment:
Should it asssert that that paging store is _not_ pageFull, rather than assuming it isnt? The below bits could fail on the first send attempt and erroneously pass since they never actually check the store.
##########
docs/user-manual/paging.adoc:
##########
@@ -244,7 +244,16 @@ This is to avoid a single destination using the entire disk in case their consum
You can configure either `page-limit-bytes` or `page-limit-messages`, along with `page-full-policy` on the address settings limiting how much data will be recorded in paging.
-If you configure `page-full-policy` as DROP, messages will be simplify dropped while the clients will not get any exceptions, while if you configured FAIL the producers will receive a JMS Exception for the error condition.
+If you configure `page-full-policy` as DROP, messages will be simply dropped while the clients will not get any exceptions, while if you configured FAIL the producers will receive a JMS Exception for the error condition.
+
+[NOTE]
+
+The `page-limit-bytes` is converted to number of page files internally (i.e. `page-limit-bytes` / `page-size-bytes`) and used to compare against the current number of page files.
+Its value should be equal or greater than `page-size-bytes` to allow at least one page to be created.
+If the applied value of `page-limit-bytes`, when converted to number of pages, is less than the number of the current page files in store,
+paging will be blocked based on `page-full-policy` until the number current page files go down below or equal to the
+value of `page-limit-bytes`, and it will become pageFull again once the number of page files is greater than
+the value determined by `page-limit-bytes` (`page-limit-bytes` / `page-size-bytes`).
Review Comment:
```suggestion
The `page-limit-bytes` is used to identify a maximum number of page files internally (i.e. `page-limit-bytes` / `page-size-bytes`) which is then compared against the current number of page files.
If configured, `page-limit-bytes` must be equal or greater than `page-size-bytes` to allow at least one page to be created.
If the limit determined from `page-limit-bytes`, once converted to a number of pages, is less than the current number of page files in the store then paging will be blocked based on `page-full-policy` until the number of current page files drops to less than or equal to the calculated file limit. It will become blocked again once the number of page files is greater than the value determined by `page-limit-bytes` (`page-limit-bytes` / `page-size-bytes`).
```
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +292,149 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ final String queueName = "FOO";
+ SimpleString queueAddr = SimpleString.of(queueName);
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server with a smaller pageLimitSize < existing pages.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch(queueName, addressSettings);
+
+ // server will start regardless current pages > pageLimitBytes
+ try {
+ server.start();
+ waitForServerToStart(server);
+
+ // verify current situation
+ queue = server.locateQueue(queueAddr);
+
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging() && queuePagingStore.isPageFull());
+
+ long currentPages = queuePagingStore.getNumberOfPages();
+ assertEquals(existingPages, currentPages);
+
+ long maxPages = queuePagingStore.getPageLimitBytes() / queuePagingStore.getPageSizeBytes();
+ assertTrue(currentPages > maxPages);
+
+ //consume messages until current pages goes down to below maxPage
+ locator = createFactory(isNetty());
+ final int numMessages = 25;
+ try (ClientSessionFactory csf = createSessionFactory(locator)) {
+ session = csf.createSession(false, true);
+ session.start();
+ ClientConsumer consumer = session.createConsumer(queueName);
+ for (int i = 0; i < numMessages; i++) {
+ ClientMessage message = consumer.receive(5000);
+ assertNotNull(message);
+ message.acknowledge();
+ session.commit();
+ }
+
+ currentPages = queuePagingStore.getNumberOfPages();
+ assertTrue(currentPages < maxPages);
+
+ //send messages one by one until page full
+ ClientProducer producer = session.createProducer(queueName);
+ boolean isFull = false;
+ try {
+ for (int i = 0; i < numMessages; i++) {
+ ClientMessage message = createMessage(session, messageSize, i, null);
+ producer.send(message);
+ session.commit();
+ }
+ } catch (ActiveMQAddressFullException e) {
+ isFull = true;
+ session.close();
+ }
+ assertTrue(isFull);
+ currentPages = queuePagingStore.getNumberOfPages();
+ //now current pages should be one more than maxPages
+ assertTrue(currentPages == maxPages + 1);
Review Comment:
Similarly, checking the store again would make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#issuecomment-2293535558
Looks good to me.. very clean.. no added semantics.. just using what's in there already..
I'm merging it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708467212
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
The validation uses storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes()) to get the effective pageSizeBytes so that would take care of the jdbc case. Anyway the jdbc bit only is only used to store the page bytes and is not involved in the calculation which is done all in pagingStoreImpl.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1707409571
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* @param addressSettings
*/
@Override
public void applySetting(final AddressSettings addressSettings) {
+
+ if (!validateNewSettings(addressSettings)) {
+ return;
+ }
Review Comment:
I dont know how this interacts with the JDBC bits, I just noticed the fact that it messes around with the effective pageSize later in this applySettings method. However I hadnt noticed it was updating a field, and the validateNewSettings() method doesnt use that does calculates the same value locally..so I was probably wrong to be concerned about that bit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1707484372
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
Review Comment:
I think this should probably just throw personally.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
+
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+
+ long existingNumberOfPages;
+ if (!running) {
+ try {
+ checkFileFactory();
+ List<String> files = this.fileFactory.listFiles("page");
+ existingNumberOfPages = files.size();
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "failed to get existing page files with exception: " + e);
+ return false;
+ }
+ } else {
+ existingNumberOfPages = this.numberOfPages;
+ }
+
+ if (existingNumberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " (pageLimitBytes / pageSize) is less than current number of pages " + existingNumberOfPages);
+ return false;
+ }
Review Comment:
Per my other comments, I now question this entirely (not the log message, but the return false). If someone wants to set a limit, that is lower than the current usage, its not clear to me that is a reason not to set it. They can consume those thigns to get back below it, and untilt hen it would act as if at the limit and stop more being added, which is as close to the desired behaviour as its possible to get. Failing to set anything might just mean usage keeps increasing beyond what it already is, i.e the exact opposite of what they want to happen.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +291,103 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
Review Comment:
Setting the address full policy to PAGE explicitly would make this more understandable, and less brittle should the default change
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +291,103 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server and the invalid settings are not applied too.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+ queuePagingStore = queue.getPagingStore();
+
+ // check settings not applied
+ assertEquals(0, queuePagingStore.getPageSizeBytes());
+ assertNull(queuePagingStore.getPageFullMessagePolicy());
+ assertNull(queuePagingStore.getPageLimitBytes());
+ assertEquals(0, queuePagingStore.getMaxSize());
Review Comment:
Its not clear to me what the behavioural state of the address would actually be in this case? Will it allow more paging? Seems like a test should validate the expected behaviour rather than just the return here. If it does allow more paging as it seems it might, then that seems questionable since a limit was attempted to be imposed, and the reason it failed is seemingly because it was already beyond it? That seems off. Why not allow it but log it is beyond limit already, rather than reject it and then permit _yet more_ paging beyond the desired limit as a result?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1705515479
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
Review Comment:
good catch! Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708490696
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingSendTest.java:
##########
@@ -288,6 +291,103 @@ public void testPagingDoesNotDuplicateBatchMessagesAfterPagingStarted() throws E
}
}
+ @Test
+ public void testPageLimitBytesValidation() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1048576;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 15;
+ int messageSize = 90000;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ // set page size bytes to be larger than pageLimitBytes
+ addressSettings.setPageSizeBytes(size * 2);
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ // check the original pageSizeBytes is not changed
+ assertEquals(size, queuePagingStore.getPageSizeBytes());
+
+ // send a messages should be allowed because the page file still have space
+ sendMessageBatch(1, messageSize, session, queueAddr);
+ assertTrue(waitForMessages(queue, totalMessages + 1, 10000));
+ }
+ }
+
+ @Test
+ public void testPageLimitBytesValidationOnRestart() throws Exception {
+
+ try (ClientSessionFactory sf = createSessionFactory(locator)) {
+ ClientSession session = sf.createSession(false, false);
+
+ SimpleString queueAddr = SimpleString.of("FOO");
+ session.createQueue(QueueConfiguration.of(queueAddr));
+
+ int size = 1024 * 50;
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setPageFullMessagePolicy(PageFullMessagePolicy.FAIL);
+ addressSettings.setPageSizeBytes(size);
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 10));
+ addressSettings.setMaxSizeBytes(size);
+
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ int totalMessages = 30;
+ int messageSize = 1024 * 10;
+ sendMessageBatch(totalMessages, messageSize, session, queueAddr);
+
+ Queue queue = server.locateQueue(queueAddr);
+
+ // Give time Queue.deliverAsync to deliver messages
+ assertTrue(waitForMessages(queue, totalMessages, 10000));
+
+ PagingStore queuePagingStore = queue.getPagingStore();
+ assertTrue(queuePagingStore != null && queuePagingStore.isPaging());
+
+ long existingPages = queuePagingStore.getNumberOfPages();
+ assertTrue(existingPages > 4);
+
+ // restart the server and the invalid settings are not applied too.
+ server.stop(true);
+ waitForServerToStop(server);
+
+ addressSettings.setPageLimitBytes(Long.valueOf(size * 4));
+ server.getAddressSettingsRepository().addMatch("FOO", addressSettings);
+
+ server.start();
+ waitForServerToStart(server);
+
+ queue = server.locateQueue(queueAddr);
+ queuePagingStore = queue.getPagingStore();
+
+ // check settings not applied
+ assertEquals(0, queuePagingStore.getPageSizeBytes());
+ assertNull(queuePagingStore.getPageFullMessagePolicy());
+ assertNull(queuePagingStore.getPageLimitBytes());
+ assertEquals(0, queuePagingStore.getMaxSize());
Review Comment:
I would expect the broker won't work at all as the pageStore didn't get any setting applied during restart. User has to fix the address settings and try restart again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1718128177
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -232,7 +233,7 @@ private void configureSizeMetric() {
* @param addressSettings
*/
@Override
- public void applySetting(final AddressSettings addressSettings) {
+ public void applySetting(final AddressSettings addressSettings, final boolean firstTime) {
Review Comment:
Since it is really an implementation detail that the constructor calls the applySetting method, I think this interface method should probably just go back to the way it was, and a private internal second method be created with the firstTime boolean that both this methods implementation and the constructor can then call. That way the interface and none of the tests etc need to be updated to support calling with false (and they shouldnt be able to call with true)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1704673365
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +229,34 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private void validateNewSettings(final AddressSettings addressSettings) throws ActiveMQException {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes.longValue() < 0) {
+ newPageLimitBytes = null;
+ }
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+ if (newPageLimitBytes != null && newPageSize > 0) {
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+ if (this.numberOfPages > newEstimatedMaxPages) {
+ throw ActiveMQMessageBundle.BUNDLE.invalidPageSettings(this.storeName.toString(), addressSettings, "estimated max pages " + newEstimatedMaxPages + " is less than current number of pages " + this.numberOfPages);
Review Comment:
You should log.warn here, from the Logger, and return false,
on the applySettings you should then if (!valideSettings) return;
because here you are throwing an exception just to log.warn... It will be more complicated than it needs to be here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-4973 pageSizeBytes/pageLimitBytes combination can cause Address full [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5128:
URL: https://github.com/apache/activemq-artemis/pull/5128#discussion_r1708991778
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -228,11 +228,58 @@ private void configureSizeMetric() {
size.setMax(maxSize, maxSize, maxMessages, maxMessages);
}
+ private boolean validateNewSettings(final AddressSettings addressSettings) {
+
+ Long newPageLimitBytes = addressSettings.getPageLimitBytes();
+
+ if (newPageLimitBytes != null && newPageLimitBytes < 0) {
+ newPageLimitBytes = null;
+ }
+
+ int newPageSize = storageManager.getAllowedPageSize(addressSettings.getPageSizeBytes());
+
+ if (newPageLimitBytes != null && newPageSize > 0) {
+
+ if (newPageSize > newPageLimitBytes) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "pageLimitBytes is smaller than pageSize. It should allow at least one page file");
+ return false;
+ }
+
+ long newEstimatedMaxPages = newPageLimitBytes / newPageSize;
+
+ long existingNumberOfPages;
+ if (!running) {
+ try {
+ checkFileFactory();
+ List<String> files = this.fileFactory.listFiles("page");
+ existingNumberOfPages = files.size();
+ } catch (Exception e) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "failed to get existing page files with exception: " + e);
+ return false;
+ }
+ } else {
+ existingNumberOfPages = this.numberOfPages;
+ }
+
+ if (existingNumberOfPages > newEstimatedMaxPages) {
+ ActiveMQServerLogger.LOGGER.pageSettingsFailedApply(storeName, addressSettings, "estimated max pages " + newEstimatedMaxPages + " (pageLimitBytes / pageSize) is less than current number of pages " + existingNumberOfPages);
+ return false;
+ }
Review Comment:
The diff makes it seem like you are adding that 'you cant set a limit lower than what is in use' behaviour here in this commit? You are saying that the existing code already does that, where is that happening? Also feels awkward to have it happening in 2 places without apparent relation.
I was actually more interested in the appearance of a 'shut broker down, decrease limit, start up...fail new validation, get no settings applied at all' issue that I commented on around the test, but runtime behaviour is also a key issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact