You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by cl...@apache.org on 2026/09/16 16:07:32 UTC
(artemis) branch main updated: ARTEMIS-5573 and ARTEMIS-5975 Improve AMQP Size estimation
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/artemis.git
The following commit(s) were added to refs/heads/main by this push:
new 02b0f0e7c2 ARTEMIS-5573 and ARTEMIS-5975 Improve AMQP Size estimation
02b0f0e7c2 is described below
commit 02b0f0e7c219fcc5b1f305397ea2ac82fbd3649d
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Jul 28 16:31:20 2026 -0400
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
Co-authored-by: Robbie Gemmell <ro...@apache.org>
---
.../apache/activemq/artemis/util/ServerUtil.java | 42 ++-
.../activemq/artemis/api/core/ActiveMQBuffer.java | 10 +
.../core/buffers/impl/ChannelBufferWrapper.java | 9 +
.../artemis/core/persistence/PersisterIDs.java | 6 +-
.../artemis/utils/AbstractMapPersister.java | 11 +-
.../artemis/utils/collections/TypedProperties.java | 5 +
.../apache/activemq/artemis/api/core/Message.java | 20 +-
.../api/core/management/SimpleManagement.java | 12 +
.../impl/CompressedLargeMessageControllerImpl.java | 5 +
.../client/impl/LargeMessageControllerImpl.java | 5 +
.../core/protocol/core/CoreRemotingConnection.java | 5 +
.../core/protocol/core/impl/PacketImpl.java | 2 +
.../src/main/resources/activemq-version.properties | 2 +-
.../protocol/amqp/broker/AMQPLargeMessage.java | 41 ++-
.../amqp/broker/AMQPLargeMessagePersister.java | 8 +-
.../amqp/broker/AMQPLargeMessagePersisterV2.java | 111 +++++++
.../artemis/protocol/amqp/broker/AMQPMessage.java | 189 +++++++----
.../amqp/broker/AMQPMessageMetadataPersister.java | 193 +++++++++++
.../protocol/amqp/broker/AMQPMessagePersister.java | 11 +
.../amqp/broker/AMQPMessagePersisterV2.java | 3 +
.../amqp/broker/AMQPMessagePersisterV3.java | 7 +-
.../amqp/broker/AMQPMessagePersisterV4.java | 97 ++++++
.../amqp/broker/AMQPMetadataDecodingState.java | 44 +++
.../protocol/amqp/broker/AMQPStandardMessage.java | 114 ++-----
.../amqp/broker/ProtonProtocolManagerFactory.java | 3 +-
.../amqp/logger/ActiveMQAMQPProtocolLogger.java | 3 +
.../protocol/amqp/broker/AMQPMessageTest.java | 120 +++++--
.../protocol/amqp/converter/TestConversions.java | 7 +-
.../core/server/cluster/impl/BridgeImpl.java | 27 +-
.../artemis/core/server/impl/QueueImpl.java | 4 +-
.../spi/core/protocol/EmbedMessageUtil.java | 26 +-
.../spi/core/protocol/MessagePersister.java | 4 +
docs/user-manual/ha.adoc | 23 ++
docs/user-manual/upgrading.adoc | 2 +
docs/user-manual/versions.adoc | 14 +
pom.xml | 2 +-
.../activemq/artemis/utils/RealServerTestBase.java | 13 +-
.../main/resources/meshTest/sendMessages.groovy | 12 +-
.../compatibility/JournalCompatibilityTest.java | 42 +--
.../compatibility/MultiVersionReplicaTest.java | 20 ++
.../amqp/AMQPReloadFromPersistenceTest.java | 322 +++++++++++++++++++
.../integration/amqp/AmqpClientTestSupport.java | 9 +-
.../integration/amqp/connect/AckManagerTest.java | 22 +-
.../amqp/journal/AmqpJournalLoadingTest.java | 29 +-
.../integration/routing/ElasticQueueTest.java | 2 +-
.../tests/soak/paging/AMQPGlobalMaxTest.java | 357 +++++++++++++++++++++
.../tests/unit/core/paging/impl/AmqpPageTest.java | 14 +-
.../unit/core/remoting/ActiveMQBufferTestBase.java | 68 ++++
48 files changed, 1843 insertions(+), 254 deletions(-)
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 ce1e89cc48..40a2f2e37e 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.InputStream;
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 class ServerUtil {
}
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<String> logCallback) throws Exception {
+ final Process process = internalStartServer(artemisInstance, serverName, brokerProperties, logCallback);
// wait for start
if (timeout > 0) {
@@ -66,7 +71,11 @@ public class ServerUtil {
}
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<String> logCallback) throws Exception {
+ final Process process = internalStartServer(artemisInstance, serverName, propertiesFile, logCallback);
// wait for start
if (timeout != 0) {
@@ -78,20 +87,30 @@ public class ServerUtil {
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<String> 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<String> logCallback, String...args) throws IOException, ClassNotFoundException {
try {
boolean IS_WINDOWS = System.getProperty("os.name").toLowerCase().trim().startsWith("win");
@@ -117,11 +136,11 @@ public class ServerUtil {
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 @@ public class ServerUtil {
private final boolean sendToErr;
+ private final Consumer<String> logCallback;
+
ProcessLogger(final boolean print,
final InputStream is,
final String logName,
- final boolean sendToErr) throws ClassNotFoundException {
+ final boolean sendToErr,
+ final Consumer<String> 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 class ServerUtil {
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/api/core/ActiveMQBuffer.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
index 3ab4573b1e..26bd5d5acf 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
@@ -822,6 +822,16 @@ public interface ActiveMQBuffer extends DataInput {
*/
void writeShort(short value);
+ /**
+ * Sets the specified unsigned 16-bit short integer at the current {@code writerIndex} and increases the {@code writerIndex}
+ * by {@code 2} in this buffer.
+ *
+ * @param value The specified unsigned 16-bit short integer (0-65535)
+ * @throws IllegalArgumentException if {@code value} is less than 0 or greater than 65535
+ * @throws IndexOutOfBoundsException if {@code this.writableBytes} is less than {@code 2}
+ */
+ void writeUnsignedShort(int value);
+
/**
* Sets the specified 32-bit integer at the current {@code writerIndex} and increases the {@code writerIndex} by
* {@code 4} in this buffer.
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
index 283137f9eb..6320e02c71 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
@@ -640,6 +640,15 @@ public class ChannelBufferWrapper implements ActiveMQBuffer {
buffer.writeShort(value);
}
+ @Override
+ public void writeUnsignedShort(final int value) {
+ if (value < 0 || value > 65535) {
+ throw new IllegalArgumentException("Value must be in range 0-65535, got: " + value);
+ }
+ buffer.writeShort((short) value);
+ }
+
+
/**
* from {@link java.io.DataInput} interface
*/
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 483f0ae670..b34fd2ae27 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 @@ package org.apache.activemq.artemis.core.persistence;
*/
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-commons/src/main/java/org/apache/activemq/artemis/utils/AbstractMapPersister.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/AbstractMapPersister.java
index 5b3aa34834..e9cf34c871 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/AbstractMapPersister.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/AbstractMapPersister.java
@@ -96,9 +96,7 @@ public abstract class AbstractMapPersister<T> {
protected void writeHeader(ActiveMQBuffer buffer, int recordSize, int entries) {
buffer.writeInt(recordSize);
- // this is equivalent to writeUnsignedShort
- // TODO: bring writeUnsignedShort to ActiveMQBuffer
- buffer.writeShort((short)entries);
+ buffer.writeUnsignedShort((short)entries);
}
protected void writeBoolean(ActiveMQBuffer buffer, short key, boolean value) {
@@ -131,6 +129,13 @@ public abstract class AbstractMapPersister<T> {
buffer.writeByte(value);
}
+ protected void writeByteArray(ActiveMQBuffer buffer, short key, byte[] byteArray) {
+ buffer.writeShort(key);
+ buffer.writeByte(Datatypes.BYTE_ARRAY.getId());
+ buffer.writeInt(byteArray.length);
+ buffer.writeBytes(byteArray);
+ }
+
protected void writeByteArray(ActiveMQBuffer buffer, short key, int size, Consumer<ActiveMQBuffer> consumer) {
buffer.writeShort(key);
buffer.writeByte(Datatypes.BYTE_ARRAY.getId());
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/TypedProperties.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/TypedProperties.java
index 40ee98d556..663a93ce95 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/TypedProperties.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/TypedProperties.java
@@ -34,6 +34,7 @@ import java.util.function.Predicate;
import java.util.function.Supplier;
import io.netty.buffer.ByteBuf;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.logs.ActiveMQUtilBundle;
@@ -659,6 +660,10 @@ public class TypedProperties {
decode(buffer, null);
}
+ public int encode(ActiveMQBuffer buffer) {
+ return encode(buffer.byteBuf());
+ }
+
public int encode(final ByteBuf buffer) {
lock.readLock().lock();
try {
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 8d03a857a3..8d76150e5f 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
@@ -431,6 +431,18 @@ public interface Message {
*/
Message setDurable(boolean durable);
+ /**
+ * Returns a wire-compatible persister for embedding this message during clustering operations.
+ * <p>
+ * EmbedMessageUtil will contain the definitions for WIRE Version.
+ *
+ * @param embedWireVersion the wire protocol version to match
+ * @return the persister compatible with the specified wire version
+ */
+ default Persister<Message> getWireCompatiblePersister(int embedWireVersion) {
+ return getPersister();
+ }
+
Persister<Message> getPersister();
String getAddress();
@@ -884,14 +896,6 @@ public interface Message {
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 f904dd50c6..91cc0c4595 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
@@ -117,6 +117,18 @@ public class SimpleManagement implements AutoCloseable {
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-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
index 18389bbeb1..51b9395bcd 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
@@ -505,6 +505,11 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
}
+ @Override
+ public void writeUnsignedShort(final int value) {
+ throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
+ }
+
@Override
public void writeInt(final int value) {
throw new IllegalAccessError(OPERATION_NOT_SUPPORTED);
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
index 7d4580b074..44c53a693b 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
@@ -852,6 +852,11 @@ public class LargeMessageControllerImpl implements LargeMessageController {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
+ @Override
+ public void writeUnsignedShort(final int value) {
+ throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
+ }
+
public void writeMedium(final int value) {
throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE);
}
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
index 1c12e3491e..b5efb04ac0 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/CoreRemotingConnection.java
@@ -81,6 +81,11 @@ public interface CoreRemotingConnection extends RemotingConnection {
return version < PacketImpl.ARTEMIS_2_28_0_VERSION;
}
+ default boolean isBeforeEmbedWireVersion2() {
+ int version = getChannelVersion();
+ return version < PacketImpl.ARTEMIS_2_58_0_VERSION;
+ }
+
/**
* Sets the client protocol used on the communication. This will determine if the client has support for certain
* packet types
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
index 6d9ef60390..61eed175f0 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
@@ -58,6 +58,8 @@ public class PacketImpl implements Packet {
// 2.37.0
public static final int ARTEMIS_2_37_0_VERSION = 136;
+ public static final int ARTEMIS_2_58_0_VERSION = 137;
+
public static final SimpleString OLD_QUEUE_PREFIX = SimpleString.of("jms.queue.");
public static final SimpleString OLD_TEMP_QUEUE_PREFIX = SimpleString.of("jms.tempqueue.");
public static final SimpleString OLD_TOPIC_PREFIX = SimpleString.of("jms.topic.");
diff --git a/artemis-core-client/src/main/resources/activemq-version.properties b/artemis-core-client/src/main/resources/activemq-version.properties
index aaacbde788..d2ee7d9b41 100644
--- a/artemis-core-client/src/main/resources/activemq-version.properties
+++ b/artemis-core-client/src/main/resources/activemq-version.properties
@@ -20,4 +20,4 @@ activemq.version.minorVersion=${activemq.version.minorVersion}
activemq.version.microVersion=${activemq.version.microVersion}
activemq.version.incrementingVersion=${activemq.version.incrementingVersion}
activemq.version.versionTag=${activemq.version.versionTag}
-activemq.version.compatibleVersionList=121,122,123,124,125,126,127,128,129,130,131,132,133,134,135,136
+activemq.version.compatibleVersionList=121,122,123,124,125,126,127,128,129,130,131,132,133,134,135,136,137
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 99dbe2d8ad..84356f7680 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,8 @@ import org.apache.activemq.artemis.core.server.MessageReference;
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.spi.core.protocol.EmbedMessageUtil;
+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 +82,8 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
private boolean reencoded = false;
+ private int applicationPropertiesSize;
+
/**
* AMQPLargeMessagePersister will save the buffer here.
*/
@@ -264,7 +268,9 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
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 +406,12 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
}
}
+ @Override
+ protected synchronized void resetMessageData() {
+ super.resetMessageData();
+ applicationPropertiesSize = 0;
+ }
+
private void genericParseLargeMessage() {
try {
parsingBuffer.position(0);
@@ -412,6 +424,16 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
}
}
+ @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 +618,16 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
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 = BASE_MEMORY_OVERHEAD + (extraProperties != null ? extraProperties.getEncodeSize() : 0) + applicationPropertiesSize * 2 + applicationPropertiesCount * DataConstants.SIZE_INT;
}
return memoryEstimate;
}
@@ -637,7 +659,16 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
@Override
public Persister<Message> getPersister() {
- return AMQPLargeMessagePersister.getInstance();
+ return getWireCompatiblePersister(EmbedMessageUtil.getDefaultWireVersion());
+ }
+
+ @Override
+ public Persister<Message> getWireCompatiblePersister(int embedWireVersion) {
+ if (embedWireVersion == EmbedMessageUtil.EMBED_WIRE_VERSION_1) {
+ return AMQPLargeMessagePersister.getInstance();
+ } else {
+ 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 7fb4b6d41f..4f23258a8a 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 class AMQPLargeMessagePersister extends MessagePersister {
}
}
- /**
- * Sub classes must add the first short as the protocol-id
- */
@Override
public void encode(ActiveMQBuffer buffer, Message record) {
super.encode(buffer, record);
@@ -105,11 +102,11 @@ public class AMQPLargeMessagePersister extends MessagePersister {
long format = buffer.readLong();
SimpleString address = buffer.readNullableSimpleString();
- int size = buffer.readInt();
+ int extraPropertiesEncodingSize = buffer.readInt();
TypedProperties properties;
- if (size != 0) {
+ if (extraPropertiesEncodingSize != 0) {
properties = new TypedProperties(Message.INTERNAL_PROPERTY_NAMES_PREDICATE);
properties.decode(buffer.byteBuf());
} else {
@@ -118,6 +115,7 @@ public class AMQPLargeMessagePersister extends MessagePersister {
AMQPLargeMessage largeMessage = new AMQPLargeMessage(id, format, properties, null, null);
+ largeMessage.reloadSetDurable(durable);
largeMessage.setFileDurable(durable);
if (address != null) {
largeMessage.setAddress(address);
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 0000000000..8a9024d761
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersisterV2.java
@@ -0,0 +1,111 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+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();
+ }
+
+ int size;
+
+ @Override
+ public int getEncodeSize(Message record) {
+ AMQPLargeMessage msgEncode = (AMQPLargeMessage) record;
+ ByteBuf buf = msgEncode.getSavedEncodeBuffer();
+
+ try {
+ size = buf.writerIndex() + getMetadataSize(record) + DataConstants.SIZE_BYTE;
+ return size;
+ } finally {
+ msgEncode.releaseEncodedBuffer();
+ }
+ }
+
+ /** Write persister data using Map like boundaries from MapPersister */
+ protected void writeMessageMetadata(ActiveMQBuffer buffer, Message record) {
+ AMQPMessageMetadataPersister.getInstance().encode(buffer, record);
+ }
+
+ protected int getMetadataSize(Message record) {
+ return AMQPMessageMetadataPersister.getInstance().getEncodeSize(record);
+ }
+
+ @Override
+ public void encode(ActiveMQBuffer buffer, Message record) {
+ AMQPLargeMessage msgEncode = (AMQPLargeMessage) record;
+ writePersisterID(buffer);
+ writeMessageMetadata(buffer, record);
+
+ ByteBuf savedEncodeBuffer = msgEncode.getSavedEncodeBuffer();
+ buffer.writeBytes(savedEncodeBuffer, 0, savedEncodeBuffer.writerIndex());
+ msgEncode.releaseEncodedBufferAfterWrite();
+ }
+
+ @Override
+ public Message decode(ActiveMQBuffer buffer, Message record, CoreMessageObjectPools pool) {
+ AMQPMessageMetadataPersister metadataPersister = AMQPMessageMetadataPersister.getInstance();
+
+ AMQPMetadataDecodingState decodingMetaData = new AMQPMetadataDecodingState();
+ metadataPersister.decode(buffer, decodingMetaData);
+
+ assert decodingMetaData.memoryEstimate != 0;
+
+ AMQPLargeMessage largeMessage = new AMQPLargeMessage(decodingMetaData.messageID, decodingMetaData.messageFormat, decodingMetaData.extraProperties, null, null);
+
+ largeMessage.reloadSetDurable(decodingMetaData.isDurable);
+ largeMessage.setFileDurable(decodingMetaData.isDurable);
+ if (decodingMetaData.address != null) {
+ largeMessage.setAddress(decodingMetaData.address);
+ }
+
+ largeMessage.readSavedEncoding(buffer.byteBuf());
+
+ largeMessage.reloadExpiration(decodingMetaData.messageExpiration);
+ largeMessage.setReencoded(decodingMetaData.isReencoded);
+ largeMessage.setMemoryEstimate(decodingMetaData.memoryEstimate);
+ largeMessage.reloadPriority(decodingMetaData.priority);
+
+ return largeMessage;
+ }
+
+}
\ No newline at end of file
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 253943aa20..ab02dc9e57 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,13 +44,13 @@ import org.apache.activemq.artemis.api.core.RoutingType;
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;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageIdHelper;
import org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSupport;
import org.apache.activemq.artemis.protocol.amqp.converter.AmqpCoreConverter;
+import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolLogger;
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;
@@ -83,6 +83,9 @@ import org.apache.qpid.proton.message.impl.MessageImpl;
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,20 @@ import static org.apache.activemq.artemis.protocol.amqp.converter.AMQPMessageSup
*/
public abstract class AMQPMessage extends RefCountMessage implements org.apache.activemq.artemis.api.core.Message {
+ /**
+ * Minimum memory footprint in bytes for any AMQP message instance.
+ * <p>
+ * This estimate accounts for the complete in-memory overhead including object instance size,
+ * member variables, and associated data structures (e.g., queue references, internal buffers).
+ * <p>
+ * The value is empirically derived from AMQPGlobalMaxTest#testValidateMemoryEstimateAMQP,
+ * which validates memory exhaustion behavior by sending minimal messages until OOM.
+ */
+ public static final int BASE_MEMORY_OVERHEAD = 1300;
+
+ private static final byte DURABLE = 1;
+ private static final byte NON_DURABLE = 0;
+
private static final SimpleString ANNOTATION_AREA_PREFIX = SimpleString.of("m.");
protected static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -146,7 +163,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 +215,23 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
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 dropped;
protected boolean isPaged;
@@ -282,6 +304,11 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
this.messageDataScanned = copy.messageDataScanned;
}
+ void setMemoryEstimate(int memoryEstimate) {
+ this.memoryEstimate = memoryEstimate;
+ }
+
+
private static MessageAnnotations copyAnnotations(MessageAnnotations messageAnnotations) {
if (messageAnnotations == null) {
return null;
@@ -558,36 +585,14 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
// 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")
@@ -673,9 +678,6 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
case NOT_SCANNED:
scanMessageData();
break;
- case RELOAD_PERSISTENCE:
- lazyScanAfterReloadPersistence();
- break;
case SCANNED:
// NO-OP
break;
@@ -689,17 +691,18 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
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;
@@ -754,7 +757,15 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
// 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();
+ int apCount = parseApCountAndSkip(data);
+ if (apCount >= 0) {
+ this.applicationPropertiesCount = apCount;
+ } else {
+ // if parseApCountAndSkip fails for an invalid message, we will try the skipValue from the original constructor
+ this.applicationPropertiesCount = 0;
+ data.position(constructorPos);
+ constructor.skipValue();
+ }
remainingBodyPosition = data.hasRemaining() ? data.position() : VALUE_NOT_PRESENT;
break;
} else {
@@ -771,6 +782,56 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
this.messageDataScanned = MessageDataScanningStatus.SCANNED.code;
}
+ public int getApplicationPropertiesCount() {
+ ensureScanning();
+ return applicationPropertiesCount;
+ }
+
+ /**
+ * @return the number of application properties in the encoding, or -1 if the encoding is invalid
+ */
+ private int parseApCountAndSkip(ReadableBuffer data) {
+ byte encodingCode = data.get();
+ int count;
+ int size;
+ switch (encodingCode) {
+ case EncodingCodes.MAP8 -> {
+ size = data.get() & 0xff;
+ if (size > data.limit()) {
+ ActiveMQAMQPProtocolLogger.LOGGER.invalidEncodingApplicationProperties(messageID, "Invalid size on encoding");
+ return -1;
+ }
+ int startPosition = data.position();
+ count = data.get() & 0xff;
+ data.position(startPosition + size);
+ }
+ case EncodingCodes.MAP32 -> {
+ size = data.getInt();
+ if (size < 0 || size > data.limit()) {
+ ActiveMQAMQPProtocolLogger.LOGGER.invalidEncodingApplicationProperties(messageID, "Invalid size on encoding");
+ return -1;
+ }
+ int startPosition = data.position();
+ count = data.getInt();
+ data.position(startPosition + size);
+ }
+ case EncodingCodes.NULL -> {
+ return 0;
+ }
+ default -> {
+ ActiveMQAMQPProtocolLogger.LOGGER.invalidEncodingApplicationProperties(messageID, "Invalid encoding type");
+ return -1;
+ }
+ }
+
+ if (count < 0 || count % 2 != 0) {
+ ActiveMQAMQPProtocolLogger.LOGGER.invalidEncodingApplicationProperties(messageID, "invalid properties count");
+ return -1;
+ }
+
+ return count / 2;
+ }
+
@Override
public abstract org.apache.activemq.artemis.api.core.Message copy();
@@ -895,16 +956,6 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
@Override
public abstract int getMemoryEstimate();
- @Override
- public int getOriginalEstimate() {
- if (originalEstimate < 0) {
- // getMemoryEstimate should initialize originalEstimate
- return getMemoryEstimate();
- } else {
- return originalEstimate;
- }
- }
-
@Override
public Map<String, Object> toPropertyMap(int valueSizeLimit) {
return toPropertyMap(false, valueSizeLimit);
@@ -1044,16 +1095,6 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
@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;
@@ -1232,13 +1273,18 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
@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) {
+ reloadSetDurable(header.getDurable());
+ } else {
+ this.durable = NON_DURABLE;
+ }
}
+ return this.durable == DURABLE;
+ }
+
+ void reloadSetDurable(boolean booleanDurable) {
+ this.durable = booleanDurable ? DURABLE : NON_DURABLE;
}
@Override
@@ -1248,6 +1294,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
}
header.setDurable(durable); // Message needs to be re-encoded following this action.
+ reloadSetDurable(durable);
return this;
}
@@ -1320,9 +1367,20 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
@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
@@ -1334,6 +1392,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
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/AMQPMessageMetadataPersister.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageMetadataPersister.java
new file mode 100644
index 0000000000..a723819531
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessageMetadataPersister.java
@@ -0,0 +1,193 @@
+/*
+ * 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.api.core.SimpleString;
+import org.apache.activemq.artemis.utils.AbstractMapPersister;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Supports the AMQPMessagePersisters ({@link AMQPLargeMessagePersisterV2} and {@link AMQPMessagePersisterV4}) by providing hashmap-like encodings
+ * on data used by the broker during reloading of AMQP messages. */
+public class AMQPMessageMetadataPersister extends AbstractMapPersister<AMQPMetadataDecodingState> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final AMQPMessageMetadataPersister theInstance = new AMQPMessageMetadataPersister();
+
+ public static AMQPMessageMetadataPersister getInstance() {
+ return theInstance;
+ }
+
+ protected static final short KEY_MESSAGE_ID = 1;
+ protected static final short KEY_MESSAGE_FORMAT = 2;
+ protected static final short KEY_EXPIRATION = 3;
+ protected static final short KEY_MEMORY_ESTIMATE = 4;
+ protected static final short KEY_PRIORITY = 5;
+ protected static final short KEY_ADDRESS = 6;
+ protected static final short KEY_IS_DURABLE = 7;
+ protected static final short KEY_EXTRA_PROPERTIES = 8;
+ protected static final short KEY_IS_REENCODED = 9; // used on large messages only
+
+ // This persister cannot generate more elements than defined keys.
+ // Currently, KEY_IS_REENCODED is the last defined key (9), with KEY_MESSAGE_ID being the first (1).
+ // Reading beyond 9 elements indicates either invalid/malicious data or a message from a future version.
+ // We set a reasonable limit to support future versions while protecting the decoder from malformed data.
+ protected static final short MAX_ELEMENTS = 100;
+
+ @Override
+ protected int getMaxAllowedElements() {
+ return MAX_ELEMENTS;
+ }
+
+ /** This is the minimum size this codec will generate, as these will always be persisted */
+ private static final int BASIC_SIZE = headerSize() +
+ payloadSizeLong() + // messageID
+ payloadSizeLong() + // message format
+ payloadSizeLong() + // message Expiration
+ payloadSizeInteger() + // memory estimate
+ payloadSizeByte() + // priority
+ payloadSizeBoolean(); // isDurable
+
+ private static final int BASIC_ELEMENTS = 6; // messageID, messageFormat, messageExpiration, memoryEstimate, priority, isDurable and address
+
+ public int getEncodeSize(Message record) {
+ AMQPMessage amqpMessage = (AMQPMessage) record;
+ int encodeSize = BASIC_SIZE;
+
+ if (amqpMessage.getAddressSimpleString() != null) {
+ encodeSize += payloadSizeByteArray(amqpMessage.getAddressSimpleString().getData().length);
+ }
+
+ if (record.isLargeMessage()) {
+ encodeSize += payloadSizeBoolean();
+ }
+
+ if (amqpMessage.getExtraProperties() != null) {
+ encodeSize += payloadSizeByteArray(amqpMessage.getExtraProperties().getEncodeSize());
+ }
+
+ return encodeSize;
+ }
+
+ public int getNumberOfElements(Message record) {
+ AMQPMessage amqpMessage = (AMQPMessage) record;
+ int numberOfElements = BASIC_ELEMENTS;
+ if (amqpMessage.getAddressSimpleString() != null) {
+ numberOfElements++;
+ }
+ if (record.isLargeMessage()) {
+ numberOfElements++;
+ }
+ if (amqpMessage.getExtraProperties() != null) {
+ numberOfElements++;
+ }
+ return numberOfElements;
+ }
+
+ public void encode(ActiveMQBuffer buffer, Message record) {
+ AMQPMessage msgEncode = (AMQPMessage) record;
+ int initialPosition = buffer.writerIndex();
+ int recordSize = getEncodeSize(record);
+ writeHeader(buffer, recordSize, getNumberOfElements(record));
+ encodeElements(buffer, msgEncode);
+
+ assert buffer.writerIndex() == initialPosition + recordSize;
+ }
+
+ protected void encodeElements(ActiveMQBuffer buffer, AMQPMessage msgEncode) {
+ writeLong(buffer, KEY_MESSAGE_ID, msgEncode.getMessageID());
+ writeLong(buffer, KEY_MESSAGE_FORMAT, msgEncode.getMessageFormat());
+ writeLong(buffer, KEY_EXPIRATION, msgEncode.getExpiration());
+ writeInteger(buffer, KEY_MEMORY_ESTIMATE, msgEncode.getMemoryEstimate());
+ writeByte(buffer, KEY_PRIORITY, msgEncode.getPriority());
+ if (msgEncode.getAddressSimpleString() != null) {
+ writeByteArray(buffer, KEY_ADDRESS, msgEncode.getAddressSimpleString().getData());
+ }
+ writeBoolean(buffer, KEY_IS_DURABLE, msgEncode.isDurable());
+ TypedProperties extraProperties = msgEncode.getExtraProperties();
+ if (extraProperties != null) {
+ writeByteArray(buffer, KEY_EXTRA_PROPERTIES, extraProperties.getEncodeSize(), extraProperties::encode);
+ }
+ if (msgEncode.isLargeMessage()) {
+ writeBoolean(buffer, KEY_IS_REENCODED, ((AMQPLargeMessage)msgEncode).isReencoded());
+ }
+ }
+
+ @Override
+ protected void onMapReadString(short key, String value, AMQPMetadataDecodingState decodingObject) {
+ }
+
+ @Override
+ public void onMapReadInteger(short key, int value, AMQPMetadataDecodingState metaData) {
+ switch (key) {
+ case KEY_MEMORY_ESTIMATE -> metaData.memoryEstimate = value;
+ default -> logger.debug("Unknown Integer key={} value={} - ignoring", key, value);
+ }
+ }
+
+ @Override
+ public void onMapReadByte(short key, byte value, AMQPMetadataDecodingState metaData) {
+ switch (key) {
+ case KEY_PRIORITY -> metaData.priority = value;
+ default -> logger.debug("Unknown Byte key={} value={} - ignoring", key, value);
+ }
+ }
+
+ @Override
+ public void onMapReadBoolean(short key, boolean value, AMQPMetadataDecodingState metaData) {
+ switch (key) {
+ case KEY_IS_DURABLE -> metaData.isDurable = value;
+ case KEY_IS_REENCODED -> metaData.isReencoded = value;
+ default -> logger.debug("Unknown Boolean key={} value={} - ignoring", key, value);
+ }
+ }
+
+ @Override
+ public void onMapReadLong(short key, long value, AMQPMetadataDecodingState metaData) {
+ switch (key) {
+ case KEY_MESSAGE_ID -> metaData.messageID = value;
+ case KEY_MESSAGE_FORMAT -> metaData.messageFormat = value;
+ case KEY_EXPIRATION -> metaData.messageExpiration = value;
+ default -> logger.debug("Unknown Long key={} value={} - ignoring", key, value);
+ }
+ }
+
+ @Override
+ protected void onMapReadByteArray(short key, ActiveMQBuffer slice, AMQPMetadataDecodingState metaData) {
+ switch (key) {
+ case KEY_ADDRESS -> {
+ byte[] addressBytes = new byte[slice.readableBytes()];
+ slice.readBytes(addressBytes);
+ metaData.address = SimpleString.of(addressBytes);
+ }
+ case KEY_EXTRA_PROPERTIES -> {
+ metaData.extraProperties = new TypedProperties(Message.INTERNAL_PROPERTY_NAMES_PREDICATE);
+ metaData.extraProperties.decode(slice.byteBuf());
+ }
+ default -> logger.debug("Unknown byteArray key={} value={} - ignoring", key, slice);
+ }
+ }
+}
+
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 267ec290a2..8c89ec9713 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 @@ public class AMQPMessagePersister extends MessagePersister {
if (address != null) {
record.setAddress(address);
}
+
+ if (ID == getID()) {
+ scanAfterReload((AMQPStandardMessage) record);
+ }
return record;
}
+
+ // notice this is muted in AMQPMessagePersisterV4 and after
+ // we will scan the message after reloading for older versions, while in V4 and after 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 7d599d1211..3034a604af 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 class AMQPMessagePersisterV2 extends AMQPMessagePersister {
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 b3d86d055e..9878e8cdb2 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 class AMQPMessagePersisterV3 extends AMQPMessagePersisterV2 {
}
- /**
- * 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 @@ public class AMQPMessagePersisterV3 extends AMQPMessagePersisterV2 {
((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 0000000000..aec9d555c0
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV4.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.AMQPMessagePersisterV4_ID;
+
+/**
+ * V4 adds a size field to determine persister boundaries, enabling forward-compatible
+ * extensions without additional versioning.
+ **/
+public class AMQPMessagePersisterV4 extends AMQPMessagePersisterV3 {
+
+ public static final byte ID = AMQPMessagePersisterV4_ID;
+
+ public static AMQPMessagePersisterV4 theInstance;
+
+ public static AMQPMessagePersisterV4 getInstance() {
+ if (theInstance == null) {
+ theInstance = new AMQPMessagePersisterV4();
+ }
+ return theInstance;
+ }
+
+ public AMQPMessagePersisterV4() {
+ super();
+ }
+
+ @Override
+ public byte getID() {
+ return ID;
+ }
+
+ @Override
+ public int getEncodeSize(Message record) {
+ return DataConstants.SIZE_BYTE + record.getPersistSize() + getMetadataSize(record);
+ }
+
+ /** Write persister data using Map like boundaries from MapPersister */
+ protected void writeMessageMetadata(ActiveMQBuffer buffer, Message record) {
+ AMQPMessageMetadataPersister.getInstance().encode(buffer, record);
+ }
+
+ protected int getMetadataSize(Message record) {
+ return AMQPMessageMetadataPersister.getInstance().getEncodeSize(record);
+ }
+
+ @Override
+ public void encode(ActiveMQBuffer buffer, Message record) {
+ writePersisterID(buffer);
+
+ writeMessageMetadata(buffer, record);
+
+ record.persist(buffer);
+ }
+
+ @Override
+ public Message decode(ActiveMQBuffer buffer, Message record, CoreMessageObjectPools pool) {
+ AMQPMessageMetadataPersister metadataPersister = AMQPMessageMetadataPersister.getInstance();
+ AMQPMetadataDecodingState decodingMetaData = new AMQPMetadataDecodingState();
+
+ metadataPersister.decode(buffer, decodingMetaData);
+
+ assert decodingMetaData.memoryEstimate > 0;
+
+ AMQPStandardMessage standardMessage = new AMQPStandardMessage(decodingMetaData.messageFormat);
+ standardMessage.reloadPersistence(buffer, pool);
+ if (decodingMetaData.extraProperties != null) {
+ standardMessage.setExtraProperties(decodingMetaData.extraProperties);
+ }
+ standardMessage.setMessageID(decodingMetaData.messageID);
+ standardMessage.setAddress(decodingMetaData.address);
+ standardMessage.setMemoryEstimate(decodingMetaData.memoryEstimate);
+ standardMessage.reloadPriority(decodingMetaData.priority);
+ standardMessage.reloadSetDurable(decodingMetaData.isDurable);
+ standardMessage.reloadExpiration(decodingMetaData.messageExpiration);
+ return standardMessage;
+ }
+}
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMetadataDecodingState.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMetadataDecodingState.java
new file mode 100644
index 0000000000..588764fee9
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMetadataDecodingState.java
@@ -0,0 +1,44 @@
+/*
+ * 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.SimpleString;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+
+/**
+ * Temporary holder for AMQP message metadata during decoding.
+ * <p>
+ * When decoding metadata from {@link AMQPMessageMetadataPersister}, certain values
+ * (e.g., messageID on large messages and messageFormat on regular messages) are needed before the Message
+ * object can be created.
+ */
+public class AMQPMetadataDecodingState {
+
+ protected long messageID; // always present
+ protected long messageFormat; // always present
+ protected long messageExpiration; // always present
+ protected int memoryEstimate; // always present
+ protected byte priority; // always present
+ protected SimpleString address; // variable size based on the string size
+ protected boolean isDurable; // always present
+ protected TypedProperties extraProperties; // variable size based on the byte array size
+ protected boolean isReencoded; // used on large messages only
+
+ public AMQPMetadataDecodingState() {
+ }
+}
\ No newline at end of file
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 c514009bce..8fedd7c876 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
@@ -23,11 +23,13 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode;
+import org.apache.activemq.artemis.spi.core.protocol.EmbedMessageUtil;
import org.apache.activemq.artemis.utils.DataConstants;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
import org.apache.qpid.proton.amqp.Symbol;
@@ -39,12 +41,8 @@ import org.apache.qpid.proton.amqp.messaging.Header;
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 +189,27 @@ public class AMQPStandardMessage extends AMQPMessage {
@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 = BASE_MEMORY_OVERHEAD + (data != null ? data.capacity() + getApplicationPropertiesEncodingSize(data) * 2 + getApplicationPropertiesCount() * DataConstants.SIZE_INT : 0);
}
return memoryEstimate;
}
+ private int getApplicationPropertiesEncodingSize(ReadableBuffer data) {
+ ensureScanning();
+
+ // if still VALUE_NOT_PRESENT, it means we have no application properties, we must return 0
+ if (applicationPropertiesPosition == VALUE_NOT_PRESENT) {
+ return 0;
+ } else if (remainingBodyPosition != VALUE_NOT_PRESENT) {
+ return remainingBodyPosition - applicationPropertiesPosition;
+ } else {
+ return data.capacity() - applicationPropertiesPosition;
+ }
+ }
+
@Override
public void persist(ActiveMQBuffer targetRecord) {
@@ -237,79 +242,9 @@ public class AMQPStandardMessage extends AMQPMessage {
// 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 +254,16 @@ public class AMQPStandardMessage extends AMQPMessage {
@Override
public Persister<org.apache.activemq.artemis.api.core.Message> getPersister() {
- return AMQPMessagePersisterV3.getInstance();
+ return getWireCompatiblePersister(EmbedMessageUtil.getDefaultWireVersion());
+ }
+
+ @Override
+ public Persister<Message> getWireCompatiblePersister(int embedWireVersion) {
+ if (embedWireVersion == EmbedMessageUtil.EMBED_WIRE_VERSION_1) {
+ return AMQPMessagePersisterV3.getInstance();
+ } else {
+ 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 faa9039338..314b5d41ae 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<Message>[] 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/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java
index 890cd48664..4b6c7f9b77 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java
@@ -88,4 +88,7 @@ public interface ActiveMQAMQPProtocolLogger {
@LogMessage(id = 111014, value = "AMQP Federation target skipped federation of queue {}, reason={}", level = LogMessage.Level.WARN)
void federationTargetSkippedQueueFederation(String address, String reason);
+
+ @LogMessage(id = 111015, value = "AMQP message {} has invalid application properties encoding. {}", level = LogMessage.Level.WARN)
+ void invalidEncodingApplicationProperties(long messageID, String description);
}
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 51228637f4..afc88353b2 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
@@ -19,6 +19,7 @@ package org.apache.activemq.artemis.protocol.amqp.broker;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import java.lang.invoke.MethodHandles;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.HashMap;
@@ -87,7 +88,6 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals;
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 +216,11 @@ public class AMQPMessageTest {
// 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 +249,11 @@ public class AMQPMessageTest {
// 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 +279,11 @@ public class AMQPMessageTest {
// 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 +331,7 @@ public class AMQPMessageTest {
}
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 +1642,16 @@ public class AMQPMessageTest {
}
@Test
- public void testExtraProperty() {
+ public void testJournalPersistenceWireVersion2() {
+ internalTestPersistence(EmbedMessageUtil.EMBED_WIRE_VERSION_2);
+ }
+
+ @Test
+ public void testJournalPersistenceWireVersion1() {
+ internalTestPersistence(EmbedMessageUtil.EMBED_WIRE_VERSION_1);
+ }
+
+ private void internalTestPersistence(int wireVersion) {
MessageImpl protonMessage = (MessageImpl) Message.Factory.create();
byte[] original = RandomUtil.randomBytes();
@@ -1655,25 +1659,30 @@ public class AMQPMessageTest {
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();
}
{
- ICoreMessage embeddedMessage = EmbedMessageUtil.embedAsCoreMessage(decoded);
+ ICoreMessage embeddedMessage = EmbedMessageUtil.embedAsCoreMessage(decoded, wireVersion);
AMQPStandardMessage readMessage = (AMQPStandardMessage) EmbedMessageUtil.extractEmbedded(embeddedMessage, null);
assertEquals(33, readMessage.getMessageID());
assertEquals("someAddress", readMessage.getAddress());
@@ -1681,6 +1690,78 @@ public class AMQPMessageTest {
}
}
+ @Test
+ public void testMixEncodingNoApplicationProperties() {
+
+ boolean[] booleanValues = new boolean[] {false, true};
+
+ // play with permutations
+ for (boolean a : booleanValues) {
+ for (boolean b : booleanValues) {
+ for (boolean c : booleanValues) {
+ internalMessageNoApplicationProperteis(a, b, c);
+ }
+ }
+ }
+ }
+
+ private void internalMessageNoApplicationProperteis(boolean useHeader, boolean useAnnotations, boolean useBody) {
+ MessageImpl message = (MessageImpl) Proton.message();
+
+ if (useHeader) {
+ message.setHeader(new Header());
+ }
+
+ if (useAnnotations) {
+ MessageAnnotations annotations = new MessageAnnotations(new LinkedHashMap<>());
+ for (int i = 0; i < 100; ++i) {
+ annotations.getValue().put(Symbol.valueOf(TEST_MESSAGE_ANNOTATION_KEY + ":" + i), TEST_MESSAGE_ANNOTATION_VALUE + ":" + i);
+ }
+ message.setMessageAnnotations(annotations);
+ } else {
+ message.setMessageAnnotations(null);
+ }
+
+ if (useBody) {
+ message.setBody(new AmqpValue(TEST_STRING_BODY));
+ } else {
+ message.setBody(null);
+ }
+
+ byte[] encodedProtonMessage = encodeMessage(message);
+
+ AMQPStandardMessage decoded = new AMQPStandardMessage(0, encodedProtonMessage, null, null);
+
+ final int expectedEstimate = AMQPMessage.BASE_MEMORY_OVERHEAD + encodedProtonMessage.length;
+ assertEquals(expectedEstimate, decoded.getMemoryEstimate());
+ }
+
+
+ @Test
+ public void testNonSenseNullApplicationProperties() {
+ ByteBuffer buffer = ByteBuffer.allocate(4);
+
+ // This is simulating a nonsense encoding data.
+ // This is effectively encoding new ApplicationProperties(null)
+ byte[] encoded;
+ {
+ buffer.put(EncodingCodes.DESCRIBED_TYPE_INDICATOR);
+ buffer.put(EncodingCodes.SMALLULONG);
+ buffer.put((byte) 0x74);
+
+ buffer.put(EncodingCodes.NULL);
+
+ encoded = new byte[buffer.position()];
+ buffer.position(0);
+ buffer.get(encoded);
+ }
+
+ AMQPStandardMessage decoded = new AMQPStandardMessage(0, encoded, null, null);
+ assertEquals(0, decoded.getApplicationPropertiesCount());
+ decoded.getMemoryEstimate(); // calling this method to make sure it won't throw exceptions
+
+ }
+
//----- Test that message decode ignores unused sections ------------------//
private static final UnsignedLong AMQPVALUE_DESCRIPTOR = UnsignedLong.valueOf(0x0000000000000077L);
@@ -2886,6 +2967,7 @@ public class AMQPMessageTest {
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 87fc94e636..bb7a7266bd 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 class TestConversions {
Map<String, Object> 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 class TestConversions {
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 class TestConversions {
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/cluster/impl/BridgeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/BridgeImpl.java
index 73ecc2e271..c6c99e47ff 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/BridgeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/BridgeImpl.java
@@ -71,6 +71,7 @@ import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.server.management.NotificationService;
import org.apache.activemq.artemis.core.server.transformer.Transformer;
+import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
import org.apache.activemq.artemis.spi.core.protocol.EmbedMessageUtil;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
@@ -154,6 +155,8 @@ public class BridgeImpl implements Bridge, SessionFailureListener, SendAcknowled
private final Lock statusLock = new ReentrantLock();
+ private volatile int embedWireVersion = EmbedMessageUtil.EMBED_WIRE_VERSION_2;
+
public BridgeImpl(final ServerLocatorInternal serverLocator,
final BridgeConfiguration configuration,
final UUID nodeUUID,
@@ -537,6 +540,24 @@ public class BridgeImpl implements Bridge, SessionFailureListener, SendAcknowled
return beforeForwardingNoCopy(message, forwardingAddress);
}
+ private void updateEmbedWireVersion() {
+ ClientSessionInternal sessionToUse = session;
+ if (sessionToUse != null) {
+ RemotingConnection connection = sessionToUse.getConnection();
+ if (connection instanceof CoreRemotingConnection coreConnection) {
+ if (coreConnection.isBeforeEmbedWireVersion2()) {
+ embedWireVersion = EmbedMessageUtil.EMBED_WIRE_VERSION_1;
+ return;
+ }
+ }
+ }
+ embedWireVersion = EmbedMessageUtil.EMBED_WIRE_VERSION_2;
+ }
+
+ private int getEmbedWireVersion() {
+ return embedWireVersion;
+ }
+
/**
* ClusterConnectionBridge already makes a copy of the message. So I needed I hook where the message is not copied.
*/
@@ -574,9 +595,9 @@ public class BridgeImpl implements Bridge, SessionFailureListener, SendAcknowled
if (transformedMessage != message) {
logger.debug("The transformer {} made a copy of the message {} as transformedMessage", transformer, message);
}
- return EmbedMessageUtil.embedAsCoreMessage(transformedMessage);
+ return EmbedMessageUtil.embedAsCoreMessage(transformedMessage, getEmbedWireVersion());
} else {
- return EmbedMessageUtil.embedAsCoreMessage(message);
+ return EmbedMessageUtil.embedAsCoreMessage(message, getEmbedWireVersion());
}
}
@@ -1085,6 +1106,8 @@ public class BridgeImpl implements Bridge, SessionFailureListener, SendAcknowled
session = (ClientSessionInternal) csf.createSession(configuration.getUser(), configuration.getPassword(), false, true, true, true, 1, configuration.getClientId());
session.getProducerCreditManager().setCallback(BridgeImpl.this);
sessionConsumer = (ClientSessionInternal) csf.createSession(configuration.getUser(), configuration.getPassword(), false, true, true, true, 1, configuration.getClientId());
+ // Update the wire version based on the newly created session
+ updateEmbedWireVersion();
}
if (configuration.getForwardingAddress() != null) {
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 9b23fd7e48..870276f147 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
@@ -776,7 +776,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
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);
@@ -794,7 +794,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
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/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/EmbedMessageUtil.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/EmbedMessageUtil.java
index 82974cd799..e287828690 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/EmbedMessageUtil.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/EmbedMessageUtil.java
@@ -27,6 +27,7 @@ import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl;
import org.apache.activemq.artemis.core.server.LargeServerMessage;
+import org.apache.activemq.artemis.utils.SystemPropertyHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
@@ -34,13 +35,27 @@ import java.lang.invoke.MethodHandles;
public class EmbedMessageUtil {
+ private static int DEFAULT_WIRE_VERSION = Integer.parseInt(SystemPropertyHelper.getProperty("org.apache.artemis.amqp.embed.wire.version", "ARTEMIS_AMQP_EMBED_WIRE_VERSION", "2"));
+
+ public static int getDefaultWireVersion() {
+ return DEFAULT_WIRE_VERSION;
+ }
+
+ public static void setDefaultWireVersion(int wireVersion) {
+ DEFAULT_WIRE_VERSION = wireVersion;
+ }
+
private static final String AMQP_ENCODE_PROPERTY = "_AMQP_EMBED_LARGE";
private static final byte[] signature = new byte[]{(byte) 'E', (byte) 'M', (byte) 'B'};
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- public static ICoreMessage embedAsCoreMessage(Message source) {
+ // EMBED_WIRE_VERSION_2 introduced AMQPMessagePersisterV4 with map-based metadata encoding
+ public static final int EMBED_WIRE_VERSION_2 = 2;
+ public static final int EMBED_WIRE_VERSION_1 = 1;
+
+ public static ICoreMessage embedAsCoreMessage(Message source, int embedWireVersion) {
if (source instanceof ICoreMessage message) {
return message;
@@ -51,16 +66,19 @@ public class EmbedMessageUtil {
LargeServerMessageImpl largeServerMessage = new LargeServerMessageImpl(Message.LARGE_EMBEDDED_TYPE, source.getMessageID(), largeSource.getStorageManager(), largeSource.getLargeBody().createFile());
largeServerMessage.setDurable(source.isDurable());
- int size = source.getPersister().getEncodeSize(source);
+
+ Persister<Message> messagePersister = source.getWireCompatiblePersister(embedWireVersion);
+
+ int size = messagePersister.getEncodeSize(source);
byte[] arrayByte = new byte[size];
ActiveMQBuffer buffer = ActiveMQBuffers.wrappedBuffer(arrayByte);
buffer.resetWriterIndex();
- source.getPersister().encode(buffer, source);
+ messagePersister.encode(buffer, source);
largeServerMessage.toMessage().putBytesProperty(AMQP_ENCODE_PROPERTY, arrayByte);
largeServerMessage.setParentRef((RefCountMessage)source);
return (ICoreMessage) largeServerMessage.toMessage();
} else {
- Persister persister = source.getPersister();
+ Persister persister = source.getWireCompatiblePersister(embedWireVersion);
CoreMessage message = new CoreMessage(source.getMessageID(), persister.getEncodeSize(source) + signature.length + CoreMessage.BODY_OFFSET).setType(Message.EMBEDDED_TYPE);
message.setDurable(source.isDurable());
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
index a2910d6db0..f462e6366a 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java
@@ -108,6 +108,10 @@ public class MessagePersister implements Persister<Message> {
*/
@Override
public void encode(ActiveMQBuffer buffer, Message record) {
+ writePersisterID(buffer);
+ }
+
+ protected void writePersisterID(ActiveMQBuffer buffer) {
buffer.writeByte(getID());
}
diff --git a/docs/user-manual/ha.adoc b/docs/user-manual/ha.adoc
index 51ed658961..711f8ecf37 100644
--- a/docs/user-manual/ha.adoc
+++ b/docs/user-manual/ha.adoc
@@ -469,6 +469,29 @@ Here's a simple example:
</ha-policy>
----
+==== Rolling Upgrade
+
+If you need to upgrade the broker version in a journal replication setup, follow this procedure to avoid downtime:
+
+. **Stop the backup server**, upgrade it to the new version, and start it.
+At this point the backup will synchronize with the primary server.
+. **Wait for synchronization to complete**.
+The backup must be fully synchronized with the primary before proceeding.
+. **Stop the primary server**, upgrade it to the new version, and start it.
+The upgraded primary will now synchronize with the backup and then fail back to become active again.
+
+[NOTE]
+====
+This procedure ensures continuous availability during the upgrade process.
+The backup takes over when the primary is stopped for upgrade, and then the primary resumes its active role after synchronization is complete.
+====
+
+[IMPORTANT]
+====
+Always upgrade the backup server first.
+Upgrading the primary first could result in compatibility issues or data synchronization problems.
+====
+
==== Apache ZooKeeper Integration
The default pluggable lock manager implementation uses https://curator.apache.org/[Apache Curator] to integrate with https://zookeeper.apache.org/[Apache ZooKeeper].
diff --git a/docs/user-manual/upgrading.adoc b/docs/user-manual/upgrading.adoc
index aacbd1dbfe..541e8cf75a 100644
--- a/docs/user-manual/upgrading.adoc
+++ b/docs/user-manual/upgrading.adoc
@@ -20,6 +20,8 @@ It also simplifies updating to newer versions.
[IMPORTANT]
====
Review the xref:versions.adoc#version-history[version] history for specific steps required for your upgrade before proceeding.
+
+If you are using journal replication for high availability, follow the xref:ha.adoc#rolling-upgrade[rolling upgrade procedure] to upgrade without downtime.
====
In general, the process is as follows:
diff --git a/docs/user-manual/versions.adoc b/docs/user-manual/versions.adoc
index 113f2e8914..31c23d2191 100644
--- a/docs/user-manual/versions.adoc
+++ b/docs/user-manual/versions.adoc
@@ -18,10 +18,24 @@ NOTE: Follow the general upgrade procedure outlined in the xref:upgrading.adoc#u
== Version 2.58.0
=== Highlights
+
* https://issues.apache.org/jira/browse/ARTEMIS-6242[ARTEMIS-6242] / https://issues.apache.org/jira/browse/ARTEMIS-6216[ARTEMIS-6216] - CLI table output (`queue stat`, `lock list`, `print-data`) now uses Unicode box-drawing characters by default.
** Use `--ascii` for plain ASCII output.
+* AMQP message size estimation has been improved to account for the full message metadata, fixing underestimates that occurred when messages carried large application properties.
=== Upgrading from 2.57.0
+
+* Due to https://issues.apache.org/jira/browse/ARTEMIS-5573[ARTEMIS-5573] and https://issues.apache.org/jira/browse/ARTEMIS-5975[ARTEMIS-5975], AMQP message size calculations have been corrected.
+If you previously inflated your sizing configuration to compensate for the underestimation, you may want to revisit those settings after upgrading.
+* If using journal replication, we recommend following the xref:ha.adoc#rolling-upgrade[rolling upgrade procedure] to ensure the primary and backup brokers negotiate the correct wire format automatically.
++
+If a rolling upgrade is not possible, you can force the broker to use the previous AMQP embed wire format by setting the following to `1` (default is `2`):
++
+** System property: `org.apache.artemis.amqp.embed.wire.version`
+** Environment variable: `ARTEMIS_AMQP_EMBED_WIRE_VERSION`
++
+Remove this override once all brokers in the cluster have been upgraded.
+
== Version 2.57.0
=== Highlights
diff --git a/pom.xml b/pom.xml
index 15d80ca0ce..c28db72595 100644
--- a/pom.xml
+++ b/pom.xml
@@ -211,7 +211,7 @@
<activemq.version.majorVersion>1</activemq.version.majorVersion>
<activemq.version.minorVersion>0</activemq.version.minorVersion>
<activemq.version.microVersion>0</activemq.version.microVersion>
- <activemq.version.incrementingVersion>136,135,134,133,132,131,130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
+ <activemq.version.incrementingVersion>137,136,135,134,133,132,131,130,129,128,127,126,125,124,123,122</activemq.version.incrementingVersion>
<activemq.version.versionTag>${project.version}</activemq.version.versionTag>
<ActiveMQ-Version>${project.version}(${activemq.version.incrementingVersion})</ActiveMQ-Version>
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 4f11d7c7d3..758d2ca812 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.HashSet;
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 class RealServerTestBase extends ActiveMQTestBase {
}
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<String> 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<String> 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 cb4b96ce5f..11ac4c23c7 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 0161a36e28..10a1432c82 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
@@ -92,39 +92,45 @@ public class JournalCompatibilityTest extends VersionedBase {
}
@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 d8bdc916c2..62936e9507 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
@@ -30,23 +30,31 @@ import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.spi.core.protocol.EmbedMessageUtil;
import org.apache.activemq.artemis.tests.compatibility.base.ClasspathBase;
import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
import org.apache.activemq.artemis.utils.FileUtil;
import org.apache.qpid.jms.JmsConnectionFactory;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
@ExtendWith(ParameterizedTestExtension.class)
public class MultiVersionReplicaTest extends ClasspathBase {
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
private static final String QUEUE_NAME = "MultiVersionReplicaTestQueue";
private final String main;
@@ -58,6 +66,18 @@ public class MultiVersionReplicaTest extends ClasspathBase {
private boolean security;
+ @BeforeAll
+ public static void beforeAll() {
+ System.setProperty("org.apache.artemis.amqp.embed.wire.version", "1");
+ EmbedMessageUtil.setDefaultWireVersion(EmbedMessageUtil.EMBED_WIRE_VERSION_1);
+ }
+
+ @AfterAll
+ public static void afterAll() {
+ System.clearProperty("org.apache.artemis.amqp.embed.wire.version");
+ EmbedMessageUtil.setDefaultWireVersion(EmbedMessageUtil.EMBED_WIRE_VERSION_2);
+ }
+
@Parameters(name = "main={0}, backup={1}, security={2}")
public static Collection getParameters() {
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 0000000000..cec3c88942
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPReloadFromPersistenceTest.java
@@ -0,0 +1,322 @@
+/*
+ * 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.logs.AssertionLoggerHandler;
+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.AMQPMessageMetadataPersister;
+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 {
+
+ long longOnFake = RandomUtil.randomLong();
+
+ @Test
+ public void testPersistCheck() throws Exception {
+ internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, AMQPMessagePersisterV4.getInstance(), AMQPMessagePersisterV4.getInstance());
+ try (AssertionLoggerHandler handler = new AssertionLoggerHandler()) {
+ AssertionLoggerHandler.LogLevel originalLevel = AssertionLoggerHandler.setLevel(AMQPMessageMetadataPersister.class.getName(), AssertionLoggerHandler.LogLevel.TRACE);
+ try {
+ internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, new FakePersisterFromTheFuture(), AMQPMessagePersisterV4.getInstance());
+ internalPersistCheck(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, AMQPMessagePersisterV4.getInstance(), new FakePersisterFromTheFuture());
+ assertTrue(handler.findText(String.valueOf(longOnFake)));
+ } finally {
+ AssertionLoggerHandler.setLevel(AMQPMessageMetadataPersister.class.getName(), originalLevel);
+ }
+ }
+ }
+
+ @Test
+ public void testPersistCheckOldVersion() throws Exception {
+ internalPersistCheck(AMQPMessage.MessageDataScanningStatus.SCANNED, AMQPMessagePersister.getInstance(), AMQPMessagePersister.getInstance());
+ internalPersistCheck(AMQPMessage.MessageDataScanningStatus.SCANNED, AMQPMessagePersisterV2.getInstance(), AMQPMessagePersisterV2.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);
+ }
+
+ // verify that accessing application properties triggers scanning
+ {
+ buffer.readerIndex(1); // first byte is the persister version
+ AMQPStandardMessage amqpStandardMessage = (AMQPStandardMessage) persisterOnRead.decode(buffer, null, null);
+ assertEquals(expectedStatusAfterReload, amqpStandardMessage.getDataScanningStatus());
+ // accessing application properties should trigger scanning
+ assertEquals(77, amqpStandardMessage.getApplicationPropertiesCount());
+ assertEquals(77, amqpStandardMessage.getApplicationProperties().getValue().size());
+ 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 {
+ internalNoAppProperties(RandomUtil.randomUUIDString());
+ internalNoAppProperties(null);
+ }
+
+ private static void internalNoAppProperties(String address) {
+ AMQPStandardMessage originalMessage = AMQPStandardMessage.createMessage(1, 0, SimpleString.of("duh"), null, null, null, null, null, null, null);
+ originalMessage.setAddress(address);
+
+ 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());
+ assertEquals(address, amqpStandardMessage.getAddress());
+ }
+
+ /**
+ * 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);
+
+ try (AssertionLoggerHandler handler = new AssertionLoggerHandler()) {
+ AssertionLoggerHandler.LogLevel originalLevel = AssertionLoggerHandler.setLevel(AMQPMessageMetadataPersister.class.getName(), AssertionLoggerHandler.LogLevel.TRACE);
+ try {
+ internalPersistCheckLargeMessage(new FakeLargePersisterFromTheFuture(), AMQPLargeMessagePersisterV2.getInstance(), true);
+ assertTrue(handler.findText(String.valueOf(longOnFake)));
+ } finally {
+ AssertionLoggerHandler.setLevel(AMQPMessageMetadataPersister.class.getName(), originalLevel);
+ }
+ }
+ }
+
+ private void internalPersistCheckLargeMessage(AMQPLargeMessagePersister persisterOnWrite, AMQPLargeMessagePersister persisterOnRead, boolean checkMemoryEstimate) throws Exception {
+ FakeSequentialFileFactory fakeSequentialFileFactory = new FakeSequentialFileFactory();
+ final StorageManager storageManager = new NullStorageManager() {
+ @Override
+ public SequentialFile createFileForLargeMessage(long messageID, boolean durable) {
+ return fakeSequentialFileFactory.createSequentialFile("message" + 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;
+ }
+
+
+
+ class FakeLargePersisterFromTheFuture extends AMQPLargeMessagePersisterV2 {
+
+ FakeLargePersisterFromTheFuture() {
+ super();
+ }
+
+ @Override
+ protected void writeMessageMetadata(ActiveMQBuffer buffer, Message record) {
+ new FutureMessageMetadata().encode(buffer, record);
+ }
+
+ @Override
+ protected int getMetadataSize(Message record) {
+ return new FutureMessageMetadata().getEncodeSize(record);
+ }
+
+ }
+
+ class FakePersisterFromTheFuture extends AMQPMessagePersisterV4 {
+
+ FakePersisterFromTheFuture() {
+ super();
+ }
+
+ @Override
+ protected void writeMessageMetadata(ActiveMQBuffer buffer, Message record) {
+ new FutureMessageMetadata().encode(buffer, record);
+ }
+
+ @Override
+ protected int getMetadataSize(Message record) {
+ return new FutureMessageMetadata().getEncodeSize(record);
+ }
+
+ }
+
+ class FutureMessageMetadata extends AMQPMessageMetadataPersister {
+
+ private final short ID_FAKE = (short) 101;
+
+ @Override
+ public int getEncodeSize(Message record) {
+ return super.getEncodeSize(record) + payloadSizeLong();
+ }
+
+ @Override
+ public int getNumberOfElements(Message record) {
+ return super.getNumberOfElements(record) + 1;
+ }
+
+ @Override
+ protected void encodeElements(ActiveMQBuffer buffer, AMQPMessage msgEncode) {
+ super.encodeElements(buffer, msgEncode);
+ writeLong(buffer, ID_FAKE, longOnFake);
+ }
+ }
+
+
+
+
+}
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 3e5a0bced9..3c8ae51f00 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 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
}
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<String, Object> properties) throws Exception {
AmqpClient client = createAmqpClient();
AmqpConnection connection = addConnection(client.connect());
try {
@@ -320,6 +324,9 @@ public class AmqpClientTestSupport extends AmqpTestSupport {
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 04124e8fba..f46db3f969 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
@@ -46,6 +46,7 @@ import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.AckReason;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.protocol.amqp.broker.ActiveMQProtonRemotingConnection;
@@ -407,6 +408,8 @@ public class AckManagerTest extends ActiveMQTestBase {
session.commit();
}
+ assertTrue(store.isPaging());
+
ReferenceIDSupplier referenceIDSupplier = new ReferenceIDSupplier(server1);
{
@@ -466,7 +469,10 @@ public class AckManagerTest extends ActiveMQTestBase {
SimpleString QUEUE_NAME = SimpleString.of("queue_" + RandomUtil.randomUUIDString());
- Queue testQueue = server1.createQueue(QueueConfiguration.of(QUEUE_NAME).setRoutingType(RoutingType.ANYCAST));
+ // I need to make sure I read all the messages in memory. Setting maxReadPageMessages will make sure we depage the 200 messages we are going to send
+ server1.getAddressSettingsRepository().addMatch(QUEUE_NAME.toString(), new AddressSettings().setMaxReadPageMessages(200));
+
+ final QueueImpl testQueue = (QueueImpl)server1.createQueue(QueueConfiguration.of(QUEUE_NAME).setRoutingType(RoutingType.ANYCAST));
ConnectionFactory connectionFactory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
@@ -480,7 +486,9 @@ public class AckManagerTest extends ActiveMQTestBase {
session.commit();
testQueue.getPagingStore().startPaging();
}
- producer.send(session.createTextMessage("hello there " + i));
+ TextMessage message = session.createTextMessage("initial batch " + i);
+ message.setIntProperty("initialBatch", i);
+ producer.send(message);
}
session.commit();
}
@@ -526,9 +534,10 @@ public class AckManagerTest extends ActiveMQTestBase {
for (int i = 0; i < 100; i++) {
AmqpMessage message = new AmqpMessage();
message.setAddress(testQueue.getAddress().toString());
- message.setText("hello again " + i);
+ message.setText("second batch " + i);
message.setDeliveryAnnotation(INTERNAL_ID.toString(), server1.getStorageManager().generateID());
message.setDeliveryAnnotation(INTERNAL_DESTINATION.toString(), testQueue.getAddress().toString());
+ message.setApplicationProperty("secondBatch", i);
sender.send(message);
}
// we should not get any credits
@@ -545,6 +554,10 @@ public class AckManagerTest extends ActiveMQTestBase {
AtomicInteger acked = new AtomicInteger(0);
ackManager.pause();
+ // making sure that reading from paging will place all messages in memory
+ // Since the forEach method will be looking on memory references only
+ Wait.assertEquals(200L, testQueue::getNumberOfReferences, 5000, 100);
+
// Adding real deletes, we should still flow control credits
testQueue.forEach(ref -> {
long messageID = ref.getMessageID();
@@ -565,7 +578,8 @@ public class AckManagerTest extends ActiveMQTestBase {
for (int i = 0; i < 100; i++) {
AmqpMessage message = new AmqpMessage();
message.setAddress(testQueue.getAddress().toString());
- message.setText("one of the last 100");
+ message.setText("last batch " + i);
+ message.setApplicationProperty("lastBatch", i);
message.setDeliveryAnnotation(INTERNAL_ID.toString(), server1.getStorageManager().generateID());
message.setDeliveryAnnotation(INTERNAL_DESTINATION.toString(), testQueue.getAddress().toString());
sender.send(message);
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 37e4dd2282..dc17eae16c 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.assertNotNull;
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 @@ import org.junit.jupiter.api.Test;
public class AmqpJournalLoadingTest extends AmqpClientTestSupport {
@Test
- public void durableMessageDataNotScannedOnRestartTest() throws Exception {
- sendMessages(getQueueName(), 1, true);
+ public void durableMessageDataAfterRestart() throws Exception {
+ Map<String, Object> 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 class AmqpJournalLoadingTest extends AmqpClientTestSupport {
List<AMQPMessage> 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 4a7f1f8d28..28a0c3691d 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 class ElasticQueueTest extends ActiveMQTestBase {
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 0000000000..35df4767db
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java
@@ -0,0 +1,357 @@
+/*
+ * 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<Session, Integer, Message> 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 {
+ return s.createMessage();
+ } 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<Session, Integer, Message> messageCreator,
+ BiConsumer<Integer, Message> 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<Session, Integer, Message> messageCreator,
+ BiConsumer<Integer, Message> 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 6949305d89..f12ac9780a 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 @@ public class AmqpPageTest extends PageTest {
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 @@ public class AmqpPageTest extends PageTest {
}
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;
}
}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/ActiveMQBufferTestBase.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/ActiveMQBufferTestBase.java
index 0f3ca2465b..b8003f9a95 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/ActiveMQBufferTestBase.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/ActiveMQBufferTestBase.java
@@ -21,6 +21,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.SimpleString;
@@ -257,6 +258,73 @@ public abstract class ActiveMQBufferTestBase extends ActiveMQTestBase {
assertEquals(65535, s2);
}
+ @Test
+ public void testWriteUnsignedShort() throws Exception {
+ // Test writing 0
+ wrapper.writeUnsignedShort(0);
+ assertEquals(0, wrapper.readUnsignedShort());
+
+ // Test writing max unsigned short value (65535)
+ wrapper.writeUnsignedShort(65535);
+ assertEquals(65535, wrapper.readUnsignedShort());
+
+ // Test writing mid-range value
+ wrapper.writeUnsignedShort(32768);
+ assertEquals(32768, wrapper.readUnsignedShort());
+
+ // Test writing value that would be negative as signed short
+ wrapper.writeUnsignedShort(40000);
+ assertEquals(40000, wrapper.readUnsignedShort());
+
+ }
+
+ // this is how unsigned shorts used to be written.
+ // this is pretty much just verifying the wire compatibility with each other
+ @Test
+ public void testPreviousWriteUnsignedWay() {
+ // Test writing 0
+ wrapper.writeShort((short) 0);
+ assertEquals(0, wrapper.readUnsignedShort());
+
+ // Test writing max unsigned short value (65535)
+ wrapper.writeShort((short)65535);
+ assertEquals(65535, wrapper.readUnsignedShort());
+
+ // Test writing mid-range value
+ wrapper.writeShort((short)32768);
+ assertEquals(32768, wrapper.readUnsignedShort());
+
+ // Test writing value that would be negative as signed short
+ wrapper.writeShort((short)40000);
+ assertEquals(40000, wrapper.readUnsignedShort());
+ }
+
+ @Test
+ public void testWriteUnsignedShortInvalidValues() throws Exception {
+ // Test that negative values throw IllegalArgumentException
+ try {
+ wrapper.writeUnsignedShort(-1);
+ fail("Expected IllegalArgumentException for negative value");
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+
+ // Test that values > 65535 throw IllegalArgumentException
+ try {
+ wrapper.writeUnsignedShort(65536);
+ fail("Expected IllegalArgumentException for value > 65535");
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+
+ try {
+ wrapper.writeUnsignedShort(100000);
+ fail("Expected IllegalArgumentException for value > 65535");
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+ }
+
@Test
public void testShort() throws Exception {
wrapper.writeShort((short) 1);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org