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/17 18:02:35 UTC
[activemq-artemis] branch master updated: ARTEMIS-2661 AMQP Journal
loading is triggering reencode
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
The following commit(s) were added to refs/heads/master by this push:
new 5f0b7dd ARTEMIS-2661 AMQP Journal loading is triggering reencode
new 579425c This closes #3023
5f0b7dd is described below
commit 5f0b7dd7cd5ee0ad78eb76edd87c2b66a59530a7
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Tue Mar 17 11:33:02 2020 +0100
ARTEMIS-2661 AMQP Journal loading is triggering reencode
---
.../artemis/protocol/amqp/broker/AMQPMessage.java | 72 +++++++++++++----
.../protocol/amqp/broker/AMQPStandardMessage.java | 11 +--
.../protocol/amqp/broker/AMQPMessageTest.java | 30 ++++++++
.../amqp/journal/AmqpJournalLoadingTest.java | 89 ++++++++++++++++++++++
4 files changed, 178 insertions(+), 24 deletions(-)
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 077d9dd..a4c9686 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
@@ -17,7 +17,6 @@
package org.apache.activemq.artemis.protocol.amqp.broker;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
@@ -122,11 +121,49 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
protected static final int VALUE_NOT_PRESENT = -1;
+ /**
+ * This has been made public just for testing purposes: it's not stable
+ * and developers shouldn't rely on this for developing purposes.
+ */
+ public enum MessageDataScanningStatus {
+ NOT_SCANNED(0), RELOAD_PERSISTENCE(1), SCANNED(2);
+
+ private static final MessageDataScanningStatus[] STATES;
+
+ static {
+ // this prevent codes to be assigned with the wrong order
+ // and ensure valueOf to work fine
+ final MessageDataScanningStatus[] states = values();
+ STATES = new MessageDataScanningStatus[states.length];
+ for (final MessageDataScanningStatus state : states) {
+ final int code = state.code;
+ if (STATES[code] != null) {
+ throw new AssertionError("code already in use: " + code);
+ }
+ STATES[code] = state;
+ }
+ }
+
+ final byte code;
+
+ MessageDataScanningStatus(int code) {
+ assert code >= 0 && code <= Byte.MAX_VALUE;
+ this.code = (byte) code;
+ }
+
+ static MessageDataScanningStatus valueOf(int code) {
+ checkCode(code);
+ return STATES[code];
+ }
+
+ private static void checkCode(int code) {
+ if (code < 0 || code > (STATES.length - 1)) {
+ throw new IllegalArgumentException("invalid status code: " + code);
+ }
+ }
+ }
// Buffer and state for the data backing this message.
- protected static final byte NOT_SCANNED = 0;
- protected static final byte RELOAD_PERSISTENCE = 1;
- protected static final byte SCANNED = 2;
- protected byte messageDataScanned;
+ protected byte messageDataScanned = MessageDataScanningStatus.NOT_SCANNED.code;
// Marks the message as needed to be re-encoded to update the backing buffer
protected boolean modified;
@@ -191,6 +228,15 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
this.coreMessageObjectPools = null;
}
+ /**
+ * Similarly to {@link MessageDataScanningStatus}, this method is made available only for testing
+ * purposes to check the message data scanning status.<br>
+ * Its access is not thread-safe and it shouldn't return {@code null}.
+ */
+ public final MessageDataScanningStatus messageDataScanned() {
+ return MessageDataScanningStatus.valueOf(messageDataScanned);
+ }
+
/** This will return application properties without attempting to decode it.
* That means, if applicationProperties were never parsed before, this will return null, even if there is application properties.
* This was created as an internal method for testing, as we need to validate if the application properties are not decoded until needed. */
@@ -448,7 +494,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
// re-encode should be done to update the backing data with the in memory elements.
protected synchronized void ensureMessageDataScanned() {
- final byte state = messageDataScanned;
+ final MessageDataScanningStatus state = MessageDataScanningStatus.valueOf(messageDataScanned);
switch (state) {
case NOT_SCANNED:
scanMessageData();
@@ -459,10 +505,6 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
case SCANNED:
// NO-OP
break;
- default:
- throw new IllegalStateException("invalid messageDataScanned state: expected within " +
- Arrays.toString(new byte[]{NOT_SCANNED, SCANNED, RELOAD_PERSISTENCE}) +
- " but " + messageDataScanned);
}
}
@@ -509,7 +551,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
TLSEncode.getDecoder().setBuffer(new NettyReadable(buf));
try {
- messageDataScanned = SCANNED;
+ messageDataScanned = MessageDataScanningStatus.SCANNED.code;
headerPosition = buf.readInt();
encodedHeaderSize = buf.readInt();
@@ -553,7 +595,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
}
protected synchronized void scanMessageData() {
- this.messageDataScanned = SCANNED;
+ this.messageDataScanned = MessageDataScanningStatus.SCANNED.code;
DecoderImpl decoder = TLSEncode.getDecoder();
decoder.setBuffer(getData().rewind());
@@ -791,9 +833,9 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
public abstract void reloadPersistence(ActiveMQBuffer record, CoreMessageObjectPools pools);
protected synchronized void lazyScanAfterReloadPersistence() {
- assert messageDataScanned == RELOAD_PERSISTENCE;
+ assert messageDataScanned == MessageDataScanningStatus.RELOAD_PERSISTENCE.code;
scanMessageData();
- messageDataScanned = SCANNED;
+ messageDataScanned = MessageDataScanningStatus.SCANNED.code;
modified = false;
// Message state should reflect that is came from persistent storage which
@@ -1164,7 +1206,7 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
private boolean anyMessageAnnotations(Symbol[] symbols, KMPNeedle[] symbolNeedles) {
assert symbols.length == symbolNeedles.length;
final int count = symbols.length;
- if (messageDataScanned == SCANNED) {
+ if (messageDataScanned == MessageDataScanningStatus.SCANNED.code) {
final MessageAnnotations messageAnnotations = this.messageAnnotations;
if (messageAnnotations == null) {
return false;
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 99d3be3..2a15adc 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
@@ -162,14 +162,7 @@ public class AMQPStandardMessage extends AMQPMessage {
// Message state is now that the underlying buffer is loaded, but the contents not yet scanned
resetMessageData();
modified = false;
- messageDataScanned = RELOAD_PERSISTENCE;
- // can happen when moved to a durable location. We must re-encode here to
- // avoid a subsequent redelivery from suddenly appearing with a durable header
- // tag when the initial delivery did not.
- if (!isDurable()) {
- setDurable(true);
- reencode();
- }
+ messageDataScanned = MessageDataScanningStatus.RELOAD_PERSISTENCE.code;
}
@Override
@@ -207,7 +200,7 @@ public class AMQPStandardMessage extends AMQPMessage {
@Override
protected synchronized void encodeMessage() {
this.modified = false;
- this.messageDataScanned = NOT_SCANNED;
+ this.messageDataScanned = MessageDataScanningStatus.NOT_SCANNED.code;
int estimated = Math.max(1500, data != null ? data.capacity() + 1000 : 0);
ByteBuf buffer = PooledByteBufAllocator.DEFAULT.heapBuffer(estimated);
EncoderImpl encoder = TLSEncode.getEncoder();
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 8822481..efe356c 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
@@ -172,11 +172,21 @@ public class AMQPMessageTest {
} catch (NullPointerException npe) {
}
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.messageDataScanned());
+
// Now reload from encoded data
message.reloadPersistence(encoded, null);
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
assertTrue(message.hasScheduledDeliveryTime());
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
message.getHeader();
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, message.messageDataScanned());
+
assertTrue(message.hasScheduledDeliveryTime());
}
@@ -195,11 +205,21 @@ public class AMQPMessageTest {
} catch (NullPointerException npe) {
}
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.messageDataScanned());
+
// Now reload from encoded data
message.reloadPersistence(encoded, null);
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
assertTrue(message.hasScheduledDeliveryTime());
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
message.getHeader();
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, message.messageDataScanned());
+
assertTrue(message.hasScheduledDeliveryTime());
}
@@ -215,11 +235,21 @@ public class AMQPMessageTest {
} catch (NullPointerException npe) {
}
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.NOT_SCANNED, message.messageDataScanned());
+
// Now reload from encoded data
message.reloadPersistence(encoded, null);
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
assertFalse(message.hasScheduledDeliveryTime());
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, message.messageDataScanned());
+
message.getHeader();
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, message.messageDataScanned());
+
assertFalse(message.hasScheduledDeliveryTime());
}
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
new file mode 100644
index 0000000..4f5315c
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/journal/AmqpJournalLoadingTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.journal;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
+import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpReceiver;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.hamcrest.Matchers;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AmqpJournalLoadingTest extends AmqpClientTestSupport {
+
+ @Test
+ public void durableMessageDataNotScannedOnRestartTest() throws Exception {
+ sendMessages(getQueueName(), 1, true);
+ final Queue queueView = getProxyToQueue(getQueueName());
+ Wait.assertTrue("All messages should arrive", () -> queueView.getMessageCount() == 1);
+
+ server.stop();
+ server.start();
+
+ final AMQPMessage amqpMessage;
+
+ final Queue afterRestartQueueView = getProxyToQueue(getQueueName());
+
+ Wait.assertTrue("All messages should arrive", () -> afterRestartQueueView.getMessageCount() == 1);
+
+ try (LinkedListIterator<MessageReference> iterator = afterRestartQueueView.iterator()) {
+ Assert.assertTrue(iterator.hasNext());
+ final MessageReference next = iterator.next();
+ Assert.assertNotNull(next);
+ Assert.assertFalse(iterator.hasNext());
+ final Message message = next.getMessage();
+ Assert.assertThat(message, Matchers.instanceOf(AMQPMessage.class));
+ amqpMessage = (AMQPMessage) message;
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.RELOAD_PERSISTENCE, amqpMessage.messageDataScanned());
+ }
+
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+ AmqpSession session = connection.createSession();
+
+ AmqpReceiver receiver = session.createReceiver(getQueueName());
+
+ receiver.flow(1);
+ AmqpMessage receive = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(receive);
+ assertTrue(receive.isDurable());
+
+ assertEquals(1, afterRestartQueueView.getMessageCount());
+
+ Assert.assertEquals(AMQPMessage.MessageDataScanningStatus.SCANNED, amqpMessage.messageDataScanned());
+
+ receive.accept();
+
+ receiver.close();
+
+ Wait.assertEquals(0, () -> afterRestartQueueView.getMessageCount());
+
+ connection.close();
+ }
+
+}