From 823d1df17b0db6b5c570a3bc7c691e0825486614 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 26 Mar 2026 19:24:39 -0400 Subject: [PATCH] ARTEMIS-5573 and ARTEMIS-5975 Improve AMQP Size estimation - making it immutable to avoid races after updates like we had in the past - avoiding scanning after reloading by persisting extra data on storage --- .../activemq/artemis/util/ServerUtil.java | 42 +- .../core/persistence/PersisterIDs.java | 6 +- .../activemq/artemis/api/core/Message.java | 8 - .../api/core/management/SimpleManagement.java | 12 + .../amqp/broker/AMQPLargeMessage.java | 31 +- .../broker/AMQPLargeMessagePersister.java | 3 - .../broker/AMQPLargeMessagePersisterV2.java | 97 +++++ .../protocol/amqp/broker/AMQPMessage.java | 161 ++++---- .../amqp/broker/AMQPMessagePersister.java | 11 + .../amqp/broker/AMQPMessagePersisterV2.java | 3 + .../amqp/broker/AMQPMessagePersisterV3.java | 7 +- .../amqp/broker/AMQPMessagePersisterV4.java | 109 ++++++ .../amqp/broker/AMQPStandardMessage.java | 101 +---- .../broker/ProtonProtocolManagerFactory.java | 3 +- .../protocol/amqp/broker/AMQPMessageTest.java | 36 +- .../amqp/converter/TestConversions.java | 7 +- .../artemis/core/server/impl/QueueImpl.java | 4 +- .../artemis/utils/RealServerTestBase.java | 13 +- .../resources/meshTest/sendMessages.groovy | 12 +- .../JournalCompatibilityTest.java | 44 ++- .../MultiVersionReplicaTest.java | 5 +- .../amqp/AMQPReloadFromPersistenceTest.java | 297 ++++++++++++++ .../amqp/AmqpClientTestSupport.java | 9 +- .../amqp/connect/AckManagerTest.java | 3 +- .../amqp/journal/AmqpJournalLoadingTest.java | 29 +- .../integration/routing/ElasticQueueTest.java | 2 +- .../tests/soak/paging/AMQPGlobalMaxTest.java | 363 ++++++++++++++++++ .../unit/core/paging/impl/AmqpPageTest.java | 14 +- 28 files changed, 1192 insertions(+), 240 deletions(-) create mode 100644 artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java create mode 100644 artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV4.java create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPReloadFromPersistenceTest.java create mode 100644 tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java index 10bd19865ce..b1d31a90ca3 100644 --- a/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java +++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/util/ServerUtil.java @@ -24,6 +24,7 @@ import java.io.InputStreamReader; import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; import org.apache.activemq.artemis.api.core.TransportConfiguration; import org.apache.activemq.artemis.api.core.client.ClientSession; @@ -51,7 +52,11 @@ public static Process startServer(String artemisInstance, String serverName, int } public static Process startServer(String artemisInstance, String serverName, int id, int timeout, File brokerProperties) throws Exception { - final Process process = internalStartServer(artemisInstance, serverName, brokerProperties); + return startServer(artemisInstance, serverName, id, timeout, brokerProperties, null); + } + + public static Process startServer(String artemisInstance, String serverName, int id, int timeout, File brokerProperties, Consumer logCallback) throws Exception { + final Process process = internalStartServer(artemisInstance, serverName, brokerProperties, logCallback); // wait for start if (timeout > 0) { @@ -66,7 +71,11 @@ public static Process startServer(String artemisInstance, String serverName, Str } public static Process startServer(String artemisInstance, String serverName, String uri, int timeout, File propertiesFile) throws Exception { - final Process process = internalStartServer(artemisInstance, serverName, propertiesFile); + return startServer(artemisInstance, serverName, uri, timeout, propertiesFile, null); + } + + public static Process startServer(String artemisInstance, String serverName, String uri, int timeout, File propertiesFile, Consumer logCallback) throws Exception { + final Process process = internalStartServer(artemisInstance, serverName, propertiesFile, logCallback); // wait for start if (timeout != 0) { @@ -78,20 +87,30 @@ public static Process startServer(String artemisInstance, String serverName, Str private static Process internalStartServer(String artemisInstance, String serverName) throws IOException, ClassNotFoundException { - return internalStartServer(artemisInstance, serverName, null); + return internalStartServer(artemisInstance, serverName, null, null); } private static Process internalStartServer(String artemisInstance, String serverName, File propertiesFile) throws IOException, ClassNotFoundException { + return internalStartServer(artemisInstance, serverName, propertiesFile, null); + } + private static Process internalStartServer(String artemisInstance, + String serverName, + File propertiesFile, + Consumer logCallback) throws IOException, ClassNotFoundException { if (propertiesFile != null) { - return execute(artemisInstance, serverName, "run", "--properties", propertiesFile.getAbsolutePath()); + return execute(artemisInstance, serverName, logCallback, "run", "--properties", propertiesFile.getAbsolutePath()); } else { - return execute(artemisInstance, serverName, "run"); + return execute(artemisInstance, serverName, logCallback, "run"); } } public static Process execute(String artemisInstance, String jobName, String...args) throws IOException, ClassNotFoundException { + return execute(artemisInstance, jobName, null, args); + } + + public static Process execute(String artemisInstance, String jobName, Consumer logCallback, String...args) throws IOException, ClassNotFoundException { try { boolean IS_WINDOWS = System.getProperty("os.name").toLowerCase().trim().startsWith("win"); @@ -117,11 +136,11 @@ public static Process execute(String artemisInstance, String jobName, String...a final Process process = builder.start(); Runtime.getRuntime().addShutdownHook(new Thread(() -> process.destroy())); - ProcessLogger outputLogger = new ProcessLogger(true, process.getInputStream(), jobName, false); + ProcessLogger outputLogger = new ProcessLogger(logCallback == null, process.getInputStream(), jobName, false, logCallback); outputLogger.start(); // Adding a reader to System.err, so the VM won't hang on a System.err.println - ProcessLogger errorLogger = new ProcessLogger(true, process.getErrorStream(), jobName, true); + ProcessLogger errorLogger = new ProcessLogger(logCallback == null, process.getErrorStream(), jobName, true, logCallback); errorLogger.start(); return process; } catch (IOException e) { @@ -215,14 +234,18 @@ static class ProcessLogger extends Thread { private final boolean sendToErr; + private final Consumer logCallback; + ProcessLogger(final boolean print, final InputStream is, final String logName, - final boolean sendToErr) throws ClassNotFoundException { + final boolean sendToErr, + final Consumer logCallback) throws ClassNotFoundException { this.is = is; this.print = print; this.logName = logName; this.sendToErr = sendToErr; + this.logCallback = logCallback; setDaemon(false); } @@ -240,6 +263,9 @@ public void run() { System.out.println(logName + "-out:" + line); } } + if (logCallback != null) { + logCallback.accept((sendToErr ? logName + "-err:" : logName + "-out:") + line); + } } } catch (IOException e) { // ok, stream closed diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/persistence/PersisterIDs.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/persistence/PersisterIDs.java index 483f0ae670b..b34fd2ae27a 100644 --- a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/persistence/PersisterIDs.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/persistence/PersisterIDs.java @@ -23,7 +23,7 @@ */ public class PersisterIDs { - public static final int MAX_PERSISTERS = 5; + public static final int MAX_PERSISTERS = 7; public static final byte CoreLargeMessagePersister_ID = (byte)0; @@ -37,4 +37,8 @@ public class PersisterIDs { public static final byte AMQPMessagePersisterV3_ID = (byte)5; + public static final byte AMQPMessagePersisterV4_ID = (byte)6; + + public static final byte AMQPLargeMessagePersisterV2_ID = (byte)7; + } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java index 0115ba9968b..65ac30dd263 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java @@ -876,14 +876,6 @@ default CompositeData toCompositeData(int fieldsLimit, int deliveryCount) throws int getMemoryEstimate(); - /** - * The first estimate that's been calculated without any updates. - */ - default int getOriginalEstimate() { - // For Core Protocol we always use the same estimate - return getMemoryEstimate(); - } - /** * This is the size of the message when persisted on disk which is used for metrics tracking Note that even if the * message itself is not persisted on disk (ie non-durable) this value is still used for metrics tracking If a normal diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java index 65e066d8a66..a00bec655f7 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java @@ -109,6 +109,18 @@ public void rebuildPageCounters() throws Exception { simpleManagementVoid("broker", "rebuildPageCounters"); } + public long getAddressSize(String address) throws Exception { + return simpleManagementLong(ResourceNames.ADDRESS + address, "getAddressSize"); + } + + public long getMessageCountOnAddress(String address) throws Exception { + return simpleManagementLong(ResourceNames.ADDRESS + address, "getMessageCount"); + } + + public int removeMessagesOnQueue(String queue, String filter) throws Exception { + return simpleManagementInt(ResourceNames.QUEUE + queue, "removeMessages", filter); + } + /** * Simple helper for management returning a string. */ diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java index 99dbe2d8adb..41ad13f819f 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java @@ -40,6 +40,7 @@ import org.apache.activemq.artemis.protocol.amqp.util.NettyReadable; import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable; import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode; +import org.apache.activemq.artemis.utils.DataConstants; import org.apache.activemq.artemis.utils.collections.TypedProperties; import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations; @@ -80,6 +81,8 @@ public Message getMessage() { private boolean reencoded = false; + private int applicationPropertiesSize; + /** * AMQPLargeMessagePersister will save the buffer here. */ @@ -264,7 +267,9 @@ protected void readSavedEncoding(ByteBuf buf) { applicationPropertiesPosition = buf.readInt(); remainingBodyPosition = buf.readInt(); + int applicationPropertiesInitialPosition = buf.readerIndex(); applicationProperties = (ApplicationProperties)TLSEncode.getDecoder().readObject(); + this.applicationPropertiesSize = buf.readerIndex() - applicationPropertiesInitialPosition; if (properties != null && properties.getAbsoluteExpiryTime() != null && properties.getAbsoluteExpiryTime().getTime() > 0) { if (!expirationReload) { @@ -400,6 +405,12 @@ protected void parseLargeMessage(byte[] data, boolean initialHeader) { } } + @Override + protected synchronized void resetMessageData() { + super.resetMessageData(); + applicationPropertiesSize = 0; + } + private void genericParseLargeMessage() { try { parsingBuffer.position(0); @@ -412,6 +423,16 @@ private void genericParseLargeMessage() { } } + @Override + protected ApplicationProperties readApplicationProperties(ReadableBuffer data, int position) { + ApplicationProperties localAP = super.readApplicationProperties(data, position); + if (localAP != null) { + this.applicationPropertiesSize = data.position() - position; + this.applicationPropertiesCount = localAP.getValue().size(); + } + return localAP; + } + protected void parseLargeMessage(ReadableBuffer data) { MessageDataScanningStatus status = getDataScanningStatus(); if (status == MessageDataScanningStatus.NOT_SCANNED) { @@ -596,16 +617,16 @@ public long getWholeMessageSize() { return largeBody.getBodySize(); } catch (Exception e) { logger.warn(e.getMessage()); - return -1; + return VALUE_NOT_PRESENT; } } @Override public synchronized int getMemoryEstimate() { - if (memoryEstimate == -1) { - memoryEstimate = memoryOffset * 2 + (extraProperties != null ? extraProperties.getEncodeSize() : 0); - originalEstimate = memoryEstimate; + if (memoryEstimate == VALUE_NOT_PRESENT) { + // This estimation was tested and validated through AMQPGlobalMaxTest on soak-tests + memoryEstimate = MINIMUM_ESTIMATE + (extraProperties != null ? extraProperties.getEncodeSize() : 0) + applicationPropertiesSize * 2 + applicationPropertiesCount * DataConstants.SIZE_INT; } return memoryEstimate; } @@ -637,7 +658,7 @@ public long getPersistentSize() { @Override public Persister getPersister() { - return AMQPLargeMessagePersister.getInstance(); + return AMQPLargeMessagePersisterV2.getInstance(); } @Override diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java index 7fb4b6d41fb..6ffa4707383 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java @@ -68,9 +68,6 @@ public int getEncodeSize(Message record) { } } - /** - * Sub classes must add the first short as the protocol-id - */ @Override public void encode(ActiveMQBuffer buffer, Message record) { super.encode(buffer, record); diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java new file mode 100644 index 00000000000..d2fd37a26b6 --- /dev/null +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java @@ -0,0 +1,97 @@ +/* + * 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.protocol.amqp.broker; + +import org.apache.activemq.artemis.api.core.ActiveMQBuffer; +import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools; +import org.apache.activemq.artemis.utils.DataConstants; + +import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPLargeMessagePersisterV2_ID; + +public class AMQPLargeMessagePersisterV2 extends AMQPLargeMessagePersister { + + public static final byte ID = AMQPLargeMessagePersisterV2_ID; + + public static AMQPLargeMessagePersisterV2 theInstance; + + public static AMQPLargeMessagePersisterV2 getInstance() { + if (theInstance == null) { + theInstance = new AMQPLargeMessagePersisterV2(); + } + return theInstance; + } + + @Override + public byte getID() { + return ID; + } + + public AMQPLargeMessagePersisterV2() { + super(); + } + + + protected static final int PERSISTER_SIZE = DataConstants.SIZE_INT + // memory estimate + DataConstants.SIZE_BYTE + // message priority + DataConstants.SIZE_BOOLEAN; // durable + + @Override + public int getEncodeSize(Message record) { + return super.getEncodeSize(record) + + DataConstants.SIZE_INT + // size delimiter for future use to keep compatibility better + PERSISTER_SIZE; + } + + @Override + public void encode(ActiveMQBuffer buffer, Message record) { + super.encode(buffer, record); + + AMQPLargeMessage msgEncode = (AMQPLargeMessage) record; + writeSizeDelimiter(buffer); + buffer.writeInt(msgEncode.getMemoryEstimate()); + buffer.writeByte(msgEncode.getPriority()); + buffer.writeBoolean(msgEncode.isDurable()); + } + + protected void writeSizeDelimiter(ActiveMQBuffer buffer) { + buffer.writeInt(PERSISTER_SIZE); // how many bytes this persister is using + } + + @Override + public Message decode(ActiveMQBuffer buffer, Message record, CoreMessageObjectPools pools) { + AMQPLargeMessage message = (AMQPLargeMessage) super.decode(buffer, record, pools); + + + int sizePersister = buffer.readInt(); + int lastPosition = buffer.readerIndex() + sizePersister; + + { + message.setMemoryEstimate(buffer.readInt()); + message.setPriority(buffer.readByte()); + message.reloadSetDurable(buffer.readBoolean()); + + assert buffer.readerIndex() <= lastPosition; + } + + // if a future version of this persister wrote more bytes than what we expected now, this will make sure we skip them. + buffer.readerIndex(lastPosition); + + return message; + } + +} diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java index e078fdb2363..397c502117b 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java @@ -44,7 +44,6 @@ import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.core.message.openmbean.CompositeDataConstants; import org.apache.activemq.artemis.core.message.openmbean.MessageOpenTypeFactory; -import org.apache.activemq.artemis.core.paging.PagingStore; import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools; import org.apache.activemq.artemis.core.persistence.Persister; import org.apache.activemq.artemis.core.server.MessageReference; @@ -57,6 +56,7 @@ import org.apache.activemq.artemis.reader.MessageUtil; import org.apache.activemq.artemis.utils.ByteUtil; import org.apache.activemq.artemis.utils.collections.TypedProperties; +import org.apache.qpid.proton.ProtonException; import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.UnsignedByte; @@ -83,6 +83,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.qpid.proton.codec.EncodingCodes; + + import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport.getCharsetForTextualContent; /** @@ -119,6 +122,12 @@ */ public abstract class AMQPMessage extends RefCountMessage implements org.apache.activemq.artemis.api.core.Message { + // The minimum size an AMQP message uses. + // This is an estimate, and it's based on the following test: + // By running AMQPGlobalMaxTest::testSendUntilOME, you look at the initial memory used by the broker without any messages. + // By the time you get the OME, you can do some basic calculations on how much each message uses and get an AVG. + public static final int MINIMUM_ESTIMATE = 1300; + private static final SimpleString ANNOTATION_AREA_PREFIX = SimpleString.of("m."); protected static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); @@ -146,7 +155,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache. * developing purposes. */ public enum MessageDataScanningStatus { - NOT_SCANNED(0), RELOAD_PERSISTENCE(1), SCANNED(2); + NOT_SCANNED(0), SCANNED(1); private static final MessageDataScanningStatus[] STATES; @@ -198,18 +207,23 @@ private static void checkCode(int code) { protected int messageAnnotationsPosition = VALUE_NOT_PRESENT; protected int propertiesPosition = VALUE_NOT_PRESENT; protected int applicationPropertiesPosition = VALUE_NOT_PRESENT; + protected int applicationPropertiesCount; protected int remainingBodyPosition = VALUE_NOT_PRESENT; // Message level meta data protected final long messageFormat; protected long messageID; protected SimpleString address; - protected volatile int memoryEstimate = -1; - protected volatile int originalEstimate = -1; + protected volatile int memoryEstimate = VALUE_NOT_PRESENT; protected long expiration; protected boolean expirationReload = false; - protected long scheduledTime = -1; - protected byte priority = DEFAULT_MESSAGE_PRIORITY; + protected long scheduledTime = VALUE_NOT_PRESENT; + protected byte priority = VALUE_NOT_PRESENT; + + /** + * 0 = false, 1 = true, or VALUE_NOT_PRESENT + */ + protected byte durable = VALUE_NOT_PRESENT; protected boolean isPaged; protected volatile boolean routed = false; @@ -270,6 +284,11 @@ protected AMQPMessage(AMQPMessage copy) { this.messageDataScanned = copy.messageDataScanned; } + void setMemoryEstimate(int memoryEstimate) { + this.memoryEstimate = memoryEstimate; + } + + private static MessageAnnotations copyAnnotations(MessageAnnotations messageAnnotations) { if (messageAnnotations == null) { return null; @@ -546,36 +565,14 @@ protected ApplicationProperties lazyDecodeApplicationProperties() { // need to synchronize access to lazyDecodeApplicationProperties to avoid clashes with getMemoryEstimate protected synchronized ApplicationProperties lazyDecodeApplicationProperties(ReadableBuffer data) { if (applicationProperties == null && applicationPropertiesPosition != VALUE_NOT_PRESENT) { - applicationProperties = scanForMessageSection(data, applicationPropertiesPosition, ApplicationProperties.class); - if (owner != null && memoryEstimate != -1) { - // the memory has already been tracked and needs to be updated to reflect the new decoding - int addition = unmarshalledApplicationPropertiesMemoryEstimateFromData(data); - - // it is difficult to track the updates for paged messages - // for that reason we won't do it if paged - // we also only do the update if the message was previously routed - // so if a debug method or an interceptor changed the size before routing we would get a different size - if (!isPaged && routed) { - ((PagingStore) owner).addSize(addition, false); - final int updatedEstimate = memoryEstimate + addition; - memoryEstimate = updatedEstimate; - } - } + this.applicationProperties = readApplicationProperties(data, applicationPropertiesPosition); } return applicationProperties; } - protected int unmarshalledApplicationPropertiesMemoryEstimateFromData(ReadableBuffer data) { - if (applicationProperties != null) { - // they have been unmarshalled, estimate memory usage based on their encoded size - if (remainingBodyPosition != VALUE_NOT_PRESENT) { - return remainingBodyPosition - applicationPropertiesPosition; - } else { - return data.capacity() - applicationPropertiesPosition; - } - } - return 0; + protected ApplicationProperties readApplicationProperties(ReadableBuffer data, int position) { + return scanForMessageSection(data, position, ApplicationProperties.class); } @SuppressWarnings("unchecked") @@ -661,9 +658,6 @@ protected synchronized void ensureMessageDataScanned() { case NOT_SCANNED: scanMessageData(); break; - case RELOAD_PERSISTENCE: - lazyScanAfterReloadPersistence(); - break; case SCANNED: // NO-OP break; @@ -677,17 +671,18 @@ protected int getEstimateSavedEncode() { protected synchronized void resetMessageData() { header = null; + applicationPropertiesCount = 0; messageAnnotations = null; properties = null; applicationProperties = null; if (!expirationReload) { expiration = 0; } - priority = DEFAULT_MESSAGE_PRIORITY; + priority = VALUE_NOT_PRESENT; + durable = VALUE_NOT_PRESENT; encodedHeaderSize = 0; - memoryEstimate = -1; - originalEstimate = -1; - scheduledTime = -1; + memoryEstimate = VALUE_NOT_PRESENT; + scheduledTime = VALUE_NOT_PRESENT; encodedDeliveryAnnotationsSize = 0; headerPosition = VALUE_NOT_PRESENT; deliveryAnnotationsPosition = VALUE_NOT_PRESENT; @@ -742,7 +737,7 @@ protected synchronized void scanMessageData(ReadableBuffer data) { // Lazy decoding will start at the TypeConstructor of these ApplicationProperties // but we scan past it to grab the location of the possible body and footer section. applicationPropertiesPosition = constructorPos; - constructor.skipValue(); + this.applicationPropertiesCount = parseApCountAndSkip(data); remainingBodyPosition = data.hasRemaining() ? data.position() : VALUE_NOT_PRESENT; break; } else { @@ -759,6 +754,44 @@ protected synchronized void scanMessageData(ReadableBuffer data) { this.messageDataScanned = MessageDataScanningStatus.SCANNED.code; } + public int getApplicationPropertiesCount() { + ensureScanning(); + return applicationPropertiesCount; + } + + + // this is "borrowed" from: + // https://github.com/apache/qpid-proton-j/blob/6dc5587f1d1b23969a8994f1755198e638e92bc4/proton-j/src/main/java/org/apache/qpid/proton/codec/messaging/FastPathApplicationPropertiesType.java#L93-L115 + private static int parseApCountAndSkip(ReadableBuffer data) { + byte encodingCode = data.get(); + int count; + int size; + int sizePosition; + switch (encodingCode) { + case EncodingCodes.MAP8: + size = data.get() & 0xff; + sizePosition = data.position(); + count = data.get() & 0xff; + break; + case EncodingCodes.MAP32: + size = data.getInt(); + sizePosition = data.position(); + count = data.getInt(); + break; + case EncodingCodes.NULL: + sizePosition = data.position(); + size = 0; + count = 0; + break; + default: + throw new ProtonException("Expected Map type but found encoding: " + encodingCode); + } + + data.position(sizePosition + size); + + return count / 2; + } + @Override public abstract org.apache.activemq.artemis.api.core.Message copy(); @@ -883,16 +916,6 @@ public final void receiveBuffer(ByteBuf buffer) { @Override public abstract int getMemoryEstimate(); - @Override - public int getOriginalEstimate() { - if (originalEstimate < 0) { - // getMemoryEstimate should initialize originalEstimate - return getMemoryEstimate(); - } else { - return originalEstimate; - } - } - @Override public Map toPropertyMap(int valueSizeLimit) { return toPropertyMap(false, valueSizeLimit); @@ -1032,16 +1055,6 @@ protected int internalPersistSize() { @Override public abstract void reloadPersistence(ActiveMQBuffer record, CoreMessageObjectPools pools); - protected synchronized void lazyScanAfterReloadPersistence() { - assert messageDataScanned == MessageDataScanningStatus.RELOAD_PERSISTENCE.code; - scanMessageData(); - messageDataScanned = MessageDataScanningStatus.SCANNED.code; - modified = false; - // reinitialise memory estimate as message will already be on a queue - // and lazy decode will want to update - getMemoryEstimate(); - } - @Override public abstract long getPersistentSize() throws ActiveMQException; @@ -1220,13 +1233,18 @@ public final Object getDuplicateProperty() { @Override public boolean isDurable() { - if (header != null && header .getDurable() != null) { - return header.getDurable(); - } else { - // if header == null and scanningStatus=RELOAD_PERSISTENCE, it means the message can only be durable - // even though the parsing hasn't happened yet - return getDataScanningStatus() == MessageDataScanningStatus.RELOAD_PERSISTENCE; + if (this.durable == VALUE_NOT_PRESENT) { + if (header != null && header.getDurable() != null) { + this.durable = header.getDurable() ? (byte)1 : (byte)0; + } else { + this.durable = 0; + } } + return this.durable == 1; + } + + void reloadSetDurable(boolean booleanDurable) { + this.durable = booleanDurable ? (byte)1 : (byte)0; } @Override @@ -1236,6 +1254,7 @@ public final org.apache.activemq.artemis.api.core.Message setDurable(boolean dur } header.setDurable(durable); // Message needs to be re-encoded following this action. + reloadSetDurable(durable); return this; } @@ -1308,9 +1327,20 @@ public final org.apache.activemq.artemis.api.core.Message setTimestamp(long time @Override public final byte getPriority() { + if (priority == VALUE_NOT_PRESENT) { + ensureScanning(); + // if still not present, it means it doesn't have it, so we use the default + if (priority == VALUE_NOT_PRESENT) { + priority = DEFAULT_MESSAGE_PRIORITY; + } + } return priority; } + void reloadPriority(byte priority) { + this.priority = priority; + } + @Override public final org.apache.activemq.artemis.api.core.Message setPriority(byte priority) { // Internally we can only deal with a limited range, but the AMQP value is allowed @@ -1322,6 +1352,7 @@ public final org.apache.activemq.artemis.api.core.Message setPriority(byte prior header = new Header(); } header.setPriority(UnsignedByte.valueOf(priority)); + this.priority = priority; return this; } diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersister.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersister.java index bc3d0903326..299aad98fc7 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersister.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersister.java @@ -80,6 +80,17 @@ record = new AMQPStandardMessage(format); if (address != null) { record.setAddress(address); } + + if (ID == getID()) { + scanAfterReload((AMQPStandardMessage) record); + } return record; } + + // notice this is muted after AMQPMessagePersisterV4 + // we will scan the message after reloading for older versions, while after V4 we keep everything in the storage. + // this is to give the broker a chance to reload messages the first time it's moved through version upgrade + protected void scanAfterReload(AMQPStandardMessage message) { + message.scanMessageData(); + } } diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV2.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV2.java index 7d599d12114..3034a604af2 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV2.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV2.java @@ -98,6 +98,9 @@ public Message decode(ActiveMQBuffer buffer, Message ignore, CoreMessageObjectPo extraProperties.decode(buffer.byteBuf(), pool != null ? pool.getPropertiesDecoderPools() : null); } record.reloadAddress(address); + if (ID == getID()) { + scanAfterReload((AMQPStandardMessage) record); + } return record; } diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV3.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV3.java index b3d86d055e8..9878e8cdb24 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV3.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV3.java @@ -54,9 +54,6 @@ public int getEncodeSize(Message record) { } - /** - * Sub classes must add the first short as the protocol-id - */ @Override public void encode(ActiveMQBuffer buffer, Message record) { super.encode(buffer, record); @@ -72,6 +69,10 @@ assert record != null && AMQPStandardMessage.class.equals(record.getClass()); ((AMQPStandardMessage)record).reloadExpiration(buffer.readLong()); + if (ID == getID()) { + scanAfterReload((AMQPStandardMessage) record); + } + return record; } diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV4.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV4.java new file mode 100644 index 00000000000..919d17cf52d --- /dev/null +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV4.java @@ -0,0 +1,109 @@ +/* + * 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.protocol.amqp.broker; + +import java.lang.invoke.MethodHandles; + +import org.apache.activemq.artemis.api.core.ActiveMQBuffer; +import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools; +import org.apache.activemq.artemis.utils.DataConstants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPMessagePersisterV4_ID; + +/** + * V4 adds a size field to determine persister boundaries, enabling forward-compatible + * extensions without additional versioning. + **/ +public class AMQPMessagePersisterV4 extends AMQPMessagePersisterV3 { + + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public static final byte ID = AMQPMessagePersisterV4_ID; + + public static AMQPMessagePersisterV4 theInstance; + + public static AMQPMessagePersisterV4 getInstance() { + if (theInstance == null) { + theInstance = new AMQPMessagePersisterV4(); + } + return theInstance; + } + + @Override + public byte getID() { + return ID; + } + + public AMQPMessagePersisterV4() { + super(); + } + + + protected static final int PERSISTER_SIZE = DataConstants.SIZE_INT + // memory estimate + DataConstants.SIZE_BYTE + + DataConstants.SIZE_BOOLEAN; // message priority + + @Override + public int getEncodeSize(Message record) { + int encodeSize = super.getEncodeSize(record) + PERSISTER_SIZE + DataConstants.SIZE_INT; // the size delimiter and whatever is written in encode + return encodeSize; + } + + + @Override + public void encode(ActiveMQBuffer buffer, Message record) { + super.encode(buffer, record); + + writeSizeDelimiter(buffer); + buffer.writeInt(record.getMemoryEstimate()); + buffer.writeByte(record.getPriority()); + buffer.writeBoolean(record.isDurable()); + } + + protected void writeSizeDelimiter(ActiveMQBuffer buffer) { + // this is to allow us to determine the boundary of this persister, for future use. + buffer.writeInt(PERSISTER_SIZE); + } + + @Override + public Message decode(ActiveMQBuffer buffer, Message ignore, CoreMessageObjectPools pool) { + Message record = super.decode(buffer, ignore, pool); + + int sizePersister = buffer.readInt(); + int lastPosition = buffer.readerIndex() + sizePersister; + + { + AMQPStandardMessage standardMessage = (AMQPStandardMessage) record; + standardMessage.setMemoryEstimate(buffer.readInt()); + standardMessage.reloadPriority(buffer.readByte()); + standardMessage.reloadSetDurable(buffer.readBoolean()); + + assert buffer.readerIndex() <= lastPosition; + } + + // if a future version of this persister wrote more bytes than what we expected now, this will take care of skipping them + buffer.readerIndex(lastPosition); + + // note that beyond V4 we are not calling scanAfterReload + + return record; + } + +} diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPStandardMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPStandardMessage.java index c514009bcee..14aab2614f8 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPStandardMessage.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPStandardMessage.java @@ -39,12 +39,8 @@ import org.apache.qpid.proton.amqp.messaging.MessageAnnotations; import org.apache.qpid.proton.amqp.messaging.Properties; import org.apache.qpid.proton.amqp.messaging.Section; -import org.apache.qpid.proton.codec.DecodeException; -import org.apache.qpid.proton.codec.DecoderImpl; import org.apache.qpid.proton.codec.EncoderImpl; -import org.apache.qpid.proton.codec.EncodingCodes; import org.apache.qpid.proton.codec.ReadableBuffer; -import org.apache.qpid.proton.codec.TypeConstructor; import org.apache.qpid.proton.codec.WritableBuffer; // see https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format @@ -191,20 +187,25 @@ protected ReadableBuffer getData() { @Override public synchronized int getMemoryEstimate() { - if (memoryEstimate == -1) { - if (isPaged) { - // When the message is paged, we don't take the unmarshalled application properties because it could be - // updated at different places. We just keep the estimate simple when paging. - memoryEstimate = memoryOffset + (data != null ? data.capacity() : 0); - } else { - memoryEstimate = memoryOffset + (data != null ? data.capacity() + unmarshalledApplicationPropertiesMemoryEstimateFromData(data) : 0); - } - originalEstimate = memoryEstimate; + if (memoryEstimate == VALUE_NOT_PRESENT) { + // This estimation was tested and validated through AMQPGlobalMaxTest on soak-tests + memoryEstimate = MINIMUM_ESTIMATE + (data != null ? data.capacity() + getApplicationPropertiesEncodingSize(data) * 2 + getApplicationPropertiesCount() * DataConstants.SIZE_INT : 0); } return memoryEstimate; } + private int getApplicationPropertiesEncodingSize(ReadableBuffer data) { + ensureScanning(); + + // ensureScanning will make sure we know the positions, and we will return the encoding used based on those positions + if (remainingBodyPosition != VALUE_NOT_PRESENT) { + return remainingBodyPosition - applicationPropertiesPosition; + } else { + return data.capacity() - applicationPropertiesPosition; + } + } + @Override public void persist(ActiveMQBuffer targetRecord) { @@ -237,79 +238,9 @@ public void reloadPersistence(ActiveMQBuffer record, CoreMessageObjectPools pool // Message state is now that the underlying buffer is loaded, but the contents not yet scanned resetMessageData(); - recoverHeaderDataFromEncoding(); modified = false; - messageDataScanned = MessageDataScanningStatus.RELOAD_PERSISTENCE.code; - } - - private void recoverHeaderDataFromEncoding() { - final DecoderImpl decoder = TLSEncode.getDecoder(); - decoder.setBuffer(data); - - try { - // At one point the broker could write the header and delivery annotations out of order - // which means a full scan is required for maximum compatibility with that older data - // where delivery annotations could be found ahead of the Header in the encoding. - // - // We manually extract the priority from the Header encoding if present to ensure we do - // not create any unneeded GC overhead during load from storage. We don't directly store - // other values from the header except for a value that is computed based on TTL and or - // absolute expiration time in the Properties section, but that value is stored in the - // data of the persisted message. - for (int section = 0; section < 2 && data.hasRemaining(); section++) { - final TypeConstructor constructor = decoder.readConstructor(); - - if (Header.class.equals(constructor.getTypeClass())) { - final byte typeCode = data.get(); - - @SuppressWarnings("unused") - int size = 0; - int count = 0; - - switch (typeCode) { - case EncodingCodes.LIST0: - break; - case EncodingCodes.LIST8: - size = data.get() & 0xff; - count = data.get() & 0xff; - break; - case EncodingCodes.LIST32: - size = data.getInt(); - count = data.getInt(); - break; - default: - throw new DecodeException("Incorrect type found in Header encoding: " + typeCode); - } - - // Priority is stored in the second slot of the Header list encoding if present - if (count >= 2) { - decoder.readBoolean(false); // Discard durable for now, it is computed elsewhere. - - final byte encodingCode = data.get(); - final int priority = switch (encodingCode) { - case EncodingCodes.UBYTE -> data.get() & 0xff; - case EncodingCodes.NULL -> DEFAULT_MESSAGE_PRIORITY; - default -> - throw new DecodeException("Expected UnsignedByte type but found encoding: " + EncodingCodes.toString(encodingCode)); - }; - - // Scaled here so do not call setPriority as that will store the set value in the AMQP header - // and we don't want to create that Header instance at this stage. - this.priority = (byte) Math.min(priority, MAX_MESSAGE_PRIORITY); - } - - return; - } else if (DeliveryAnnotations.class.equals(constructor.getTypeClass())) { - constructor.skipValue(); - } else { - return; - } - } - } finally { - decoder.setBuffer(null); - data.rewind(); // Ensure next scan start at the beginning. - } + messageDataScanned = MessageDataScanningStatus.NOT_SCANNED.code; } @Override @@ -319,7 +250,7 @@ public long getPersistentSize() throws ActiveMQException { @Override public Persister getPersister() { - return AMQPMessagePersisterV3.getInstance(); + return AMQPMessagePersisterV4.getInstance(); } @Override diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManagerFactory.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManagerFactory.java index faa9039338f..314b5d41ae0 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManagerFactory.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManagerFactory.java @@ -52,7 +52,8 @@ public class ProtonProtocolManagerFactory extends AbstractProtocolManagerFactory @Override public Persister[] getPersister() { - Persister[] persisters = new Persister[]{AMQPMessagePersister.getInstance(), AMQPMessagePersisterV2.getInstance(), AMQPLargeMessagePersister.getInstance(), AMQPMessagePersisterV3.getInstance()}; + Persister[] persisters = new Persister[]{AMQPMessagePersister.getInstance(), AMQPMessagePersisterV2.getInstance(), AMQPMessagePersisterV3.getInstance(), + AMQPMessagePersisterV4.getInstance(), AMQPLargeMessagePersister.getInstance(), AMQPLargeMessagePersisterV2.getInstance()}; return persisters; } diff --git a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageTest.java b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageTest.java index 51228637f4d..8a10f33ce07 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageTest.java +++ b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageTest.java @@ -87,7 +87,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -216,11 +215,11 @@ public void testHasScheduledDeliveryTimeReloadPersistence() { // Now reload from encoded data message.reloadPersistence(encoded, null); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); assertTrue(message.hasScheduledDeliveryTime()); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); message.getHeader(); @@ -249,11 +248,11 @@ public void testHasScheduledDeliveryDelayReloadPersistence() { // Now reload from encoded data message.reloadPersistence(encoded, null); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); assertTrue(message.hasScheduledDeliveryTime()); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); message.getHeader(); @@ -279,11 +278,11 @@ public void testNoScheduledDeliveryTimeOrDelayReloadPersistence() { // Now reload from encoded data message.reloadPersistence(encoded, null); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); assertFalse(message.hasScheduledDeliveryTime()); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); message.getHeader(); @@ -331,12 +330,7 @@ private void testGetMemoryEstimateWithDecodedApplicationProperties(boolean paged } assertEquals(TEST_APPLICATION_PROPERTY_VALUE, decodedWithApplicationPropertiesUnmarshalled.getStringProperty(TEST_APPLICATION_PROPERTY_KEY)); - - if (paged) { - assertEquals(decodedWithApplicationPropertiesUnmarshalled.getMemoryEstimate(), decoded.getMemoryEstimate()); - } else { - assertNotEquals(decodedWithApplicationPropertiesUnmarshalled.getMemoryEstimate(), decoded.getMemoryEstimate()); - } + assertEquals(decodedWithApplicationPropertiesUnmarshalled.getMemoryEstimate(), decoded.getMemoryEstimate()); } //----- Test Connection ID access -----------------------------------------// @@ -1647,7 +1641,7 @@ public void testExtraByteProperty() { } @Test - public void testExtraProperty() { + public void testJournalPersistence() { MessageImpl protonMessage = (MessageImpl) Message.Factory.create(); byte[] original = RandomUtil.randomBytes(); @@ -1655,19 +1649,24 @@ public void testExtraProperty() { AMQPStandardMessage decoded = encodeAndDecodeMessage(protonMessage); decoded.setAddress("someAddress"); decoded.setMessageID(33); + decoded.setPriority((byte) 2); + decoded.putStringProperty("hello", "world"); decoded.putExtraBytesProperty(name, original); - - ICoreMessage coreMessage = decoded.toCore(); - assertSame(original, coreMessage.getBytesProperty(name)); + decoded.reencode(); ActiveMQBuffer buffer = ActiveMQBuffers.pooledBuffer(10 * 1024); try { decoded.getPersister().encode(buffer, decoded); - assertEquals(AMQPMessagePersisterV3.getInstance().getID(), buffer.readByte()); // the journal reader will read 1 byte to find the persister + assertEquals(AMQPMessagePersisterV4.getInstance().getID(), buffer.readByte()); // the journal reader will read 1 byte to find the persister AMQPStandardMessage readMessage = (AMQPStandardMessage)decoded.getPersister().decode(buffer, null, null); assertEquals(33, readMessage.getMessageID()); assertEquals("someAddress", readMessage.getAddress()); assertArrayEquals(original, readMessage.getExtraBytesProperty(name)); + assertEquals((byte) 2, readMessage.getPriority()); + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, readMessage.getDataScanningStatus()); + assertEquals("world", readMessage.getStringProperty("hello")); + assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, readMessage.getDataScanningStatus()); + } finally { buffer.release(); } @@ -2886,6 +2885,7 @@ private boolean isEquals(MessageImpl left, MessageImpl right) { assertTrue(isEquals(left.getBody(), right.getBody())); assertFootersEquals(left.getFooter(), right.getFooter()); } catch (Throwable e) { + logger.debug(e.getMessage(), e); return false; } diff --git a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java index 87fc94e6365..bb7a7266bd6 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java +++ b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java @@ -443,7 +443,6 @@ public void testExpandPropertiesAndConvert() throws Exception { Map mapprop = createPropertiesMap(); ApplicationProperties properties = new ApplicationProperties(mapprop); - properties.getValue().put("hello", "hello"); MessageImpl message = (MessageImpl) Message.Factory.create(); MessageAnnotations annotations = new MessageAnnotations(new HashMap<>()); message.setMessageAnnotations(annotations); @@ -457,7 +456,9 @@ public void testExpandPropertiesAndConvert() throws Exception { encodedMessage.setAddress(SimpleString.of("xxxx.v1.queue")); for (int i = 0; i < 100; i++) { - encodedMessage.putStringProperty("another" + i, "value" + i); + if (i > 5) { // first 5 without application properties, to make it more challenging + encodedMessage.putStringProperty("another" + i, "value" + i); + } encodedMessage.messageChanged(); encodedMessage.reencode(); AmqpValue value = (AmqpValue) encodedMessage.getProtonMessage().getBody(); @@ -465,7 +466,7 @@ public void testExpandPropertiesAndConvert() throws Exception { ICoreMessage coreMessage = encodedMessage.toCore(); logger.debug("Converted message: {}", coreMessage); - // I'm going to replace the message every 10 messages by a re-encoded version to check if the wiring still acturate. + // I'm going to replace the message every 10 messages by a re-encoded version to check if the wiring is still accurate. // I want to mix replacing and not replacing to make sure the re-encoding is not giving me any surprises if (i > 0 && i % 10 == 0) { ByteBuf buf = Unpooled.buffer(15 * 1024, 150 * 1024); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index 345f1bb8605..e693589a0f6 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java @@ -775,7 +775,7 @@ public void refUp(MessageReference messageReference) { if (isMirrorController() && owner != null && pagingStore != owner) { // When using mirror in this situation, it means the address belong to another queue // it's acting as if the message is being copied - pagingStore.addSize(messageReference.getMessage().getOriginalEstimate(), false, false); + pagingStore.addSize(messageReference.getMessage().getMemoryEstimate(), false, false); } pagingStore.refUp(messageReference.getMessage(), count); @@ -793,7 +793,7 @@ public void refDown(MessageReference messageReference) { if (isMirrorController() && owner != null && pagingStore != owner) { // When using mirror in this situation, it means the address belong to another queue // it's acting as if the message is being copied - pagingStore.addSize(-messageReference.getMessage().getOriginalEstimate(), false, false); + pagingStore.addSize(-messageReference.getMessage().getMemoryEstimate(), false, false); } pagingStore.refDown(messageReference.getMessage(), count); } diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/utils/RealServerTestBase.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/utils/RealServerTestBase.java index 43b41cba771..394c3b6dd38 100644 --- a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/utils/RealServerTestBase.java +++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/utils/RealServerTestBase.java @@ -38,6 +38,7 @@ import java.util.Properties; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import org.apache.activemq.artemis.api.core.RoutingType; import org.apache.activemq.artemis.api.core.SimpleString; @@ -152,13 +153,21 @@ public Process startServer(String serverName, int portID, int timeout) throws Ex } public Process startServer(String serverName, int portID, int timeout, File brokerProperties) throws Exception { - Process process = ServerUtil.startServer(getServerLocation(serverName), serverName, portID, timeout, brokerProperties); + return startServer(serverName, portID, timeout, brokerProperties, null); + } + + public Process startServer(String serverName, int portID, int timeout, File brokerProperties, Consumer logCallback) throws Exception { + Process process = ServerUtil.startServer(getServerLocation(serverName), serverName, portID, timeout, brokerProperties, logCallback); addProcess(process); return process; } public Process startServer(String serverName, String uri, int timeout) throws Exception { - Process process = ServerUtil.startServer(getServerLocation(serverName), serverName, uri, timeout); + return startServer(serverName, uri, timeout, null); + } + + public Process startServer(String serverName, String uri, int timeout, Consumer logCallback) throws Exception { + Process process = ServerUtil.startServer(getServerLocation(serverName), serverName, uri, timeout, null, logCallback); addProcess(process); return process; } diff --git a/tests/compatibility-tests/src/main/resources/meshTest/sendMessages.groovy b/tests/compatibility-tests/src/main/resources/meshTest/sendMessages.groovy index cb4b96ce5f7..11ac4c23c7b 100644 --- a/tests/compatibility-tests/src/main/resources/meshTest/sendMessages.groovy +++ b/tests/compatibility-tests/src/main/resources/meshTest/sendMessages.groovy @@ -26,11 +26,20 @@ String serverType = arg[0]; String clientType = arg[1]; String operation = arg[2]; String protocol = null; +int messageSize; if (arg.length > 3) { protocol = arg[3]; } +if (arg.length > 4) { + messageSize = Integer.parseInt(arg[4]); +} else { + messageSize = 100; +} + +System.out.println("Message size = " + messageSize + ", operation = " + operation) + try { legacyOption = legacy; } catch (Throwable e) { @@ -61,8 +70,7 @@ BYTES_BODY[0] = (byte) 0x77; BYTES_BODY[1] = (byte) 0x77; BYTES_BODY[2] = (byte) 0x77; -String textBody = "a rapadura e doce mas nao e mole nao"; - +String textBody = "a".repeat(messageSize); if (clientType.startsWith("ARTEMIS")) { // Can't depend directly on artemis, otherwise it wouldn't compile in hornetq diff --git a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/JournalCompatibilityTest.java b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/JournalCompatibilityTest.java index 9008aacecd3..997eaf37a7d 100644 --- a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/JournalCompatibilityTest.java +++ b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/JournalCompatibilityTest.java @@ -17,6 +17,7 @@ package org.apache.activemq.artemis.tests.compatibility; +import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_33_0; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.SNAPSHOT; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_4_0; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_1_0; @@ -47,6 +48,7 @@ public static Collection getParameters() { combinations.add(new Object[]{ARTEMIS_2_1_0, SNAPSHOT}); combinations.add(new Object[]{ARTEMIS_2_4_0, SNAPSHOT}); + combinations.add(new Object[]{ARTEMIS_2_33_0, SNAPSHOT}); // the purpose on this one is just to validate the test itself. /// if it can't run against itself it won't work at all combinations.add(new Object[]{SNAPSHOT, SNAPSHOT}); @@ -90,39 +92,45 @@ public void testSendReceive() throws Throwable { } @TestTemplate - public void testSendReceivePaging() throws Throwable { - setVariable(senderClassloader, "persistent", true); - startServer(serverFolder, sender, senderClassloader, "journalTest", null, true); - evaluate(senderClassloader, "journalcompatibility/forcepaging.groovy"); - evaluate(senderClassloader, "meshTest/sendMessages.groovy", sender, sender, "sendAckMessages"); - evaluate(senderClassloader, "journalcompatibility/ispaging.groovy"); - stopServer(senderClassloader); - - setVariable(receiverClassloader, "persistent", true); - startServer(serverFolder, receiver, receiverClassloader, "journalTest", null, false); - evaluate(receiverClassloader, "journalcompatibility/ispaging.groovy"); + public void testSendReceiveAMQP() throws Throwable { + internalSendReceiveAMQP("AMQP", 100, false); + } - setVariable(receiverClassloader, "latch", null); - evaluate(receiverClassloader, "meshTest/sendMessages.groovy", receiver, receiver, "receiveMessages"); + @TestTemplate + public void testSendReceivePaging() throws Throwable { + internalSendReceiveAMQP("CORE", 100, false); } @TestTemplate public void testSendReceiveAMQPPaging() throws Throwable { + internalSendReceiveAMQP("AMQP", 100, false); + } + + @TestTemplate + public void testSendReceiveLargeAMQPPaging() throws Throwable { + internalSendReceiveAMQP("AMQP", 200 * 1024, true); + } + + + private void internalSendReceiveAMQP(String protocol, int size, boolean paging) throws Throwable { + setVariable(senderClassloader, "persistent", true); startServer(serverFolder, sender, senderClassloader, "journalTest", null, true); - evaluate(senderClassloader, "journalcompatibility/forcepaging.groovy"); - evaluate(senderClassloader, "meshTest/sendMessages.groovy", sender, sender, "sendAckMessages", "AMQP"); - evaluate(senderClassloader, "journalcompatibility/ispaging.groovy"); + if (paging) { + evaluate(senderClassloader, "journalcompatibility/forcepaging.groovy"); + } + evaluate(senderClassloader, "meshTest/sendMessages.groovy", sender, sender, "sendAckMessages", "AMQP", String.valueOf(size)); stopServer(senderClassloader); setVariable(receiverClassloader, "persistent", true); startServer(serverFolder, receiver, receiverClassloader, "journalTest", null, false); - evaluate(receiverClassloader, "journalcompatibility/ispaging.groovy"); setVariable(receiverClassloader, "latch", null); - evaluate(receiverClassloader, "meshTest/sendMessages.groovy", receiver, receiver, "receiveMessages", "AMQP"); + evaluate(receiverClassloader, "meshTest/sendMessages.groovy", receiver, receiver, "receiveMessages", "AMQP", String.valueOf(size)); } + + /** * Test that the server starts properly using an old journal even though persistent size metrics were not originaly * stored diff --git a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/MultiVersionReplicaTest.java b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/MultiVersionReplicaTest.java index 490c3352616..fa6fece3981 100644 --- a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/MultiVersionReplicaTest.java +++ b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/MultiVersionReplicaTest.java @@ -17,6 +17,7 @@ package org.apache.activemq.artemis.tests.compatibility; +import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_33_0; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.SNAPSHOT; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_44_0; import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.ARTEMIS_2_17_0; @@ -63,13 +64,11 @@ public static Collection getParameters() { if (getJavaVersion() <= 22) { // Old 2.x servers fail on JDK23+ without workarounds. combinations.add(new Object[]{ARTEMIS_2_22_0, SNAPSHOT}); - combinations.add(new Object[]{SNAPSHOT, ARTEMIS_2_22_0}); combinations.add(new Object[]{ARTEMIS_2_17_0, SNAPSHOT}); - combinations.add(new Object[]{SNAPSHOT, ARTEMIS_2_17_0}); + combinations.add(new Object[]{ARTEMIS_2_33_0, SNAPSHOT}); } combinations.add(new Object[]{ARTEMIS_2_44_0, SNAPSHOT}); - combinations.add(new Object[]{SNAPSHOT, ARTEMIS_2_44_0}); // The SNAPSHOT/SNAPSHOT is here as a test validation only, like in other cases where SNAPSHOT/SNAPSHOT is used. combinations.add(new Object[]{SNAPSHOT, SNAPSHOT}); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPReloadFromPersistenceTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPReloadFromPersistenceTest.java new file mode 100644 index 00000000000..07e4cf1bbfd --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPReloadFromPersistenceTest.java @@ -0,0 +1,297 @@ +/* + * 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.amqp; + +import java.util.HashMap; +import java.util.Map; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.apache.activemq.artemis.api.core.ActiveMQBuffer; +import org.apache.activemq.artemis.api.core.ActiveMQBuffers; +import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.io.SequentialFile; +import org.apache.activemq.artemis.core.persistence.StorageManager; +import org.apache.activemq.artemis.core.persistence.impl.nullpm.NullStorageManager; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPLargeMessage; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPLargeMessagePersister; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPLargeMessagePersisterV2; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessagePersister; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessagePersisterV2; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessagePersisterV3; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessagePersisterV4; +import org.apache.activemq.artemis.protocol.amqp.broker.AMQPStandardMessage; +import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable; +import org.apache.activemq.artemis.tests.unit.core.journal.impl.fakes.FakeSequentialFileFactory; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.utils.RandomUtil; +import org.apache.activemq.transport.amqp.client.AmqpMessage; +import org.apache.qpid.proton.message.impl.MessageImpl; +import org.jspecify.annotations.NonNull; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class AMQPReloadFromPersistenceTest extends ActiveMQTestBase { + + + @Test + public void testPersistCheck() throws Exception { + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, AMQPMessagePersisterV4.getInstance(), AMQPMessagePersisterV4.getInstance()); + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, new FakePersisterFromTheFuture(), AMQPMessagePersisterV4.getInstance()); + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, AMQPMessagePersisterV4.getInstance(), new FakePersisterFromTheFuture()); + } + + @Test + public void testPersistCheckOldVersion() throws Exception { + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.SCANNED, AMQPMessagePersisterV2.getInstance(), AMQPMessagePersisterV2.getInstance()); + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.SCANNED, AMQPMessagePersister.getInstance(), AMQPMessagePersister.getInstance()); + internalPersistCheck(AMQPMessage.MessageDataScanningStatus.SCANNED, AMQPMessagePersisterV3.getInstance(), AMQPMessagePersisterV3.getInstance()); + } + + private void internalPersistCheck(AMQPMessage.MessageDataScanningStatus expectedStatusAfterReload, AMQPMessagePersister persisterOnWrite, AMQPMessagePersister persisterOnRead) throws Exception { + + Map map = new HashMap(); + for (int i = 0; i < 77; i++) { + map.put("stuff" + i, "value" + i); // just filling stuff + } + boolean originalDurable = true; + AMQPStandardMessage originalMessage = AMQPStandardMessage.createMessage(1, 0, SimpleString.of("duh"), null, null, null, null, map, null, null); + byte originalPriority = 1; + originalMessage.setPriority(originalPriority); + originalMessage.setDurable(originalDurable); + originalMessage.reencode(); + int originalMemoryEstimate = originalMessage.getMemoryEstimate(); + + byte[] originalRandomBytes = RandomUtil.randomBytes(10); + + ActiveMQBuffer buffer = ActiveMQBuffers.dynamicBuffer(1024); + persisterOnWrite.encode(buffer, originalMessage); + buffer.writeBytes(originalRandomBytes); + + { + buffer.readerIndex(1); // first byte is the persister version + AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persisterOnRead.decode(buffer, null, null); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + assertEquals(originalPriority, amqpStandardMessage.getPriority()); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + + validateExtraBytes(originalRandomBytes, buffer); + } + + { + buffer.readerIndex(1); // first byte is the persister version + AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persisterOnRead.decode(buffer, null, null); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + assertEquals(originalMemoryEstimate, amqpStandardMessage.getMemoryEstimate()); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + + validateExtraBytes(originalRandomBytes, buffer); + } + + { + buffer.readerIndex(1); // first byte is the persister version + AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persisterOnRead.decode(buffer, null, null); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + assertEquals(originalDurable, amqpStandardMessage.isDurable()); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + + validateExtraBytes(originalRandomBytes, buffer); + } + + // none of these should make it scan anything + { + buffer.readerIndex(1); // first byte is the persister version + AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persisterOnRead.decode(buffer, null, null); + assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus()); + assertEquals(77, amqpStandardMessage.getApplicationPropertiesCount()); + assertEquals(77, amqpStandardMessage.getApplicationProperties().getValue().size()); + // this check here should always be scanned, no matter the version you use + assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, amqpStandardMessage.getDataScanningStatus()); + + validateExtraBytes(originalRandomBytes, buffer); + } + } + + private static void validateExtraBytes(byte[] originalRandomBytes, ActiveMQBuffer buffer) { + byte[] randomBytes = new byte[originalRandomBytes.length]; + buffer.readBytes(randomBytes); + assertArrayEquals(originalRandomBytes, randomBytes); + } + + @Test + public void testNoApplicationProperties() throws Exception { + AMQPStandardMessage originalMessage = AMQPStandardMessage.createMessage(1, 0, SimpleString.of("duh"), null, null, null, null, null, null, null); + + AMQPMessagePersisterV4 persister = AMQPMessagePersisterV4.getInstance(); + + ActiveMQBuffer buffer = ActiveMQBuffers.dynamicBuffer(1024); + persister.encode(buffer, originalMessage); + + buffer.readerIndex(1); // skip version byte from the persister + AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persister.decode(buffer, null, null); + + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, amqpStandardMessage.getDataScanningStatus()); + assertEquals(0, amqpStandardMessage.getApplicationPropertiesCount()); + assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, amqpStandardMessage.getDataScanningStatus()); + assertNull(amqpStandardMessage.getApplicationProperties()); + } + + /** + * Pretending to be a persister from the future with extra stuff, the current version should handle it okay by skipping the extra unkown data. + * */ + @Test + public void testPersistCheckLargeMessage() throws Exception { + internalPersistCheckLargeMessage(AMQPLargeMessagePersister.getInstance(), AMQPLargeMessagePersister.getInstance(), false); + internalPersistCheckLargeMessage(AMQPLargeMessagePersisterV2.getInstance(), AMQPLargeMessagePersisterV2.getInstance(), true); + internalPersistCheckLargeMessage(AMQPLargeMessagePersisterV2.getInstance(), new FakeLargePersisterFromTheFuture(), true); + internalPersistCheckLargeMessage(new FakeLargePersisterFromTheFuture(), AMQPLargeMessagePersisterV2.getInstance(), true); + } + + private void internalPersistCheckLargeMessage(AMQPLargeMessagePersister persisterOnWrite, AMQPLargeMessagePersister persisterOnRead, boolean checkMemoryEstimate) throws Exception { + FakeSequentialFileFactory fakeSequentialFileFactory = new FakeSequentialFileFactory(); + final StorageManager storageManager = new NullStorageManager() { + public SequentialFile createFileForLargeMessage(long messageID, boolean durable) { + return fakeSequentialFileFactory.createSequentialFile("messageID.large"); + } + }; + + storageManager.start(); + + AMQPLargeMessage originalMessage = createLargeMessage(storageManager, "test", 100, (byte) 3); + ActiveMQBuffer buffer = ActiveMQBuffers.dynamicBuffer(1024); + persisterOnWrite.encode(buffer, originalMessage); + + byte[] originalRandomBytes = RandomUtil.randomBytes(10); + buffer.writeBytes(originalRandomBytes); + + + { + buffer.readerIndex(1); // first byte is the persister version + AMQPLargeMessage largeMessageRead = (AMQPLargeMessage) persisterOnRead.decode(buffer, null, null); + assertEquals((byte) 3, largeMessageRead.getPriority()); + assertTrue(largeMessageRead.isDurable()); + if (checkMemoryEstimate) { + assertEquals(originalMessage.getMemoryEstimate(), largeMessageRead.getMemoryEstimate()); + } + validateExtraBytes(originalRandomBytes, buffer); + } + } + + private static MessageImpl createProtonMessage(String address, byte priority) { + AmqpMessage message = new AmqpMessage(); + message.setBytes(RandomUtil.randomBytes(10)); + message.setAddress(address); + message.setDurable(true); + message.setPriority(priority); + MessageImpl protonMessage = (MessageImpl) message.getWrappedMessage(); + return protonMessage; + } + + private static AMQPLargeMessage createLargeMessage(StorageManager storageManager, + String address, + long msgId, + byte priority) throws Exception { + MessageImpl protonMessage = createProtonMessage(address, priority); + byte[] messageContent = encodeProtonMessage(protonMessage, 1024); + final AMQPLargeMessage amqpMessage = new AMQPLargeMessage(msgId, 0, null, null, storageManager); + amqpMessage.setAddress(address); + amqpMessage.setFileDurable(true); + amqpMessage.addBytes(messageContent); + amqpMessage.reloadExpiration(0); + return amqpMessage; + } + + private static byte @NonNull [] encodeProtonMessage(MessageImpl message, int expectedSize) { + ByteBuf nettyBuffer = Unpooled.buffer(expectedSize); + message.encode(new NettyWritable(nettyBuffer)); + byte[] bytes = new byte[nettyBuffer.writerIndex()]; + nettyBuffer.readBytes(bytes); + return bytes; + } + + + + static class FakeLargePersisterFromTheFuture extends AMQPLargeMessagePersisterV2 { + + FakeLargePersisterFromTheFuture() { + super(); + } + + // how much stuff from the future is coming + private final int EXTRA_STUFF = 77; + + @Override + protected void writeSizeDelimiter(ActiveMQBuffer buffer) { + buffer.writeInt(AMQPLargeMessagePersisterV2.PERSISTER_SIZE + EXTRA_STUFF); + } + + @Override + public int getEncodeSize(Message record) { + // I am adding 77 bytes extra on this fake from the future + return super.getEncodeSize(record) + EXTRA_STUFF; + } + + @Override + public void encode(ActiveMQBuffer buffer, Message record) { + super.encode(buffer, record); + + for (int i = 0; i < EXTRA_STUFF; i++) { + buffer.writeByte((byte) 0xf); + } + } + } + + static class FakePersisterFromTheFuture extends AMQPMessagePersisterV4 { + + FakePersisterFromTheFuture() { + super(); + } + + // how much stuff from the future is coming + private final int EXTRA_STUFF = 77; + + @Override + protected void writeSizeDelimiter(ActiveMQBuffer buffer) { + buffer.writeInt(AMQPMessagePersisterV4.PERSISTER_SIZE + EXTRA_STUFF); + } + + @Override + public int getEncodeSize(Message record) { + // I am adding 77 bytes extra on this fake from the future + return super.getEncodeSize(record) + EXTRA_STUFF; + } + + @Override + public void encode(ActiveMQBuffer buffer, Message record) { + super.encode(buffer, record); + + for (int i = 0; i < EXTRA_STUFF; i++) { + buffer.writeByte((byte) 0xf); + } + } + } + + + + +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java index 3e5a0bced90..3c8ae51f009 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java @@ -303,10 +303,14 @@ protected void sendMessages(String destinationName, } protected void sendMessages(String destinationName, int count, boolean durable) throws Exception { - sendMessages(destinationName, count, durable, null); + sendMessages(destinationName, count, durable, null, null); } protected void sendMessages(String destinationName, int count, boolean durable, byte[] payload) throws Exception { + sendMessages(destinationName, count, durable, payload, null); + } + + protected void sendMessages(String destinationName, int count, boolean durable, byte[] payload, Map properties) throws Exception { AmqpClient client = createAmqpClient(); AmqpConnection connection = addConnection(client.connect()); try { @@ -320,6 +324,9 @@ protected void sendMessages(String destinationName, int count, boolean durable, if (payload != null) { message.setBytes(payload); } + if (properties != null) { + properties.forEach((a, b) -> message.setApplicationProperty(a, b)); + } sender.send(message); } } finally { diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AckManagerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AckManagerTest.java index c237b0de439..8745970b1d2 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AckManagerTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AckManagerTest.java @@ -97,7 +97,8 @@ public void setUp() throws Exception { super.setUp(); server1 = createServer(true, createDefaultConfig(0, true), 100024, -1, -1, -1); - server1.getConfiguration().addAddressSetting(SNF_NAME, new AddressSettings().setMaxSizeBytes(-1).setMaxSizeMessages(-1).setMaxReadPageMessages(20)); + server1.getConfiguration().addAddressSetting(SNF_NAME, new AddressSettings().setMaxSizeBytes(-1).setMaxSizeMessages(-1).setMaxReadPageMessages(20).setMaxReadPageMessages(-1)); + server1.getConfiguration().addAddressSetting("#", new AddressSettings().setMaxSizeBytes(-1).setMaxSizeMessages(-1).setMaxReadPageMessages(-1).setMaxReadPageMessages(-1)); server1.getConfiguration().getAcceptorConfigurations().clear(); server1.getConfiguration().addAcceptorConfiguration("server", "tcp://localhost:61616"); } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/journal/AmqpJournalLoadingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/journal/AmqpJournalLoadingTest.java index 37e4dd2282d..dc17eae16cb 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/journal/AmqpJournalLoadingTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/journal/AmqpJournalLoadingTest.java @@ -21,8 +21,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.activemq.artemis.core.server.Queue; import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; @@ -38,11 +41,18 @@ public class AmqpJournalLoadingTest extends AmqpClientTestSupport { @Test - public void durableMessageDataNotScannedOnRestartTest() throws Exception { - sendMessages(getQueueName(), 1, true); + public void durableMessageDataAfterRestart() throws Exception { + Map properties = new HashMap<>(); + properties.put("largeOne", "a".repeat(10 * 1024)); + sendMessages(getQueueName(), 1, true, null, properties); final Queue queueView = getProxyToQueue(getQueueName()); Wait.assertTrue("All messages should arrive", () -> queueView.getMessageCount() == 1); + AtomicInteger messageSize = new AtomicInteger(0); + queueView.forEach(r -> { + messageSize.addAndGet(r.getMessage().getMemoryEstimate()); + }); + server.stop(); server.start(); @@ -52,15 +62,22 @@ public void durableMessageDataNotScannedOnRestartTest() throws Exception { List messageReference = new ArrayList<>(1); + AtomicInteger messageSizeAfterRestart = new AtomicInteger(0); + afterRestartQueueView.forEach((next) -> { final AMQPMessage message = (AMQPMessage)next.getMessage(); - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); - assertTrue(message.isDurable()); - // Doing the check again in case isDurable messed up the scanning status. It should not change the status by definition - assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.getDataScanningStatus()); + long memoryEstimate = message.getMemoryEstimate(); // it should not change it + assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.getDataScanningStatus()); + message.getApplicationProperties(); // this one should change it + assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, message.getDataScanningStatus()); + // the estimate should be the same even after scanning + assertEquals(memoryEstimate, message.getMemoryEstimate()); messageReference.add(message); + messageSizeAfterRestart.addAndGet(next.getMessage().getMemoryEstimate()); }); + assertEquals(messageSize.get(), messageSizeAfterRestart.get()); + assertEquals(1, messageReference.size()); AmqpClient client = createAmqpClient(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/routing/ElasticQueueTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/routing/ElasticQueueTest.java index 41cc2771a8d..f404307d5f5 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/routing/ElasticQueueTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/routing/ElasticQueueTest.java @@ -421,7 +421,7 @@ public boolean validateUserAndRole(final String username, private void prepareNodesAndStartCombinedHeadTail() throws Exception { AddressSettings blockingQueue = new AddressSettings(); blockingQueue - .setMaxSizeBytes(100 * 1024) + .setMaxSizeBytes(400 * 1024) .setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL) .setSlowConsumerPolicy(SlowConsumerPolicy.KILL).setSlowConsumerThreshold(0).setSlowConsumerCheckPeriod(1) .setAutoDeleteQueues(false).setAutoDeleteAddresses(false); // so slow consumer can kick in! diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java new file mode 100644 index 00000000000..aac05d3be3e --- /dev/null +++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java @@ -0,0 +1,363 @@ +/* + * 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.soak.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 java.io.File; +import java.lang.invoke.MethodHandles; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; + +import org.apache.activemq.artemis.api.core.management.SimpleManagement; +import org.apache.activemq.artemis.cli.commands.helper.HelperCreate; +import org.apache.activemq.artemis.tests.soak.SoakTestBase; +import org.apache.activemq.artemis.tests.util.CFUtil; +import org.apache.activemq.artemis.utils.FileUtil; +import org.apache.activemq.artemis.utils.RandomUtil; +import org.apache.activemq.artemis.utils.Wait; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +public class AMQPGlobalMaxTest extends SoakTestBase { + + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + public static final String QUEUE_NAME = "simpleTest"; + public static final String SERVER_NAME = "global-max-test"; + private static File serverLocation; + + private int lastMessageSent; + + volatile String memoryUsed; + volatile long highestMemoryUsed; + volatile boolean hasOME = false; + + private Process server; + + private long avgSizeFirstEstimate; + + public static void createServers(long globalMaxSize, String vmSize) throws Exception { + serverLocation = getFileServerLocation(SERVER_NAME); + deleteDirectory(serverLocation); + + HelperCreate cliCreateServer = helperCreate(); + cliCreateServer.setUseAIO(false).setAllowAnonymous(true).setNoWeb(true).setArtemisInstance(serverLocation); + // to speedup producers + cliCreateServer.addArgs("--no-fsync"); + cliCreateServer.addArgs("--queues", QUEUE_NAME); + cliCreateServer.addArgs("--global-max-size", String.valueOf(globalMaxSize)); + // limiting memory to make the test more predictable + cliCreateServer.addArgs("--java-memory", vmSize); + cliCreateServer.createServer(); + + // Setting deduplication off, to not cause noise on the memory usage calculations + FileUtil.findReplace(new File(serverLocation, "/etc/artemis.profile"), "-Xms512M", "-Xms10M -verbose:gc -XX:-UseStringDeduplication"); + } + + @Test + public void testValidateMemoryEstimateLargeAMQP() throws Exception { + validateOME("AMQP", (s, i) -> { + try { + Message m = s.createTextMessage("a".repeat(200 * 1024) + RandomUtil.randomUUIDString()); + for (int propCount = 0; propCount < 5; propCount++) { + // making each string unique to avoid string deduplication from Garbage Collection + m.setStringProperty("prop" + propCount, "a".repeat(10 * 1024) + RandomUtil.randomUUIDString()); + } + return m; + } catch (Exception e) { + fail(e.getMessage()); + return null; + } + }, "30M", 1, 100); + } + + private void validateOME(String protocol, BiFunction messageCreator, String vmSize, int commitInterval, int printInterval) throws Exception { + // making the broker OME on purpose + // this is to help us calculate the size of each message + // for this reason the global-max-size is set really high on this test + createServers(1024 * 1024 * 1024, vmSize); + startServerWithLog(); + + executeTest(protocol, messageCreator, (a, b) -> { }, commitInterval, printInterval, 20_000_000, TimeUnit.MINUTES.toMillis(10), true); + } + + @Test + public void testValidateMemoryEstimateAMQP() throws Exception { + validateOME("AMQP", (s, i) -> { + try { + Message m = s.createMessage(); + for (int propCount = 0; propCount < 10; propCount++) { + // making each string unique to avoid string deduplication from Garbage Collection + m.setStringProperty("string" + propCount, RandomUtil.randomUUIDString()); + m.setLongProperty("myLong" + propCount, RandomUtil.randomLong()); + } + return m; + } catch (Throwable e) { + Assertions.fail(e.getMessage()); + return null; + } + }, "30M", 1000, 1000); + } + + private void startServerWithLog() throws Exception { + server = startServer(SERVER_NAME, 0, 5000, null, s -> { + logger.debug("{}", s); + if (s.contains("GC") && s.contains("->")) { + AMQPGlobalMaxTest.this.memoryUsed = parseMemoryUsageFromGCLOG(s); + long memoryUsedBytes = parseMemoryToBytes(memoryUsed); + if (memoryUsedBytes > highestMemoryUsed) { + highestMemoryUsed = memoryUsedBytes; + logger.info("Currently using {} on the server", memoryUsed); + } + } + + // Stop Page, start page or anything important + if (s.contains("INFO") || s.contains("WARN")) { + logger.info("{}", s); + } + + if (s.contains("OutOfMemoryError")) { + logger.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> OME!!!"); + AMQPGlobalMaxTest.this.hasOME = true; + } + }); + } + + @Test + public void testGlobalMax() throws Exception { + createServers(100 * 1024 * 1024, "200M"); + startServerWithLog(); + + executeTest("AMQP", (s, i) -> { + try { + Message message; + if (i % 100 == 0) { + // introduce a few large messages in the mix + message = s.createTextMessage("a".repeat(101 * 1024) + RandomUtil.randomUUIDString()); + message.setStringProperty("i", "a".repeat(10 * 1024) + RandomUtil.randomUUIDString()); + } else { + message = s.createMessage(); + } + message.setIntProperty("i", i); + // Making strings unique to avoid deduplication on GC + message.setStringProperty("someString", "a".repeat(1024) + RandomUtil.randomUUIDString()); + return message; + } catch (Throwable e) { + Assertions.fail(e.getMessage()); + return null; + } + }, (i, m) -> { + try { + assertEquals(i, m.getIntProperty("i")); + } catch (Throwable e) { + Assertions.fail(e.getMessage()); + } + }, 10_000, 10_000, 80_000, TimeUnit.MINUTES.toMillis(10), false); + } + + private void executeTest(String protocol, BiFunction messageCreator, + BiConsumer messageVerifier, + int commitInterval, + int printInterval, + int maxMessages, + long timeoutMilliseconds, + boolean expectOME) throws Exception { + ExecutorService service = Executors.newFixedThreadPool(2); + runAfter(service::shutdownNow); + + CountDownLatch latchDone = new CountDownLatch(1); + AtomicInteger errors = new AtomicInteger(0); + + // use some management operation to make AMQPMessages to unmarshal their application properties + service.execute(() -> { + try (SimpleManagement simpleManagement = new SimpleManagement("tcp://localhost:61616", null, null)) { + while (!latchDone.await(1, TimeUnit.SECONDS)) { + // this filter will not remove any messages, but it will force the AMQPMessage to unmarshal the applicationProperties + simpleManagement.removeMessagesOnQueue(QUEUE_NAME, "i=-1"); + } + } catch (Throwable e) { + logger.warn(e.getMessage(), e); + } + }); + + String initialMemory = this.memoryUsed; + + service.execute(() -> { + try { + theTest(protocol, messageCreator, messageVerifier, maxMessages, commitInterval, printInterval); + } catch (Throwable e) { + logger.warn(e.getMessage(), e); + errors.incrementAndGet(); + } finally { + latchDone.countDown(); + } + }); + + Wait.waitFor(() -> latchDone.await(1, TimeUnit.SECONDS) || hasOME, timeoutMilliseconds, 100); + + if (errors.get() != 0) { + logger.info("The client had an error, probably because the server is going slow as it's close to OME. We are assuming the expected OME from the test"); + hasOME = true; // we will assume the server was slow and the client failed badly + } + + if (hasOME) { + assertTrue(lastMessageSent > 0); + long allocatedBytes = parseMemoryToBytes(memoryUsed); + long initialBytes = parseMemoryToBytes(initialMemory); + long finalMemory = allocatedBytes - initialBytes; + long avgSizeUsed = finalMemory / lastMessageSent; + if (!expectOME) { + Assertions.fail("OME on broker. Current memory = " + memoryUsed + ", with " + lastMessageSent + " messages sent, possible AVG for each message is " + (finalMemory / lastMessageSent) + " while initial estimate advertised by AMQPMessage is " + avgSizeFirstEstimate); + } + assertTrue(avgSizeFirstEstimate >= avgSizeUsed, "Estimated message size is below actually used. First estimate = " + avgSizeFirstEstimate + " supposed to be >= " + avgSizeUsed); + logger.info("Original AVG Size estimate: {}, actual AVG after OME: {}", avgSizeFirstEstimate, avgSizeUsed); + + server.destroyForcibly(); + } else { + assertTrue(latchDone.await(1, TimeUnit.MINUTES)); + if (expectOME) { + Assertions.fail("Broker was supposed to fail with OME on this test"); + } + } + } + + private void theTest(String protocol, BiFunction messageCreator, + BiConsumer messageVerifier, int maxMessages, int commitInterval, int printInterval) throws Exception { + + logger.info("CommitInterval = {}", commitInterval, printInterval); + assert printInterval % commitInterval == 0; + + ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616"); + + boolean firstTX = true; + + SimpleManagement simpleManagement = new SimpleManagement("tcp://localhost:61616", null, null); + try (Connection connection = factory.createConnection()) { + Session session = connection.createSession(Session.SESSION_TRANSACTED); + MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME)); + + for (int i = 0; i < maxMessages; i++) { + Message message = messageCreator.apply(session, i); + producer.send(message); + lastMessageSent = i; + + if (firstTX) { + // we measure one message in the queue to get how much we estimate for each message + session.commit(); + Wait.assertEquals(1, () -> simpleManagement.getMessageCountOnAddress(QUEUE_NAME)); + long addressSize = simpleManagement.getAddressSize(QUEUE_NAME); + this.avgSizeFirstEstimate = addressSize; + logger.info("addressSize={}", addressSize); + firstTX = false; + } + + if ((i + 1) % commitInterval == 0) { + if ((i + 1) % printInterval == 0) { + logger.info("sent {} out of {}", i, maxMessages); + } + session.commit(); + } + } + session.commit(); + } + + try (Connection connection = factory.createConnection()) { + Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE); + MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME)); + connection.start(); + + for (int i = 0; i < maxMessages; i++) { + if (i % 1000 == 0) { + logger.info("Received {}", i); + } + Message message = consumer.receive(5000); + assertNotNull(message); + messageVerifier.accept(i, message); + } + assertNull(consumer.receiveNoWait()); + } + } + + /** + * Parses GC log output to extract memory after GC. + * Example: "GC(25) Pause Remark 15M->15M(32M) 2.351ms" returns "15M" + * + * @param gcLogLine the GC log line + * @return the memory value after GC (e.g., "15M"), or null if not found + */ + private static String parseMemoryUsageFromGCLOG(String gcLogLine) { + int arrowIndex = gcLogLine.indexOf("->"); + if (arrowIndex == -1) { + return null; + } + int startIndex = arrowIndex + 2; + int endIndex = gcLogLine.indexOf('(', startIndex); + if (endIndex == -1) { + return null; + } + return gcLogLine.substring(startIndex, endIndex); + } + + /** + * Parses memory string (e.g., "256M", "15M", "512K", "2G") to bytes. + * + * @param memoryStr the memory string + * @return the memory value in bytes + */ + private static long parseMemoryToBytes(String memoryStr) { + if (memoryStr == null || memoryStr.isEmpty()) { + return 0; + } + + char unit = memoryStr.charAt(memoryStr.length() - 1); + long value = Long.parseLong(memoryStr.substring(0, memoryStr.length() - 1)); + + switch (unit) { + case 'K': + case 'k': + return value * 1024; + case 'M': + case 'm': + return value * 1024 * 1024; + case 'G': + case 'g': + return value * 1024 * 1024 * 1024; + default: + return value; + } + } + +} diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java index 6949305d89e..f12ac9780a3 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java @@ -52,10 +52,12 @@ private static AMQPLargeMessage createLargeMessage(StorageManager storageManager SimpleString address, long msgId, byte[] content) throws Exception { + MessageImpl protonMessage = createProtonMessage(address.toString(), content); + byte[] messageContent = encodeProtonMessage(protonMessage, 1024); final AMQPLargeMessage amqpMessage = new AMQPLargeMessage(msgId, 0, null, null, storageManager); amqpMessage.setAddress(address); amqpMessage.setFileDurable(true); - amqpMessage.addBytes(content); + amqpMessage.addBytes(messageContent); amqpMessage.reloadExpiration(0); return amqpMessage; } @@ -78,12 +80,16 @@ protected void writeMessage(StorageManager storageManager, } public static AMQPStandardMessage encodeAndDecodeMessage(int messageFormat, MessageImpl message, int expectedSize) { - ByteBuf nettyBuffer = Unpooled.buffer(expectedSize); + byte[] bytes = encodeProtonMessage(message, expectedSize); + + return new AMQPStandardMessage(messageFormat, bytes, null); + } + private static byte[] encodeProtonMessage(MessageImpl message, int expectedSize) { + ByteBuf nettyBuffer = Unpooled.buffer(expectedSize); message.encode(new NettyWritable(nettyBuffer)); byte[] bytes = new byte[nettyBuffer.writerIndex()]; nettyBuffer.readBytes(bytes); - - return new AMQPStandardMessage(messageFormat, bytes, null); + return bytes; } }