activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject activemq-artemis git commit: ARTEMIS-332 - test added / better dealing with critical errors on paging
Date Thu, 07 Jan 2016 00:43:21 GMT
Repository: activemq-artemis
Updated Branches:
  refs/heads/master d4dd07023 -> a5a993ed9


ARTEMIS-332 - test added / better dealing with critical errors on paging


Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/a5a993ed
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/a5a993ed
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/a5a993ed

Branch: refs/heads/master
Commit: a5a993ed9d67e47e97a0a8198d75600188df2eae
Parents: d4dd070
Author: Clebert Suconic <clebertsuconic@apache.org>
Authored: Wed Jan 6 15:45:19 2016 -0500
Committer: Clebert Suconic <clebertsuconic@apache.org>
Committed: Wed Jan 6 19:42:45 2016 -0500

----------------------------------------------------------------------
 .../artemis/core/paging/PagingStore.java        |   2 -
 .../core/paging/cursor/PageCursorProvider.java  |   5 +-
 .../core/paging/cursor/PageSubscription.java    |   5 +-
 .../core/paging/cursor/PagedReference.java      |   3 +-
 .../core/paging/cursor/PagedReferenceImpl.java  |  17 +-
 .../cursor/impl/PageCursorProviderImpl.java     |  10 +-
 .../cursor/impl/PageSubscriptionImpl.java       | 102 +++------
 .../core/paging/impl/PagingStoreImpl.java       |   6 -
 .../artemis/core/server/MessageReference.java   |   4 +-
 .../activemq/artemis/core/server/Queue.java     |   2 +-
 .../core/server/impl/LastValueQueue.java        |  79 ++-----
 .../artemis/core/server/impl/QueueImpl.java     |  63 +++---
 .../artemis/core/server/impl/RefsOperation.java |  37 ++--
 .../tests/extras/byteman/PagingOMETest.java     | 206 +++++++++++++++++++
 .../storage/PersistMultiThreadTest.java         |   4 -
 15 files changed, 320 insertions(+), 225 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
index c8808b3..e831966 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
@@ -47,8 +47,6 @@ public interface PagingStore extends ActiveMQComponent {
 
    int getNumberOfPages();
 
-   void criticalException(Throwable e);
-
    /**
     * Returns the page id of the current page in which the system is writing files.
     */

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageCursorProvider.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageCursorProvider.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageCursorProvider.java
index c8404ba..951b83c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageCursorProvider.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageCursorProvider.java
@@ -16,7 +16,6 @@
  */
 package org.apache.activemq.artemis.core.paging.cursor;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 
@@ -25,7 +24,7 @@ import org.apache.activemq.artemis.core.paging.PagedMessage;
  */
 public interface PageCursorProvider {
 
-   PageCache getPageCache(long pageNr) throws ActiveMQException;
+   PageCache getPageCache(long pageNr);
 
    PagedReference newReference(final PagePosition pos, final PagedMessage msg, PageSubscription
sub);
 
@@ -39,7 +38,7 @@ public interface PageCursorProvider {
 
    PageSubscription createSubscription(long queueId, Filter filter, boolean durable);
 
-   PagedMessage getMessage(PagePosition pos) throws ActiveMQException;
+   PagedMessage getMessage(PagePosition pos);
 
    void processReload() throws Exception;
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
index 386f21f..df2ccc3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
@@ -18,7 +18,6 @@ package org.apache.activemq.artemis.core.paging.cursor;
 
 import java.util.concurrent.Executor;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.paging.PagingStore;
 import org.apache.activemq.artemis.core.paging.impl.Page;
@@ -96,7 +95,7 @@ public interface PageSubscription {
 
    void reloadPageCompletion(PagePosition position);
 
-   void reloadPageInfo(long pageNr) throws ActiveMQException;
+   void reloadPageInfo(long pageNr);
 
    /**
     * To be called when the cursor decided to ignore a position.
@@ -148,7 +147,7 @@ public interface PageSubscription {
     * @param pos
     * @return
     */
-   PagedMessage queryMessage(PagePosition pos) throws ActiveMQException;
+   PagedMessage queryMessage(PagePosition pos);
 
    /**
     * @return executor used by the PageSubscription

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReference.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReference.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReference.java
index 46041c5..c1ff089 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReference.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReference.java
@@ -16,7 +16,6 @@
  */
 package org.apache.activemq.artemis.core.paging.cursor;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.server.MessageReference;
 
@@ -24,5 +23,5 @@ public interface PagedReference extends MessageReference {
 
    PagePosition getPosition();
 
-   PagedMessage getPagedMessage() throws ActiveMQException;
+   PagedMessage getPagedMessage();
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReferenceImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReferenceImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReferenceImpl.java
index 964737f..82b0e92 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReferenceImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagedReferenceImpl.java
@@ -19,7 +19,6 @@ package org.apache.activemq.artemis.core.paging.cursor;
 import java.lang.ref.WeakReference;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
@@ -50,12 +49,12 @@ public class PagedReferenceImpl implements PagedReference {
    private boolean alreadyAcked;
 
    @Override
-   public ServerMessage getMessage() throws ActiveMQException {
+   public ServerMessage getMessage() {
       return getPagedMessage().getMessage();
    }
 
    @Override
-   public synchronized PagedMessage getPagedMessage() throws ActiveMQException {
+   public synchronized PagedMessage getPagedMessage() {
       PagedMessage returnMessage = message != null ? message.get() : null;
 
       // We only keep a few references on the Queue from paging...
@@ -111,7 +110,7 @@ public class PagedReferenceImpl implements PagedReference {
          try {
             messageEstimate = getMessage().getMemoryEstimate();
          }
-         catch (ActiveMQException e) {
+         catch (Throwable e) {
             ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
          }
       }
@@ -120,13 +119,7 @@ public class PagedReferenceImpl implements PagedReference {
 
    @Override
    public MessageReference copy(final Queue queue) {
-      try {
-         return new PagedReferenceImpl(this.position, this.getPagedMessage(), this.subscription);
-      }
-      catch (ActiveMQException e) {
-         ActiveMQServerLogger.LOGGER.warn(e);
-         return this;
-      }
+      return new PagedReferenceImpl(this.position, this.getPagedMessage(), this.subscription);
    }
 
    @Override
@@ -141,7 +134,7 @@ public class PagedReferenceImpl implements PagedReference {
                deliveryTime = 0L;
             }
          }
-         catch (ActiveMQException e) {
+         catch (Throwable e) {
             ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
             return 0L;
          }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
index ef57e1c..5f5e1b3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
@@ -24,8 +24,6 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
-import org.apache.activemq.artemis.api.core.ActiveMQIOErrorException;
 import org.apache.activemq.artemis.core.filter.Filter;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.paging.PagingStore;
@@ -111,7 +109,7 @@ public class PageCursorProviderImpl implements PageCursorProvider {
    }
 
    @Override
-   public PagedMessage getMessage(final PagePosition pos) throws ActiveMQException {
+   public PagedMessage getMessage(final PagePosition pos) {
       PageCache cache = getPageCache(pos.getPageNr());
 
       if (cache == null || pos.getMessageNr() >= cache.getNumberOfMessages()) {
@@ -130,7 +128,7 @@ public class PageCursorProviderImpl implements PageCursorProvider {
    }
 
    @Override
-   public PageCache getPageCache(final long pageId) throws ActiveMQException {
+   public PageCache getPageCache(final long pageId) {
       try {
          PageCache cache;
          synchronized (softCache) {
@@ -157,8 +155,8 @@ public class PageCursorProviderImpl implements PageCursorProvider {
 
          return cache;
       }
-      catch (Throwable e) {
-         throw new ActiveMQIOErrorException("Couldn't complete paging due to an IO Exception
on Paging - " + e.getMessage(), e);
+      catch (Exception e) {
+         throw new RuntimeException(e.getMessage(), e);
       }
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
index d7a6ded..9c1702e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
@@ -333,7 +333,7 @@ final class PageSubscriptionImpl implements PageSubscription {
       return "PageSubscriptionImpl [cursorId=" + cursorId + ", queue=" + queue + ", filter
= " + filter + "]";
    }
 
-   private PagedReference getReference(PagePosition pos) throws ActiveMQException {
+   private PagedReference getReference(PagePosition pos) {
       return cursorProvider.newReference(pos, cursorProvider.getMessage(pos), this);
    }
 
@@ -342,7 +342,7 @@ final class PageSubscriptionImpl implements PageSubscription {
       return new CursorIterator();
    }
 
-   private PagedReference internalGetNext(final PagePosition pos) throws ActiveMQException
{
+   private PagedReference internalGetNext(final PagePosition pos) {
       PagePosition retPos = pos.nextMessage();
 
       PageCache cache = cursorProvider.getPageCache(pos.getPageNr());
@@ -471,17 +471,11 @@ final class PageSubscriptionImpl implements PageSubscription {
          public void onError(final int errorCode, final String errorMessage) {
             error = " errorCode=" + errorCode + ", msg=" + errorMessage;
             ActiveMQServerLogger.LOGGER.pageSubscriptionError(this, error);
-            getPagingStore().criticalException(new ActiveMQException(errorMessage));
          }
 
          @Override
          public void done() {
-            try {
-               processACK(position);
-            }
-            catch (ActiveMQException e) {
-               getPagingStore().criticalException(e);
-            }
+            processACK(position);
          }
 
          @Override
@@ -511,11 +505,10 @@ final class PageSubscriptionImpl implements PageSubscription {
 
    @Override
    public void addPendingDelivery(final PagePosition position) {
-      try {
-         getPageInfo(position).incrementPendingTX();
-      }
-      catch (Exception e) {
-         getPagingStore().criticalException(e);
+      PageCursorInfo info = getPageInfo(position);
+
+      if (info != null) {
+         info.incrementPendingTX();
       }
    }
 
@@ -535,7 +528,7 @@ final class PageSubscriptionImpl implements PageSubscription {
    }
 
    @Override
-   public PagedMessage queryMessage(PagePosition pos) throws ActiveMQException {
+   public PagedMessage queryMessage(PagePosition pos) {
       return cursorProvider.getMessage(pos);
    }
 
@@ -554,32 +547,17 @@ final class PageSubscriptionImpl implements PageSubscription {
    @Override
    public void reloadPreparedACK(final Transaction tx, final PagePosition position) {
       deliveredCount.incrementAndGet();
-      try {
-         installTXCallback(tx, position);
-      }
-      catch (Exception e) {
-         getPagingStore().criticalException(e);
-      }
+      installTXCallback(tx, position);
    }
 
    @Override
    public void positionIgnored(final PagePosition position) {
-      try {
-         processACK(position);
-      }
-      catch (Exception e) {
-         getPagingStore().criticalException(e);
-      }
+      processACK(position);
    }
 
    public void lateDeliveryRollback(PagePosition position) {
-      try {
-         PageCursorInfo cursorInfo = processACK(position);
-         cursorInfo.decrementPendingTX();
-      }
-      catch (ActiveMQException e) {
-         getPagingStore().criticalException(e);
-      }
+      PageCursorInfo cursorInfo = processACK(position);
+      cursorInfo.decrementPendingTX();
    }
 
    @Override
@@ -750,15 +728,15 @@ final class PageSubscriptionImpl implements PageSubscription {
    }
 
    @Override
-   public void reloadPageInfo(long pageNr) throws ActiveMQException {
+   public void reloadPageInfo(long pageNr) {
       getPageInfo(pageNr, true);
    }
 
-   private PageCursorInfo getPageInfo(final PagePosition pos) throws ActiveMQException {
+   private PageCursorInfo getPageInfo(final PagePosition pos) {
       return getPageInfo(pos.getPageNr(), true);
    }
 
-   private PageCursorInfo getPageInfo(final long pageNr, boolean create) throws ActiveMQException
{
+   private PageCursorInfo getPageInfo(final long pageNr, boolean create) {
       synchronized (consumedPages) {
          PageCursorInfo pageInfo = consumedPages.get(pageNr);
 
@@ -792,7 +770,7 @@ final class PageSubscriptionImpl implements PageSubscription {
 
    // To be called only after the ACK has been processed and guaranteed to be on storage
    // The only exception is on non storage events such as not matching messages
-   private PageCursorInfo processACK(final PagePosition pos) throws ActiveMQException {
+   private PageCursorInfo processACK(final PagePosition pos) {
       if (lastAckedPosition == null || pos.compareTo(lastAckedPosition) > 0) {
          if (isTrace) {
             ActiveMQServerLogger.LOGGER.trace("a new position is being processed as ACK");
@@ -828,7 +806,7 @@ final class PageSubscriptionImpl implements PageSubscription {
     * @param tx
     * @param position
     */
-   private void installTXCallback(final Transaction tx, final PagePosition position) throws
ActiveMQException {
+   private void installTXCallback(final Transaction tx, final PagePosition position) {
       if (position.getRecordID() >= 0) {
          // It needs to persist, otherwise the cursor will return to the fist page position
          tx.setContainsPersistent();
@@ -960,13 +938,7 @@ final class PageSubscriptionImpl implements PageSubscription {
       }
 
       public boolean isDone() {
-         try {
-            return completePage != null || (getNumberOfMessages() == confirmed.get() &&
pendingTX.get() == 0);
-         }
-         catch (ActiveMQException e) {
-            getPagingStore().criticalException(e);
-            throw new RuntimeException(e.getMessage(), e);
-         }
+         return completePage != null || (getNumberOfMessages() == confirmed.get() &&
pendingTX.get() == 0);
       }
 
       public boolean isPendingDelete() {
@@ -1047,7 +1019,7 @@ final class PageSubscriptionImpl implements PageSubscription {
          }
       }
 
-      private int getNumberOfMessages() throws ActiveMQException {
+      private int getNumberOfMessages() {
          if (wasLive) {
             // if the page was live at any point, we need to
             // get the number of messages from the page-cache
@@ -1089,12 +1061,7 @@ final class PageSubscriptionImpl implements PageSubscription {
             List<PagePosition> positions = entry.getValue();
 
             for (PagePosition confirmed : positions) {
-               try {
-                  cursor.processACK(confirmed);
-               }
-               catch (ActiveMQException e) {
-                  getPagingStore().criticalException(e);
-               }
+               cursor.processACK(confirmed);
                cursor.deliveredCount.decrementAndGet();
             }
 
@@ -1165,21 +1132,15 @@ final class PageSubscriptionImpl implements PageSubscription {
             return currentDelivery;
          }
 
-         try {
-            if (position == null) {
-               position = getStartPosition();
-            }
-
-            currentDelivery = moveNext();
-            return currentDelivery;
-         }
-         catch (ActiveMQException e) {
-            getPagingStore().criticalException(e);
-            throw new IllegalStateException(e.getMessage(), e);
+         if (position == null) {
+            position = getStartPosition();
          }
+
+         currentDelivery = moveNext();
+         return currentDelivery;
       }
 
-      private PagedReference moveNext() throws ActiveMQException {
+      private PagedReference moveNext() {
          synchronized (PageSubscriptionImpl.this) {
             boolean match = false;
 
@@ -1309,14 +1270,9 @@ final class PageSubscriptionImpl implements PageSubscription {
          deliveredCount.incrementAndGet();
          PagedReference delivery = currentDelivery;
          if (delivery != null) {
-            try {
-               PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(currentDelivery.getPosition());
-               if (info != null) {
-                  info.remove(currentDelivery.getPosition());
-               }
-            }
-            catch (ActiveMQException e) {
-               getPagingStore().criticalException(e);
+            PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(currentDelivery.getPosition());
+            if (info != null) {
+               info.remove(currentDelivery.getPosition());
             }
          }
       }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
index 9136c17..1463b3c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
@@ -176,12 +176,6 @@ public class PagingStoreImpl implements PagingStore {
 
    }
 
-   @Override
-   public void criticalException(Throwable e) {
-      ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
-      storeFactory.criticalException(e);
-   }
-
    /**
     * @param addressSettings
     */

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/MessageReference.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/MessageReference.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/MessageReference.java
index 95b30b2..0ff55ac 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/MessageReference.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/MessageReference.java
@@ -16,8 +16,6 @@
  */
 package org.apache.activemq.artemis.core.server;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
-
 /**
  * A reference to a message.
  *
@@ -27,7 +25,7 @@ public interface MessageReference {
 
    boolean isPaged();
 
-   ServerMessage getMessage() throws ActiveMQException;
+   ServerMessage getMessage();
 
    /**
     * We define this method aggregation here because on paging we need to hold the original
estimate,

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index 9ea60cd..81bd565 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -239,7 +239,7 @@ public interface Queue extends Bindable {
     */
    void deliverScheduledMessages() throws ActiveMQException;
 
-   void postAcknowledge(MessageReference ref) throws ActiveMQException;
+   void postAcknowledge(MessageReference ref);
 
    float getRate();
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
index c6d5aee..5420688 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java
@@ -21,7 +21,6 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.filter.Filter;
@@ -67,15 +66,7 @@ public class LastValueQueue extends QueueImpl {
 
    @Override
    public synchronized void addTail(final MessageReference ref, final boolean direct) {
-      SimpleString prop;
-
-      try {
-         prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
-      }
-      catch (ActiveMQException e) {
-         criticalError(e);
-         throw new IllegalStateException(e);
-      }
+      SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
 
       if (prop != null) {
          HolderReference hr = map.get(prop);
@@ -112,59 +103,45 @@ public class LastValueQueue extends QueueImpl {
 
    @Override
    public synchronized void addHead(final MessageReference ref) {
-      try {
-         SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
+      SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
 
-         if (prop != null) {
-            HolderReference hr = map.get(prop);
+      if (prop != null) {
+         HolderReference hr = map.get(prop);
 
-            if (hr != null) {
-               // We keep the current ref and ack the one we are returning
+         if (hr != null) {
+            // We keep the current ref and ack the one we are returning
 
-               super.referenceHandled();
+            super.referenceHandled();
 
-               try {
-                  super.acknowledge(ref);
-               }
-               catch (Exception e) {
-                  ActiveMQServerLogger.LOGGER.errorAckingOldReference(e);
-               }
+            try {
+               super.acknowledge(ref);
             }
-            else {
-               map.put(prop, (HolderReference) ref);
-
-               super.addHead(ref);
+            catch (Exception e) {
+               ActiveMQServerLogger.LOGGER.errorAckingOldReference(e);
             }
          }
          else {
+            map.put(prop, (HolderReference) ref);
+
             super.addHead(ref);
          }
       }
-      catch (ActiveMQException e) {
-         criticalError(e);
-         throw new IllegalStateException(e);
+      else {
+         super.addHead(ref);
       }
    }
 
    @Override
    protected void refRemoved(MessageReference ref) {
-      try {
-
-         synchronized (this) {
-            SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
+      synchronized (this) {
+         SimpleString prop = ref.getMessage().getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
 
-            if (prop != null) {
-               map.remove(prop);
-            }
+         if (prop != null) {
+            map.remove(prop);
          }
-
-         super.refRemoved(ref);
-      }
-      catch (ActiveMQException e) {
-         criticalError(e);
-         throw new IllegalStateException(e);
       }
 
+      super.refRemoved(ref);
    }
 
    private class HolderReference implements MessageReference {
@@ -223,13 +200,7 @@ public class LastValueQueue extends QueueImpl {
 
       @Override
       public ServerMessage getMessage() {
-         try {
-            return ref.getMessage();
-         }
-         catch (ActiveMQException e) {
-            criticalError(e);
-            throw new IllegalStateException(e);
-         }
+         return ref.getMessage();
       }
 
       @Override
@@ -285,13 +256,7 @@ public class LastValueQueue extends QueueImpl {
        */
       @Override
       public int getMessageMemoryEstimate() {
-         try {
-            return ref.getMessage().getMemoryEstimate();
-         }
-         catch (ActiveMQException e) {
-            criticalError(e);
-            throw new IllegalStateException(e);
-         }
+         return ref.getMessage().getMemoryEstimate();
       }
 
       /* (non-Javadoc)

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index c963e4d..12b5231 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -1054,13 +1054,7 @@ public class QueueImpl implements Queue {
 
    @Override
    public void cancel(final Transaction tx, final MessageReference reference, boolean ignoreRedeliveryCheck)
{
-      try {
-         getRefsOperation(tx, ignoreRedeliveryCheck).addAck(reference);
-      }
-      catch (ActiveMQException e) {
-         criticalError(e);
-         getPageSubscription().getPagingStore().criticalException(e);
-      }
+      getRefsOperation(tx, ignoreRedeliveryCheck).addAck(reference);
    }
 
    @Override
@@ -1777,12 +1771,7 @@ public class QueueImpl implements Queue {
 
    private synchronized void internalAddTail(final MessageReference ref) {
       refAdded(ref);
-      try {
-         messageReferences.addTail(ref, ref.getMessage().getPriority());
-      }
-      catch (ActiveMQException e) {
-         criticalError(e);
-      }
+      messageReferences.addTail(ref, getPriority(ref));
    }
 
    /**
@@ -1793,20 +1782,24 @@ public class QueueImpl implements Queue {
     * @param ref
     */
    private void internalAddHead(final MessageReference ref) {
+      queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
+      refAdded(ref);
+
+      int priority = getPriority(ref);
+
+      messageReferences.addHead(ref, priority);
+   }
+
+   private int getPriority(MessageReference ref) {
       try {
-         queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
-         refAdded(ref);
-         messageReferences.addHead(ref, ref.getMessage().getPriority());
+         return ref.getMessage().getPriority();
       }
-      catch (ActiveMQException e) {
-         criticalError(e);
+      catch (Throwable e) {
+         ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+         return 4; // the default one in case of failure
       }
    }
 
-   void criticalError(ActiveMQException e) {
-      storageManager.criticalError(e);
-   }
-
    private synchronized void doInternalPoll() {
 
       int added = 0;
@@ -2036,9 +2029,9 @@ public class QueueImpl implements Queue {
             // But we don't use the groupID on internal queues (clustered queues) otherwise
the group map would leak forever
             return ref.getMessage().getSimpleStringProperty(Message.HDR_GROUP_ID);
          }
-         catch (ActiveMQException e) {
-            criticalError(e);
-            throw new IllegalStateException(e);
+         catch (Throwable e) {
+            ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+            return null;
          }
       }
    }
@@ -2543,9 +2536,9 @@ public class QueueImpl implements Queue {
             return false;
          }
       }
-      catch (ActiveMQException e) {
-         criticalError(e);
-         throw new IllegalStateException(e);
+      catch (Throwable e) {
+         ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+         return false;
       }
    }
 
@@ -2584,7 +2577,7 @@ public class QueueImpl implements Queue {
    }
 
    @Override
-   public void postAcknowledge(final MessageReference ref) throws ActiveMQException {
+   public void postAcknowledge(final MessageReference ref) {
       QueueImpl queue = (QueueImpl) ref.getQueue();
 
       queue.decDelivering();
@@ -2594,9 +2587,17 @@ public class QueueImpl implements Queue {
          return;
       }
 
-      final ServerMessage message = ref.getMessage();
+      ServerMessage message;
+
+      try {
+         message = ref.getMessage();
+      }
+      catch (Throwable e) {
+         ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+         message = null;
+      }
 
-      boolean durableRef = message.isDurable() && queue.durable;
+      boolean durableRef = message != null && message.isDurable() && queue.durable;
 
       try {
          message.decrementRefCount();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
index 92d1a61..9b72f51 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
@@ -22,7 +22,6 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.core.persistence.StorageManager;
 import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
 import org.apache.activemq.artemis.core.server.MessageReference;
@@ -38,7 +37,7 @@ public class RefsOperation extends TransactionOperationAbstract {
    private Queue queue;
    List<MessageReference> refsToAck = new ArrayList<>();
 
-   List<ServerMessage> pagedMessagesToPostACK = null;
+   List<MessageReference> pagedMessagesToPostACK = null;
 
    /**
     * It will ignore redelivery check, which is used during consumer.close
@@ -56,13 +55,13 @@ public class RefsOperation extends TransactionOperationAbstract {
       ignoreRedeliveryCheck = true;
    }
 
-   synchronized void addAck(final MessageReference ref) throws ActiveMQException {
+   synchronized void addAck(final MessageReference ref) {
       refsToAck.add(ref);
       if (ref.isPaged()) {
          if (pagedMessagesToPostACK == null) {
             pagedMessagesToPostACK = new ArrayList<>();
          }
-         pagedMessagesToPostACK.add(ref.getMessage());
+         pagedMessagesToPostACK.add(ref);
       }
    }
 
@@ -148,32 +147,26 @@ public class RefsOperation extends TransactionOperationAbstract {
    public void afterCommit(final Transaction tx) {
       for (MessageReference ref : refsToAck) {
          synchronized (ref.getQueue()) {
-            try {
-               queue.postAcknowledge(ref);
-            }
-            catch (ActiveMQException e) {
-               if (queue instanceof QueueImpl) {
-                  ((QueueImpl) queue).criticalError(e);
-               }
-               else {
-                  ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
-               }
-            }
+            queue.postAcknowledge(ref);
          }
       }
 
       if (pagedMessagesToPostACK != null) {
-         for (ServerMessage msg : pagedMessagesToPostACK) {
-            try {
-               msg.decrementRefCount();
-            }
-            catch (Exception e) {
-               ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
-            }
+         for (MessageReference refmsg : pagedMessagesToPostACK) {
+            decrementRefCount(refmsg);
          }
       }
    }
 
+   private void decrementRefCount(MessageReference refmsg) {
+      try {
+         refmsg.getMessage().decrementRefCount();
+      }
+      catch (Exception e) {
+         ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+      }
+   }
+
    @Override
    public synchronized List<MessageReference> getRelatedMessageReferences() {
       List<MessageReference> listRet = new LinkedList<>();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/PagingOMETest.java
----------------------------------------------------------------------
diff --git a/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/PagingOMETest.java
b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/PagingOMETest.java
new file mode 100644
index 0000000..aa3ef38
--- /dev/null
+++ b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/PagingOMETest.java
@@ -0,0 +1,206 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.extras.byteman;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientMessage;
+import org.apache.activemq.artemis.api.core.client.ClientProducer;
+import org.apache.activemq.artemis.api.core.client.ClientSession;
+import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
+import org.apache.activemq.artemis.api.core.client.ServerLocator;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.jboss.byteman.contrib.bmunit.BMRule;
+import org.jboss.byteman.contrib.bmunit.BMRules;
+import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(BMUnitRunner.class)
+public class PagingOMETest extends ActiveMQTestBase {
+
+   private ServerLocator locator;
+   private ActiveMQServer server;
+   private ClientSessionFactory sf;
+   static final int MESSAGE_SIZE = 1024; // 1k
+
+   private static final int RECEIVE_TIMEOUT = 5000;
+
+   private static final int PAGE_MAX = 100 * 1024;
+
+   private static final int PAGE_SIZE = 10 * 1024;
+
+   static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+   static boolean failureActive = false;
+
+   public static void refCheck() {
+      if (failureActive) {
+         throw new OutOfMemoryError("fake error");
+      }
+   }
+
+   @Override
+   @Before
+   public void setUp() throws Exception {
+      super.setUp();
+      failureActive = false;
+      locator = createInVMNonHALocator();
+   }
+
+   @Test
+   @BMRules(
+      rules = {@BMRule(
+         name = "fakeOME",
+         targetClass = "org.apache.activemq.artemis.core.paging.cursor.PagedReferenceImpl",
+         targetMethod = "getPagedMessage",
+         targetLocation = "ENTRY",
+         action = "org.apache.activemq.artemis.tests.extras.byteman.PagingOMETest.refCheck()")})
+   public void testPageCleanup() throws Exception {
+      clearDataRecreateServerDirs();
+
+      Configuration config = createDefaultConfig(false);
+
+      config.setJournalSyncNonTransactional(false);
+
+      HashMap<String, AddressSettings> map = new HashMap<>();
+      AddressSettings value = new AddressSettings();
+      map.put(ADDRESS.toString(), value);
+      server = createServer(true, config, PAGE_SIZE, PAGE_MAX, map);
+
+      server.start();
+
+      final int numberOfMessages = 2;
+
+      locator = createInVMNonHALocator();
+
+      locator.setBlockOnNonDurableSend(true);
+      locator.setBlockOnDurableSend(true);
+      locator.setBlockOnAcknowledge(false);
+      locator.setConsumerWindowSize(0);
+
+      sf = createSessionFactory(locator);
+
+      ClientSession session = sf.createSession(false, false, false);
+
+      session.createQueue(ADDRESS, ADDRESS, null, true);
+
+      Queue queue = server.locateQueue(ADDRESS);
+      queue.getPageSubscription().getPagingStore().startPaging();
+
+      Assert.assertTrue(queue.getPageSubscription().getPagingStore().isPaging());
+
+      ClientProducer producer = session.createProducer(PagingOMETest.ADDRESS);
+
+      ClientMessage message = null;
+
+      byte[] body = new byte[MESSAGE_SIZE];
+
+      ByteBuffer bb = ByteBuffer.wrap(body);
+
+      for (int j = 1; j <= MESSAGE_SIZE; j++) {
+         bb.put(getSamplebyte(j));
+      }
+
+      for (int i = 0; i < numberOfMessages; i++) {
+         message = session.createMessage(true);
+
+         ActiveMQBuffer bodyLocal = message.getBodyBuffer();
+
+         bodyLocal.writeBytes(body);
+
+         producer.send(message);
+         if (i % 1000 == 0) {
+            session.commit();
+         }
+      }
+      session.commit();
+
+      session = sf.createSession(false, false, false);
+
+      session.start();
+
+      assertEquals(numberOfMessages, queue.getMessageCount());
+
+      // The consumer has to be created after the queue.getMessageCount assertion
+      // otherwise delivery could alter the messagecount and give us a false failure
+      ClientConsumer consumer = session.createConsumer(PagingOMETest.ADDRESS);
+      ClientMessage msg = null;
+
+      msg = consumer.receive(1000);
+
+      failureActive = true;
+      msg.individualAcknowledge();
+      try {
+         session.commit();
+         Assert.fail("exception expected");
+      }
+      catch (Exception expected) {
+      }
+      failureActive = false;
+      session.rollback();
+
+      session.close();
+
+      sf.close();
+
+      locator.close();
+
+      server.stop();
+
+      server.start();
+
+      locator = createInVMNonHALocator();
+
+      locator.setBlockOnNonDurableSend(true);
+      locator.setBlockOnDurableSend(true);
+      locator.setBlockOnAcknowledge(false);
+      locator.setConsumerWindowSize(0);
+
+      sf = createSessionFactory(locator);
+
+      session = sf.createSession(false, false, false);
+
+      consumer = session.createConsumer(PagingOMETest.ADDRESS);
+
+      session.start();
+
+      for (int i = 0; i < numberOfMessages; i++) {
+         msg = consumer.receive(1000);
+         Assert.assertNotNull(msg);
+         msg.individualAcknowledge();
+      }
+      Assert.assertNull(consumer.receiveImmediate());
+      session.commit();
+
+      session.close();
+      sf.close();
+      server.stop();
+
+   }
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a5a993ed/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
----------------------------------------------------------------------
diff --git a/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
b/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
index 6244330..6351357 100644
--- a/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
+++ b/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
@@ -257,10 +257,6 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
       }
 
       @Override
-      public void criticalException(Throwable e) {
-      }
-
-      @Override
       public int getNumberOfPages() {
          return 0;
       }


Mime
View raw message