ARTEMIS-890 Improving Paging consistencies with broker.persistent = false. Block, Page and Drop will now work under non persistent

(cherry picked from commit 332338d018)

artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Create.java

@@ -714,6 +714,12 @@ public class Create extends InputAbstract {
          filters.put("${hornetq-acceptor}", applyFilters(readTextFile(ETC_HORNETQ_ACCEPTOR_TXT), filters));
       }
 
+      if (disablePersistence) {
+         filters.put("${full-policy}", "BLOCK");
+      } else {
+         filters.put("${full-policy}", "PAGE");
+      }
+
       performAutoTune(filters, aio, dataFolder);
 
       write(ETC_BROKER_XML, filters, false);

artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml

@@ -95,7 +95,7 @@ ${cluster-security.settings}${cluster.settings}${replicated.settings}${shared-st
             <!-- with -1 only the global-max-size is in use for limiting -->
             <max-size-bytes>-1</max-size-bytes>
             <message-counter-history-day-limit>10</message-counter-history-day-limit>
-            <address-full-policy>PAGE</address-full-policy>
+            <address-full-policy>${full-policy}</address-full-policy>
          </address-setting>
       </address-settings>
    </core>

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java

@@ -797,7 +797,9 @@ public class PagingStoreImpl implements PagingStore {
          lock.readLock().unlock();
       }
 
+      if (managerLock != null) {
          managerLock.lock();
+      }
       try {
          lock.writeLock().lock();
 
@@ -852,9 +854,11 @@ public class PagingStoreImpl implements PagingStore {
             lock.writeLock().unlock();
          }
       } finally {
+         if (managerLock != null) {
             managerLock.unlock();
          }
       }
+   }
 
    /**
     * This method will disable cleanup of pages. No page will be deleted after this call.

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java

@@ -529,12 +529,23 @@ public class NullStorageManager implements StorageManager {
       // no-op
    }
 
+
    @Override
-   public boolean addToPage(PagingStore s,
+   public boolean addToPage(PagingStore store,
                             ServerMessage msg,
                             Transaction tx,
                             RouteContextList listCtx) throws Exception {
-      return false;
+      /**
+       * Exposing the read-lock here is an encapsulation violation done in order to keep the code
+       * simpler. The alternative would be to add a second method, say 'verifyPaging', to
+       * PagingStore.
+       * <p>
+       * Adding this second method would also be more surprise prone as it would require a certain
+       * calling order.
+       * <p>
+       * The reasoning is that exposing the lock is more explicit and therefore `less bad`.
+       */
+      return store.page(msg, tx, listCtx, null);
    }
 
    @Override

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingTest.java

@@ -3327,7 +3327,16 @@ public class PagingTest extends ActiveMQTestBase {
    }
 
    @Test
-   public void testDropMessages() throws Exception {
+   public void testDropMessagesPersistent() throws Exception {
+      testDropMessages(true);
+   }
+
+   @Test
+   public void testDropMessagesNonPersistent() throws Exception {
+      testDropMessages(false);
+   }
+
+   public void testDropMessages(final boolean persistent) throws Exception {
       clearDataRecreateServerDirs();
 
       HashMap<String, AddressSettings> settings = new HashMap<>();
@@ -3337,7 +3346,7 @@ public class PagingTest extends ActiveMQTestBase {
 
       settings.put(PagingTest.ADDRESS.toString(), set);
 
-      server = createServer(true, createDefaultInVMConfig(), 1024, 10 * 1024, settings);
+      server = createServer(persistent, createDefaultInVMConfig(), 1024, 10 * 1024, settings);
 
       server.start();