You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/03/16 20:44:11 UTC
[activemq-artemis] 01/03: ARTEMIS-2658 AMQP message read from page
has wrong encode size
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 2105479304a7efead14ecb343686bf31042f1731
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Thu Mar 12 19:38:13 2020 +0100
ARTEMIS-2658 AMQP message read from page has wrong encode size
---
.../artemis/utils/collections/TypedProperties.java | 22 +++-------
.../artemis/utils/TypedPropertiesTest.java | 5 +++
.../amqp/broker/AMQPMessagePersisterV2.java | 46 ++++++++++++-------
.../core/paging/cursor/impl/PageReaderTest.java | 51 ++++++++++++++++++----
4 files changed, 85 insertions(+), 39 deletions(-)
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 df81c28..81755ae 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
@@ -446,21 +446,17 @@ public class TypedProperties {
}
public synchronized void decode(final ByteBuf buffer,
- final TypedPropertiesDecoderPools keyValuePools,
- boolean replaceExisting) {
+ final TypedPropertiesDecoderPools keyValuePools) {
byte b = buffer.readByte();
if (b == DataConstants.NULL) {
- if (replaceExisting) {
- properties = null;
- size = 0;
- }
+ properties = null;
+ size = 0;
} else {
int numHeaders = buffer.readInt();
- if (replaceExisting || properties == null) {
- //optimize the case of no collisions to avoid any resize (it doubles the map size!!!) when load factor is reached
- properties = new HashMap<>(numHeaders, 1.0f);
- }
- size = properties.size();
+
+ //optimize the case of no collisions to avoid any resize (it doubles the map size!!!) when load factor is reached
+ properties = new HashMap<>(numHeaders, 1.0f);
+ size = 0;
for (int i = 0; i < numHeaders; i++) {
final SimpleString key = SimpleString.readSimpleString(buffer, keyValuePools == null ? null : keyValuePools.getPropertyKeysPool());
@@ -533,10 +529,6 @@ public class TypedProperties {
}
}
- public synchronized void decode(final ByteBuf buffer, final TypedPropertiesDecoderPools keyValuePools) {
- decode(buffer, keyValuePools, true);
- }
-
public void decode(final ByteBuf buffer) {
decode(buffer, null);
}
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/TypedPropertiesTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/TypedPropertiesTest.java
index e9e7322..0786dad 100644
--- a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/TypedPropertiesTest.java
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/TypedPropertiesTest.java
@@ -32,6 +32,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import static org.apache.activemq.artemis.utils.collections.TypedProperties.searchProperty;
+import static org.hamcrest.Matchers.greaterThan;
public class TypedPropertiesTest {
@@ -96,8 +97,12 @@ public class TypedPropertiesTest {
Assert.assertTrue(props.containsProperty(key));
Assert.assertNotNull(props.getProperty(key));
+ Assert.assertThat(props.getEncodeSize(), greaterThan(0));
+
props.clear();
+ Assert.assertEquals(1, props.getEncodeSize());
+
Assert.assertFalse(props.containsProperty(key));
Assert.assertNull(props.getProperty(key));
}
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 1438f76..d17cbab 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
@@ -17,8 +17,11 @@
package org.apache.activemq.artemis.protocol.amqp.broker;
+import java.util.Objects;
+
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.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.utils.DataConstants;
import org.apache.activemq.artemis.utils.collections.TypedProperties;
@@ -72,25 +75,36 @@ public class AMQPMessagePersisterV2 extends AMQPMessagePersister {
}
@Override
- public Message decode(ActiveMQBuffer buffer, Message record, CoreMessageObjectPools pool) {
- AMQPMessage message = (AMQPMessage) super.decode(buffer, record, pool);
+ public Message decode(ActiveMQBuffer buffer, Message ignore, CoreMessageObjectPools pool) {
+ // IMPORTANT:
+ // This is a sightly modified copy of the AMQPMessagePersister::decode body
+ // to save extraProperties to be created twice: this would kill GC during journal loading
+ long id = buffer.readLong();
+ long format = buffer.readLong();
+ // this instance is being used only if there are no extraProperties or just for debugging purposes:
+ // on journal loading pool shouldn't be null so it shouldn't create any garbage.
+ final SimpleString address;
+ if (pool == null) {
+ address = buffer.readNullableSimpleString();
+ } else {
+ address = SimpleString.readNullableSimpleString(buffer.byteBuf(), pool.getAddressDecoderPool());
+ }
+ AMQPStandardMessage record = new AMQPStandardMessage(format);
+ record.reloadPersistence(buffer, pool);
+ record.setMessageID(id);
+ // END of AMQPMessagePersister::decode body copy
int size = buffer.readInt();
-
if (size != 0) {
- // message::setAddress could have populated extra properties
- // hence, we can safely replace the value on the properties
- // if it has been encoded differently in the rest of the buffer
- TypedProperties existingExtraProperties = message.getExtraProperties();
- TypedProperties extraProperties = existingExtraProperties;
- if (existingExtraProperties == null) {
- extraProperties = new TypedProperties(Message.INTERNAL_PROPERTY_NAMES_PREDICATE);
- }
- extraProperties.decode(buffer.byteBuf(), pool != null ? pool.getPropertiesDecoderPools() : null, existingExtraProperties == null);
- if (extraProperties != existingExtraProperties) {
- message.setExtraProperties(extraProperties);
- }
+ final TypedProperties extraProperties = record.createExtraProperties();
+ extraProperties.decode(buffer.byteBuf(), pool != null ? pool.getPropertiesDecoderPools() : null);
+ assert Objects.equals(address, extraProperties.getSimpleStringProperty(AMQPMessage.ADDRESS_PROPERTY)) :
+ "AMQPMessage address and extraProperties address should match";
+ } else if (address != null) {
+ // this shouldn't really happen: this code path has been preserved
+ // because of the behaviour before "ARTEMIS-2617 Improve AMQP Journal loading"
+ record.setAddress(address);
}
- return message;
+ return record;
}
}
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageReaderTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageReaderTest.java
index 138bf5a..f347714 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageReaderTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageReaderTest.java
@@ -17,7 +17,10 @@
package org.apache.activemq.artemis.core.paging.cursor.impl;
+import java.util.Arrays;
+
import org.apache.activemq.artemis.api.core.ICoreMessage;
+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.io.SequentialFileFactory;
@@ -58,7 +61,7 @@ public class PageReaderTest extends ActiveMQTestBase {
int nextFileOffset = pagedMessage == null ? -1 : offsets[i - 1] + pagedMessage.getEncodeSize() + Page.SIZE_RECORD;
PagePositionAndFileOffset startPosition = new PagePositionAndFileOffset(nextFileOffset, new PagePositionImpl(10, i - 1));
PagePosition pagePosition = startPosition.nextPagePostion();
- assertEquals(offsets[i], pagePosition.getFileOffset());
+ assertEquals("Message " + i + " has wrong offset", offsets[i], pagePosition.getFileOffset());
pagedMessage = pageReader.getMessage(pagePosition);
}
assertNotNull(pagedMessage);
@@ -70,6 +73,30 @@ public class PageReaderTest extends ActiveMQTestBase {
}
@Test
+ public void testShortPageReadMessage() throws Exception {
+ recreateDirectory(getTestDir());
+ int num = 2;
+ int[] offsets = createPage(num);
+ PageReader pageReader = getPageReader();
+
+ PagedMessage[] pagedMessages = pageReader.getMessages();
+ assertEquals(pagedMessages.length, num);
+
+ PagePosition pagePosition = new PagePositionImpl(10, 0);
+ PagedMessage firstPagedMessage = pageReader.getMessage(pagePosition);
+ assertEquals("Message 0 has a wrong encodeSize", pagedMessages[0].getEncodeSize(), firstPagedMessage.getEncodeSize());
+ int nextFileOffset = offsets[0] + firstPagedMessage.getEncodeSize() + Page.SIZE_RECORD;
+ PagePositionAndFileOffset startPosition = new PagePositionAndFileOffset(nextFileOffset, new PagePositionImpl(10, 0));
+ PagePosition nextPagePosition = startPosition.nextPagePostion();
+ assertEquals("Message 1 has a wrong offset", offsets[1], nextPagePosition.getFileOffset());
+ PagedMessage pagedMessage = pageReader.getMessage(nextPagePosition);
+ assertNotNull(pagedMessage);
+ assertEquals(pagedMessage.getMessage().getMessageID(), 1);
+ assertEquals(pagedMessages[1].getMessage().getMessageID(), 1);
+ pageReader.close();
+ }
+
+ @Test
public void testPageReadMessageBeyondPage() throws Exception {
recreateDirectory(getTestDir());
@@ -113,15 +140,12 @@ public class PageReaderTest extends ActiveMQTestBase {
Page page = new Page(new SimpleString("something"), new NullStorageManager(), factory, file, 10);
page.open();
SimpleString simpleDestination = new SimpleString("Test");
+ final int msgSize = 100;
+ final byte[] content = new byte[msgSize];
+ Arrays.fill(content, (byte) 'b');
int[] offsets = new int[num];
for (int i = 0; i < num; i++) {
- ICoreMessage msg = new CoreMessage().setMessageID(i).initBuffer(1024);
-
- for (int j = 0; j < 100; j++) {
- msg.getBodyBuffer().writeByte((byte) 'b');
- }
-
- msg.setAddress(simpleDestination);
+ Message msg = createMessage(simpleDestination, i, content);
offsets[i] = (int)page.getFile().position();
page.write(new PagedMessageImpl(msg, new long[0]));
@@ -131,6 +155,17 @@ public class PageReaderTest extends ActiveMQTestBase {
return offsets;
}
+ protected Message createMessage(SimpleString address, int msgId, byte[] content) {
+ ICoreMessage msg = new CoreMessage().setMessageID(msgId).initBuffer(1024);
+
+ for (byte b : content) {
+ msg.getBodyBuffer().writeByte(b);
+ }
+
+ msg.setAddress(address);
+ return msg;
+ }
+
private PageReader getPageReader() throws Exception {
SequentialFileFactory factory = new NIOSequentialFileFactory(getTestDirfile(), 1);
SequentialFile file = factory.createSequentialFile("00010.page");