Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-2658 AMQP message read from page has wrong encode size

This commit is contained in:
Francesco Nigro 2020-03-12 19:38:13 +01:00
parent 31f00fda36
commit 2105479304
4 changed files with 87 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/TypedProperties.java
View File

@ -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);
}

5
artemis-commons/src/test/java/org/apache/activemq/artemis/utils/TypedPropertiesTest.java
View File

@ -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));
}

50
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessagePersisterV2.java
View File

@ -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);
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);
}
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());
}
return message;
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) {
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 record;
}
}

51
artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageReaderTest.java
View File

@ -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);
@ -69,6 +72,30 @@ public class PageReaderTest extends ActiveMQTestBase {
pageReader.close();
}
@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");