[activemq-artemis] branch master updated: ARTEMIS-2669 not durable AMQP messages cannot became durable on depaging

[cl...@apache.org]

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 085a74c  ARTEMIS-2669 not durable AMQP messages cannot became durable on depaging
     new f44803d  This closes #3036
085a74c is described below

commit 085a74cf06a1157d85876a723741dd51c1de30b3
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Fri Mar 20 11:56:04 2020 +0100

    ARTEMIS-2669 not durable AMQP messages cannot became durable on depaging
---
 .../artemis/protocol/amqp/broker/AMQPMessage.java  |  9 ---------
 .../integration/amqp/paging/AmqpPagingTest.java    | 22 ++++++++++++++++++++++
 2 files changed, 22 insertions(+), 9 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 b47013a..64fec6e 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
@@ -836,15 +836,6 @@ public abstract class AMQPMessage extends RefCountMessage implements org.apache.
       scanMessageData();
       messageDataScanned = MessageDataScanningStatus.SCANNED.code;
       modified = false;
-
-      // Message state should reflect that is came from persistent storage which
-      // 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();
-      }
    }
 
    @Override
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpPagingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpPagingTest.java
index 2ed9845..b4085ed 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpPagingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpPagingTest.java
@@ -16,9 +16,12 @@
  */
 package org.apache.activemq.artemis.tests.integration.amqp.paging;
 
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
 import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
@@ -30,9 +33,23 @@ import org.apache.activemq.transport.amqp.client.AmqpSender;
 import org.apache.activemq.transport.amqp.client.AmqpSession;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
+@RunWith(Parameterized.class)
 public class AmqpPagingTest extends AmqpClientTestSupport {
 
+   @Parameterized.Parameters(name = "durability={0}")
+   public static Collection getParams() {
+      return Arrays.asList(new Object[][]{{Boolean.TRUE}, {Boolean.FALSE}, {null}});
+   }
+
+   private final Boolean durable;
+
+   public AmqpPagingTest(Boolean durable) {
+      this.durable = durable;
+   }
+
    @Override
    protected void addConfiguration(ActiveMQServer server) {
       super.addConfiguration(server);
@@ -64,13 +81,18 @@ public class AmqpPagingTest extends AmqpClientTestSupport {
       for (int i = 0; i < MSG_COUNT; i++) {
          AmqpMessage message = new AmqpMessage();
          message.setText(data);
+         if (durable != null) {
+            message.setDurable(durable);
+         }
          sender.send(message);
       }
+      Assert.assertTrue(server.getPagingManager().getPageStore(SimpleString.toSimpleString(getQueueName())).isPaging());
       sender.close();
       receiver.flow(MSG_COUNT);
       for (int i = 0; i < MSG_COUNT; i++) {
          AmqpMessage receive = receiver.receive(10, TimeUnit.SECONDS);
          assertNotNull("Not received anything after " + i + " receive", receive);
+         Assert.assertEquals(durable == null ? false : durable.booleanValue(), receive.isDurable());
          receive.accept();
       }
       receiver.close();