You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@activemq.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2023/09/29 11:51:00 UTC
[jira] [Work logged] (ARTEMIS-4447) Add paging prefetch parameters into address settings
[ https://issues.apache.org/jira/browse/ARTEMIS-4447?focusedWorklogId=882612&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-882612 ]
ASF GitHub Bot logged work on ARTEMIS-4447:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 29/Sep/23 11:50
Start Date: 29/Sep/23 11:50
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #4637:
URL: https://github.com/apache/activemq-artemis/pull/4637#discussion_r1341238040
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+ private ActiveMQServer server;
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = newActiveMQServer();
+
+ server.start();
+ waitForServerToStart(server);
+ }
+
+ private ActiveMQServer newActiveMQServer() throws Exception {
+ ActiveMQServer server = createServer(true, isNetty());
+
+ AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeMessages(10);
+
+ server.getAddressSettingsRepository().addMatch("#", defaultSetting);
+
+ return server;
+ }
+
+ @Test
+ public void testPageFilter() throws Exception {
+
+ ConnectionFactory connectionFactory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
+
+ int numberOfMessages = 3 * 24;
+ String filter = "partial=true";
+ String negfilter = "partial=false";
+
+ server.addAddressInfo(new AddressInfo(getName()).addRoutingType(RoutingType.MULTICAST));
+
+
+
+ try (Connection connection = connectionFactory.createConnection();
+ Connection otherConnection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ session.createSharedDurableConsumer(topic, "wholeThing", negfilter);
+ MessageConsumer partialThing = session.createSharedDurableConsumer(topic, "partialThing", filter);
+
+ final String oneThousandCharacters;
+ {
+ StringBuilder builder = new StringBuilder(1000);
+ for (int i = 0; i < 1000; i++) {
+ builder.append('a');
+ }
+ oneThousandCharacters = builder.toString();
+ }
+
+ MessageProducer producer = session.createProducer(topic);
+
+
+ Session txsession = otherConnection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer txProducer = txsession.createProducer(topic);
+ for (int i = 0; i < 100; i++) {
+ Message message = session.createTextMessage(oneThousandCharacters + i);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("partial", i % 2 == 0);
+ producer.send(message);
+ if (i % 4 == 0) {
+ Message txMessage = session.createTextMessage("ignored");
+ txMessage.setBooleanProperty("partial", false);
+ txMessage.setBooleanProperty("ignored", true);
+ txProducer.send(txMessage);
+ }
+ }
+
+ connection.start();
+
+ int receivedI = 0;
+ do {
+ Message message = partialThing.receive(1000);
+ Assert.assertNotNull(message);
+ receivedI = message.getIntProperty("i");
+ }
+ while (receivedI < 10);
+
+ server.stop();
+ }
+
+ server.start();
+
+ try (Connection connection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ connection.start();
+ MessageConsumer consumerWhole = session.createSharedDurableConsumer(topic, "partialThing", filter);
+ int receivedMessages = 0;
+ for (int i = 0; i < numberOfMessages; i++) {
+ Message message = consumerWhole.receive(1000);
+ if (message == null) {
+ break;
+ }
+ Assert.assertNotNull(message);
+ receivedMessages++;
+ }
+ logger.debug("Received {} messages", receivedMessages);
Review Comment:
We aren't even asserting how many messages were actually received?
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+ private ActiveMQServer server;
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = newActiveMQServer();
+
+ server.start();
+ waitForServerToStart(server);
+ }
+
+ private ActiveMQServer newActiveMQServer() throws Exception {
+ ActiveMQServer server = createServer(true, isNetty());
+
+ AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeMessages(10);
+
+ server.getAddressSettingsRepository().addMatch("#", defaultSetting);
+
+ return server;
+ }
+
+ @Test
+ public void testPageFilter() throws Exception {
+
+ ConnectionFactory connectionFactory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
+
+ int numberOfMessages = 3 * 24;
+ String filter = "partial=true";
+ String negfilter = "partial=false";
+
+ server.addAddressInfo(new AddressInfo(getName()).addRoutingType(RoutingType.MULTICAST));
+
+
+
+ try (Connection connection = connectionFactory.createConnection();
+ Connection otherConnection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ session.createSharedDurableConsumer(topic, "wholeThing", negfilter);
+ MessageConsumer partialThing = session.createSharedDurableConsumer(topic, "partialThing", filter);
+
+ final String oneThousandCharacters;
+ {
+ StringBuilder builder = new StringBuilder(1000);
+ for (int i = 0; i < 1000; i++) {
+ builder.append('a');
+ }
+ oneThousandCharacters = builder.toString();
+ }
+
+ MessageProducer producer = session.createProducer(topic);
+
+
+ Session txsession = otherConnection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer txProducer = txsession.createProducer(topic);
+ for (int i = 0; i < 100; i++) {
+ Message message = session.createTextMessage(oneThousandCharacters + i);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("partial", i % 2 == 0);
+ producer.send(message);
+ if (i % 4 == 0) {
+ Message txMessage = session.createTextMessage("ignored");
+ txMessage.setBooleanProperty("partial", false);
+ txMessage.setBooleanProperty("ignored", true);
+ txProducer.send(txMessage);
+ }
+ }
+
+ connection.start();
+
+ int receivedI = 0;
+ do {
+ Message message = partialThing.receive(1000);
+ Assert.assertNotNull(message);
+ receivedI = message.getIntProperty("i");
+ }
+ while (receivedI < 10);
+
+ server.stop();
+ }
+
+ server.start();
+
+ try (Connection connection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ connection.start();
+ MessageConsumer consumerWhole = session.createSharedDurableConsumer(topic, "partialThing", filter);
Review Comment:
consumerWhole using the "partialThing" sub seems confusing (as opposed to e.g the "wholeThing" sub that was created but not used...and which itself doesnt get everything anyway). Perhaps some clearer naming?
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+ private ActiveMQServer server;
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = newActiveMQServer();
+
+ server.start();
+ waitForServerToStart(server);
+ }
+
+ private ActiveMQServer newActiveMQServer() throws Exception {
+ ActiveMQServer server = createServer(true, isNetty());
+
+ AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeMessages(10);
+
+ server.getAddressSettingsRepository().addMatch("#", defaultSetting);
+
+ return server;
+ }
+
+ @Test
+ public void testPageFilter() throws Exception {
+
+ ConnectionFactory connectionFactory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
+
+ int numberOfMessages = 3 * 24;
+ String filter = "partial=true";
+ String negfilter = "partial=false";
+
+ server.addAddressInfo(new AddressInfo(getName()).addRoutingType(RoutingType.MULTICAST));
+
+
+
+ try (Connection connection = connectionFactory.createConnection();
+ Connection otherConnection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ session.createSharedDurableConsumer(topic, "wholeThing", negfilter);
+ MessageConsumer partialThing = session.createSharedDurableConsumer(topic, "partialThing", filter);
+
+ final String oneThousandCharacters;
+ {
+ StringBuilder builder = new StringBuilder(1000);
+ for (int i = 0; i < 1000; i++) {
+ builder.append('a');
+ }
+ oneThousandCharacters = builder.toString();
+ }
+
+ MessageProducer producer = session.createProducer(topic);
+
+
+ Session txsession = otherConnection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer txProducer = txsession.createProducer(topic);
+ for (int i = 0; i < 100; i++) {
Review Comment:
hard coding 100 here when numberOfMessages is defined to the rather random looking 3 * 24 = 72 at the start just seems confusing.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
Review Comment:
looks unused?
##########
docs/user-manual/paging.adoc:
##########
@@ -99,15 +99,21 @@ If the value is `BLOCK` then client message producers will block when they try a
| `PAGE`
| `max-read-page-messages`
-| how many message can be read from paging into the Queue whenever more messages are needed.
-The system wtill stop reading if `max-read-page-bytes hits the limit first.
-| -1
+| how many paged messages can be in the memory per queue, including delivering and ready to deliver messages.
+| -1 (disabled)
| `max-read-page-bytes`
-| how much memory the messages read from paging can take on the Queue whenever more messages are needed.
-The system will stop reading if `max-read-page-messages` hits the limit first.
+| how much memory paged messages can occupy in memory per queue, including delivering and ready to deliver messages. The system will not bring any more messages from paging until this limit is cleared (Messages Acked).
| 2 * page-size-bytes
+|prefetch-page-bytes
+|How many bytes should be prefetch from paging into each queue.
+|if not defined, `max-read-page-bytes`
+
+|prefetch-page-messages
+|How many messages should be prefetch from paging into each queue.
+|if not defined, `max-read-page-messages`
Review Comment:
Flipping the order of these two (or the ones above) so the order is consistently covering _bytes_ then covering _messages_, or vice versa, would make things more readable.
It isn't really clear from this that if you set either/both of _these_ options, they wont necessarily do what is suggested/expected unless max-read-page-bytes and/or max-read-page-messages are _also_ set to something that is coordinated as appropriate to these values. E.g the max-read-page-bytes would also need set to be changed to be >= prefetch-page-bytes if you ever set prefetch-page-bytes to >= 2*page-size-bytes default for max-read-page-bytes to actually get the desired behaviour. If max-read-page-bytes and max-read-page-messages were both set to -1, these values wont even be considered, which isnt clear.
The "should be prefetch" wording doesnt work and needs tweaked. Maybe "the broker should prefetch"?
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+ private ActiveMQServer server;
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ server = newActiveMQServer();
+
+ server.start();
+ waitForServerToStart(server);
+ }
+
+ private ActiveMQServer newActiveMQServer() throws Exception {
+ ActiveMQServer server = createServer(true, isNetty());
+
+ AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(10 * 1024).setMaxSizeMessages(10);
+
+ server.getAddressSettingsRepository().addMatch("#", defaultSetting);
+
+ return server;
+ }
+
+ @Test
+ public void testPageFilter() throws Exception {
+
+ ConnectionFactory connectionFactory = CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616");
+
+ int numberOfMessages = 3 * 24;
+ String filter = "partial=true";
+ String negfilter = "partial=false";
+
+ server.addAddressInfo(new AddressInfo(getName()).addRoutingType(RoutingType.MULTICAST));
+
+
+
+ try (Connection connection = connectionFactory.createConnection();
+ Connection otherConnection = connectionFactory.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic topic = session.createTopic(getName());
+ session.createSharedDurableConsumer(topic, "wholeThing", negfilter);
+ MessageConsumer partialThing = session.createSharedDurableConsumer(topic, "partialThing", filter);
+
+ final String oneThousandCharacters;
+ {
+ StringBuilder builder = new StringBuilder(1000);
+ for (int i = 0; i < 1000; i++) {
+ builder.append('a');
+ }
+ oneThousandCharacters = builder.toString();
+ }
+
+ MessageProducer producer = session.createProducer(topic);
+
+
+ Session txsession = otherConnection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer txProducer = txsession.createProducer(topic);
+ for (int i = 0; i < 100; i++) {
+ Message message = session.createTextMessage(oneThousandCharacters + i);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("partial", i % 2 == 0);
+ producer.send(message);
+ if (i % 4 == 0) {
+ Message txMessage = session.createTextMessage("ignored");
+ txMessage.setBooleanProperty("partial", false);
+ txMessage.setBooleanProperty("ignored", true);
+ txProducer.send(txMessage);
+ }
+ }
+
+ connection.start();
+
+ int receivedI = 0;
+ do {
+ Message message = partialThing.receive(1000);
+ Assert.assertNotNull(message);
+ receivedI = message.getIntProperty("i");
+ }
+ while (receivedI < 10);
Review Comment:
This doesnt even check what it receives?
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PageFilterTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.PrintDataUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageFilterTest extends ActiveMQTestBase {
Review Comment:
This class doesnt appear to even use the new config it seems. Is it actually related?
It doesnt really assert much of anything either. Is it a dev 'utility/debug test' ?
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -3946,6 +3954,14 @@
</xsd:annotation>
</xsd:element>
+ <xsd:element name="prefetch-page-bytes" type="xsd:string" default="20M" maxOccurs="1" minOccurs="0">
+ <xsd:annotation>
+ <xsd:documentation>
+ How many bytes we are reading from paging and keeping ready in queues ready to deliver to consumers.
Review Comment:
The default could be spelled out clearer? It isnt really just 20M, but whatever max-read-page-bytes is (the default for which is also not just 20M but 2x page size...which similarly isnt clear). Similar to other comment, interaction of the two things isnt that clear either.
Similar for the messages config.
##########
artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java:
##########
@@ -1623,6 +1623,32 @@ public void testAddressSettingsPageLimitInvalidConfiguration5() throws Throwable
Assert.assertEquals(null, storeImpl.getPageFullMessagePolicy());
}
+ @Test
+ public void testPagePrefetch() throws Throwable {
+ ConfigurationImpl configuration = new ConfigurationImpl();
+
+ Properties properties = new Properties();
+
+ String randomString = RandomUtil.randomString();
+
+ properties.put("addressSettings.#.expiryAddress", randomString);
+ properties.put("addressSettings.#.prefetchPageMessages", "333"); // this should make the PagingStore to parse it as null
+ properties.put("addressSettings.#.prefetchPageBytes", "777"); // this should make the PagingStore to parse it as null
Review Comment:
Comments dont seem correct, its explicitly checking they aren't null?
##########
artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java:
##########
@@ -1623,6 +1623,32 @@ public void testAddressSettingsPageLimitInvalidConfiguration5() throws Throwable
Assert.assertEquals(null, storeImpl.getPageFullMessagePolicy());
}
+ @Test
+ public void testPagePrefetch() throws Throwable {
+ ConfigurationImpl configuration = new ConfigurationImpl();
+
+ Properties properties = new Properties();
+
+ String randomString = RandomUtil.randomString();
+
+ properties.put("addressSettings.#.expiryAddress", randomString);
+ properties.put("addressSettings.#.prefetchPageMessages", "333"); // this should make the PagingStore to parse it as null
+ properties.put("addressSettings.#.prefetchPageBytes", "777"); // this should make the PagingStore to parse it as null
+
+ configuration.parsePrefixedProperties(properties, null);
+
+ Assert.assertEquals(1, configuration.getAddressSettings().size());
+ Assert.assertEquals(SimpleString.toSimpleString(randomString), configuration.getAddressSettings().get("#").getExpiryAddress());
+ Assert.assertEquals(333, configuration.getAddressSettings().get("#").getPrefetchPageMessages());
+ Assert.assertEquals(777, configuration.getAddressSettings().get("#").getPrefetchPageBytes());
+
+ PagingStore storeImpl = new PagingStoreImpl(new SimpleString("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), new SimpleString("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+
+ Assert.assertEquals(333, storeImpl.getPrefetchPageMessages());
+ Assert.assertEquals(777, storeImpl.getPrefetchPageBytes());
+ }
+
+
Review Comment:
A test [somewhere] checking the values are are as expected when _not_ explicitly set would seem appropriate.
Issue Time Tracking
-------------------
Worklog Id: (was: 882612)
Time Spent: 20m (was: 10m)
> Add paging prefetch parameters into address settings
> ----------------------------------------------------
>
> Key: ARTEMIS-4447
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4447
> Project: ActiveMQ Artemis
> Issue Type: Improvement
> Affects Versions: 2.31.0
> Reporter: Clebert Suconic
> Priority: Major
> Fix For: 2.32.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> we should include optimal prefetch values for paging. Right now the system will always read up to max-read-page-messages and max-read-page-bytes. We should have a separate configuration attribute for the prefetch value.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)