ARTEMIS-4233 Disabling RefCounting debug and error after paged messages

apache · Apr 12, 2023 · 4eb978b · 4eb978b
1 parent 7892841
commit 4eb978b
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 5 deletions.
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/RefCountMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/RefCountMessage.java
@@ -139,6 +139,8 @@ String debugLocations() {
 
    private volatile boolean released = false;
 
+   private volatile boolean errorCheck = true;
+
    /** has the refCount fired the action already? */
    public boolean isReleased() {
       return released;
@@ -203,7 +205,7 @@ protected void onUp() {
 
    protected void released() {
       released = true;
-      accountedFor();
+      disableErrorCheck();
    }
 
    void runOnLeak(Runnable run) {
@@ -212,7 +214,8 @@ void runOnLeak(Runnable run) {
       }
    }
 
-   public void accountedFor() {
+   public void disableErrorCheck() {
+      this.errorCheck = false;
       if (debugStatus != null) {
          debugStatus.accountedFor();
       }
@@ -276,7 +279,7 @@ public int durableDown() {
          return parentRef.durableDown();
       }
       int count = DURABLE_REF_COUNT_UPDATER.decrementAndGet(this);
-      if (count < 0) {
+      if (errorCheck && count < 0) {
          reportNegativeCount();
       }
       onDown();
@@ -295,7 +298,7 @@ public int refDown() {
          return parentRef.refDown();
       }
       int count = REF_COUNT_UPDATER.decrementAndGet(this);
-      if (count < 0) {
+      if (errorCheck && count < 0) {
          reportNegativeCount();
       }
       onDown();

diff --git a/...ocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java b/...ocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java
@@ -445,7 +445,7 @@ public void releaseResources(boolean sync, boolean sendEvent) {
 
    @Override
    public void deleteFile() throws Exception {
-      accountedFor(); // if LargeServerMessage.DEBUG this will make sure this message is not reported
+      disableErrorCheck(); // if LargeServerMessage.DEBUG this will make sure this message is not reported
       largeBody.deleteFile();
    }
 

diff --git a/...s-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java b/...s-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java
@@ -23,6 +23,7 @@
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ICoreMessage;
 import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
 import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.paging.cursor.PagePosition;
 import org.apache.activemq.artemis.core.paging.cursor.impl.PagePositionImpl;
@@ -107,6 +108,7 @@ public PagedMessageImpl(final Message message, final long[] queueIDs) {
       this.queueIDs = queueIDs;
       this.message = message;
       this.storedSize = 0;
+      checkLargeMessage();
    }
 
    public PagedMessageImpl(int storedSize, StorageManager storageManager) {
@@ -170,6 +172,7 @@ public void initMessage(StorageManager storage) {
          lgMessage.setPaged();
          this.message = lgMessage.toMessage();
          largeMessageLazyData = null;
+         checkLargeMessage();
       } else {
          if (message != null && message instanceof LargeServerMessage) {
             ((LargeServerMessage)message).setStorageManager(storageManager);
@@ -222,6 +225,7 @@ public void decode(final ActiveMQBuffer buffer) {
          }
       }
 
+      checkLargeMessage();
 
       int queueIDsSize = buffer.readInt();
 
@@ -232,6 +236,14 @@ public void decode(final ActiveMQBuffer buffer) {
       }
    }
 
+   private void checkLargeMessage() {
+      if (message != null && message.isLargeMessage()) {
+         // large messages will be read and released multiple times, we have to disable their checks
+         // also we will ack them without refUp, so they will be reported negative
+         ((RefCountMessage) message).disableErrorCheck();
+      }
+   }
+
    @Override
    public void encode(final ActiveMQBuffer buffer) {
       buffer.writeLong(transactionID);

diff --git a/.../src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java b/.../src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java
@@ -2086,6 +2086,10 @@ public void testReceiveMultipleMessages() throws Exception {
    // JBPAPP-6237
    @Test
    public void testPageOnLargeMessageMultipleQueues() throws Exception {
+
+      AssertionLoggerHandler.startCapture();
+      runAfter(AssertionLoggerHandler::stopCapture);
+
       Configuration config = createDefaultConfig(isNetty());
 
       final int PAGE_MAX = 20 * 1024;
@@ -2208,6 +2212,9 @@ public void testPageOnLargeMessageMultipleQueues() throws Exception {
 
          session.close();
       }
+
+      // Reference Counting negative errrors
+      Assert.assertFalse(AssertionLoggerHandler.findText("AMQ214034"));
    }
 
    // JBPAPP-6237