activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [4/6] activemq-artemis git commit: ARTEMIS-1663 - Add new message count and size metrics
Date Thu, 08 Feb 2018 17:10:23 GMT
ARTEMIS-1663 - Add new message count and size metrics

Adding new metrics for tracking message counts and sizes on a Queue.
This includes tracking metrics for pending, delivering and scheduled
messages.  The paging store also tracks message size now.


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

Branch: refs/heads/master
Commit: ea70af15a3fbfc1b6ac86589e1e2e04a79ca3e23
Parents: 2eac195
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Mon Feb 5 13:24:31 2018 -0500
Committer: Clebert Suconic <clebertsuconic@apache.org>
Committed: Thu Feb 8 11:35:12 2018 -0500

----------------------------------------------------------------------
 .../activemq/artemis/api/core/Message.java      |  11 +-
 .../api/core/management/QueueControl.java       |  58 ++
 .../artemis/core/message/impl/CoreMessage.java  |   5 +
 .../core/message/impl/MessageInternalImpl.java  |   5 +
 .../artemis/core/journal/impl/JournalImpl.java  |   6 +-
 .../protocol/amqp/broker/AMQPMessage.java       |  13 +-
 .../core/protocol/openwire/OpenwireMessage.java |   9 +-
 .../core/management/impl/QueueControlImpl.java  | 110 +++-
 .../artemis/core/paging/PagedMessage.java       |  10 +
 .../core/paging/cursor/PagePosition.java        |   4 +
 .../core/paging/cursor/PageSubscription.java    |   9 +
 .../paging/cursor/PageSubscriptionCounter.java  |  16 +-
 .../core/paging/cursor/PagedReferenceImpl.java  |  19 +-
 .../paging/cursor/impl/PagePositionImpl.java    |  22 +
 .../impl/PageSubscriptionCounterImpl.java       | 146 ++++-
 .../cursor/impl/PageSubscriptionImpl.java       |  69 +-
 .../core/paging/impl/PagedMessageImpl.java      |   6 +
 .../core/paging/impl/PagingStoreImpl.java       |  11 +-
 .../core/persistence/StorageManager.java        |   6 +-
 .../journal/AbstractJournalStorageManager.java  |  37 +-
 .../impl/journal/DescribeJournal.java           |  61 +-
 .../impl/journal/LargeServerMessageImpl.java    |  10 +-
 .../journal/codec/PageCountPendingImpl.java     |   5 +-
 .../impl/journal/codec/PageCountRecord.java     |  18 +-
 .../impl/journal/codec/PageCountRecordInc.java  |  18 +-
 .../impl/nullpm/NullStorageManager.java         |   6 +-
 .../core/postoffice/impl/PostOfficeImpl.java    |   2 +-
 .../core/server/ActiveMQServerLogger.java       |   4 +
 .../artemis/core/server/MessageReference.java   |  11 +
 .../activemq/artemis/core/server/Queue.java     |  34 +-
 .../core/server/ScheduledDeliveryHandler.java   |   6 +
 .../core/server/cluster/impl/BridgeImpl.java    |   2 +-
 .../core/server/impl/LastValueQueue.java        |  10 +-
 .../core/server/impl/MessageReferenceImpl.java  |   8 +-
 .../server/impl/PostOfficeJournalLoader.java    |  13 +-
 .../artemis/core/server/impl/QueueImpl.java     | 144 +++-
 .../server/impl/QueuePendingMessageMetrics.java | 147 +++++
 .../impl/ScheduledDeliveryHandlerImpl.java      |  29 +-
 .../impl/ScheduledDeliveryHandlerTest.java      |  75 ++-
 .../transaction/impl/TransactionImplTest.java   |   6 +-
 .../main/resources/metrics/queueMetrics.groovy  |  37 ++
 .../main/resources/servers/artemisServer.groovy |   6 +
 .../compatibility/JournalCompatibilityTest.java |  43 +-
 .../tests/compatibility/VersionedBaseTest.java  |   7 +-
 .../integration/client/AcknowledgeTest.java     |   5 +
 .../integration/client/SendAckFailTest.java     |  12 +-
 .../management/ManagementTestBase.java          |  15 +
 .../management/QueueControlTest.java            | 452 +++++++------
 .../management/QueueControlUsingCoreTest.java   |  55 +-
 .../integration/paging/PagingCounterTest.java   |  21 +-
 .../AbstractPersistentStatTestSupport.java      | 213 ++++++
 .../metrics/JournalPageCountSizeTest.java       | 144 ++++
 .../metrics/JournalPendingMessageTest.java      | 651 +++++++++++++++++++
 .../unit/core/postoffice/impl/FakeQueue.java    |  47 +-
 .../core/server/impl/fakes/FakeConsumer.java    |   4 +-
 55 files changed, 2502 insertions(+), 391 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
index d24cd95..031c426 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
@@ -657,6 +657,15 @@ public interface Message {
 
    int getMemoryEstimate();
 
-
+   /**
+    * This is the size of the message when persisted on disk which is used for metrics tracking
+    * Note that even if the message itself is not persisted on disk (ie non-durable) this value is
+    * still used for metrics tracking
+    * If a normal message it will be the encoded message size
+    * If a large message it will be encoded message size + large message body size
+    * @return
+    * @throws ActiveMQException
+    */
+   long getPersistentSize() throws ActiveMQException;
 
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
index 447417f..2578684 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/QueueControl.java
@@ -81,12 +81,52 @@ public interface QueueControl {
    long getMessageCount();
 
    /**
+    * Returns the persistent size of all messages currently in this queue. The persistent size of a message
+    * is the amount of space the message would take up on disk which is used to track how much data there
+    * is to consume on this queue
+    */
+   @Attribute(desc = "persistent size of all messages (including durable and non-durable) currently in this queue (includes scheduled, paged, and in-delivery messages)")
+   long getPersistentSize();
+
+   /**
+    * Returns the number of durable messages currently in this queue.
+    */
+   @Attribute(desc = "number of durable messages currently in this queue (includes scheduled, paged, and in-delivery messages)")
+   long getDurableMessageCount();
+
+   /**
+    * Returns the persistent size of durable messages currently in this queue. The persistent size of a message
+    * is the amount of space the message would take up on disk which is used to track how much data there
+    * is to consume on this queue
+    */
+   @Attribute(desc = "persistent size of durable messages currently in this queue (includes scheduled, paged, and in-delivery messages)")
+   long getDurablePersistentSize();
+
+   /**
     * Returns the number of scheduled messages in this queue.
     */
    @Attribute(desc = "number of scheduled messages in this queue")
    long getScheduledCount();
 
    /**
+    * Returns the size of scheduled messages in this queue.
+    */
+   @Attribute(desc = "persistent size of scheduled messages in this queue")
+   long getScheduledSize();
+
+   /**
+    * Returns the number of durable scheduled messages in this queue.
+    */
+   @Attribute(desc = "number of durable scheduled messages in this queue")
+   long getDurableScheduledCount();
+
+   /**
+    * Returns the size of durable scheduled messages in this queue.
+    */
+   @Attribute(desc = "persistent size of durable scheduled messages in this queue")
+   long getDurableScheduledSize();
+
+   /**
     * Returns the number of consumers consuming messages from this queue.
     */
    @Attribute(desc = "number of consumers consuming messages from this queue")
@@ -99,6 +139,24 @@ public interface QueueControl {
    int getDeliveringCount();
 
    /**
+    * Returns the persistent size of messages that this queue is currently delivering to its consumers.
+    */
+   @Attribute(desc = "persistent size of messages that this queue is currently delivering to its consumers")
+   long getDeliveringSize();
+
+   /**
+    * Returns the number of durable messages that this queue is currently delivering to its consumers.
+    */
+   @Attribute(desc = "number of durable messages that this queue is currently delivering to its consumers")
+   int getDurableDeliveringCount();
+
+   /**
+    * Returns the size of durable messages that this queue is currently delivering to its consumers.
+    */
+   @Attribute(desc = "persistent size of durable messages that this queue is currently delivering to its consumers")
+   long getDurableDeliveringSize();
+
+   /**
     * Returns the number of messages added to this queue since it was created.
     */
    @Attribute(desc = "number of messages added to this queue since it was created")

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
index 0fb7c3e..172cc18 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
@@ -1150,4 +1150,9 @@ public class CoreMessage extends RefCountMessage implements ICoreMessage {
    private SimpleString.StringSimpleStringPool getPropertyValuesPool() {
       return coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesStringSimpleStringPools().getPropertyValuesPool();
    }
+
+   @Override
+   public long getPersistentSize() throws ActiveMQException {
+      return getEncodeSize();
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageInternalImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageInternalImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageInternalImpl.java
index 56ff816..17cb828 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageInternalImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageInternalImpl.java
@@ -695,4 +695,9 @@ public class MessageInternalImpl implements MessageInternal {
       return new TypedProperties(message.getTypedProperties());
    }
 
+   @Override
+   public long getPersistentSize() throws ActiveMQException {
+      return message.getPersistentSize();
+   }
+
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
----------------------------------------------------------------------
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
index 88204d4..34ee72e 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
@@ -191,7 +191,7 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
 
    private Executor appendExecutor = null;
 
-   private ConcurrentHashSet<CountDownLatch> latches = new ConcurrentHashSet<>();
+   private final ConcurrentHashSet<CountDownLatch> latches = new ConcurrentHashSet<>();
 
    private final ExecutorFactory providedIOThreadPool;
    protected ExecutorFactory ioExecutorFactory;
@@ -2413,7 +2413,7 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
                                                  final List<JournalFile> newFiles,
                                                  final List<Pair<String, String>> renames) throws Exception {
 
-      return JournalCompactor.writeControlFile(fileFactory, files, newFiles, renames);
+      return AbstractJournalUpdateTask.writeControlFile(fileFactory, files, newFiles, renames);
    }
 
 
@@ -2763,7 +2763,7 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
       ArrayList<String> newFiles = new ArrayList<>();
       ArrayList<Pair<String, String>> renames = new ArrayList<>();
 
-      SequentialFile controlFile = JournalCompactor.readControlFile(fileFactory, dataFiles, newFiles, renames);
+      SequentialFile controlFile = AbstractJournalUpdateTask.readControlFile(fileFactory, dataFiles, newFiles, renames);
       if (controlFile != null) {
          for (String dataFile : dataFiles) {
             SequentialFile file = fileFactory.createSequentialFile(dataFile);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
----------------------------------------------------------------------
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 cdab412..2d72cf9 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
@@ -24,10 +24,8 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.Unpooled;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
 import org.apache.activemq.artemis.api.core.ICoreMessage;
 import org.apache.activemq.artemis.api.core.RefCountMessage;
@@ -60,6 +58,10 @@ import org.apache.qpid.proton.codec.WritableBuffer;
 import org.apache.qpid.proton.message.Message;
 import org.apache.qpid.proton.message.impl.MessageImpl;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+
 // see https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format
 public class AMQPMessage extends RefCountMessage {
 
@@ -1179,4 +1181,9 @@ public class AMQPMessage extends RefCountMessage {
    private SimpleString.StringSimpleStringPool getPropertyValuesPool() {
       return coreMessageObjectPools == null ? null : coreMessageObjectPools.getPropertiesStringSimpleStringPools().getPropertyValuesPool();
    }
+
+   @Override
+   public long getPersistentSize() throws ActiveMQException {
+      return getEncodeSize();
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
index c63fe19..45e8953 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
@@ -19,8 +19,8 @@ package org.apache.activemq.artemis.core.protocol.openwire;
 
 import java.util.Set;
 
-import io.netty.buffer.ByteBuf;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
 import org.apache.activemq.artemis.api.core.ICoreMessage;
 import org.apache.activemq.artemis.api.core.Message;
@@ -29,6 +29,8 @@ import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
 import org.apache.activemq.artemis.core.persistence.Persister;
 
+import io.netty.buffer.ByteBuf;
+
 // TODO: Implement this
 public class OpenwireMessage implements Message {
 
@@ -496,4 +498,9 @@ public class OpenwireMessage implements Message {
    public int getMemoryEstimate() {
       return 0;
    }
+
+   @Override
+   public long getPersistentSize() throws ActiveMQException {
+      return 0;
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
index cefcbf9..e678ab8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
@@ -227,6 +227,42 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
    }
 
    @Override
+   public long getPersistentSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getPersistentSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public long getDurableMessageCount() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurableMessageCount();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public long getDurablePersistentSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurablePersistentSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
    public int getConsumerCount() {
       checkStarted();
 
@@ -251,6 +287,42 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
    }
 
    @Override
+   public long getDeliveringSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDeliveringSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public int getDurableDeliveringCount() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurableDeliveringCount();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public long getDurableDeliveringSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurableDeliveringSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
    public long getMessagesAdded() {
       checkStarted();
 
@@ -323,6 +395,42 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
    }
 
    @Override
+   public long getScheduledSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getScheduledSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public long getDurableScheduledCount() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurableScheduledCount();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
+   public long getDurableScheduledSize() {
+      checkStarted();
+
+      clearIO();
+      try {
+         return queue.getDurableScheduledSize();
+      } finally {
+         blockOnIO();
+      }
+   }
+
+   @Override
    public String getDeadLetterAddress() {
       checkStarted();
 
@@ -998,7 +1106,7 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
       try {
          long index = 0;
          long start = (page - 1) * pageSize;
-         long end = Math.min((long)(page * pageSize), queue.getMessageCount());
+         long end = Math.min(page * pageSize, queue.getMessageCount());
 
          ArrayList<CompositeData> c = new ArrayList<>();
          Filter thefilter = FilterImpl.createFilter(filter);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagedMessage.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagedMessage.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagedMessage.java
index 0124f09..5b39691 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagedMessage.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagedMessage.java
@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.artemis.core.paging;
 
+import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.journal.EncodingSupport;
 import org.apache.activemq.artemis.core.persistence.StorageManager;
@@ -38,4 +39,13 @@ public interface PagedMessage extends EncodingSupport {
    void initMessage(StorageManager storageManager);
 
    long getTransactionID();
+
+   /**
+    * This is the size of the message when persisted on disk and is used for metrics tracking
+    * If a normal message it will be the encoded message size
+    * If a large message it will be encoded message size + large message body size
+    * @return
+    * @throws ActiveMQException
+    */
+   long getPersistentSize() throws ActiveMQException;
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagePosition.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagePosition.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagePosition.java
index 00955b7..a9e0537 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagePosition.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PagePosition.java
@@ -28,6 +28,10 @@ public interface PagePosition extends Comparable<PagePosition> {
 
    int getMessageNr();
 
+   long getPersistentSize();
+
+   void setPersistentSize(long persistentSize);
+
    PagePosition nextMessage();
 
    PagePosition nextPage();

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 985f563..b11362d 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
@@ -44,6 +44,8 @@ public interface PageSubscription {
 
    long getMessageCount();
 
+   long getPersistentSize();
+
    long getId();
 
    boolean isPersistent();
@@ -161,4 +163,11 @@ public interface PageSubscription {
     * @throws Exception
     */
    void onDeletePage(Page deletedPage) throws Exception;
+
+   long getDeliveredCount();
+
+   long getDeliveredSize();
+
+   void incrementDeliveredSize(long size);
+
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
index 37cdb3b..33b744f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscriptionCounter.java
@@ -26,13 +26,17 @@ public interface PageSubscriptionCounter {
 
    long getValue();
 
-   void increment(Transaction tx, int add) throws Exception;
+   long getPersistentSizeAdded();
 
-   void loadValue(long recordValueID, long value);
+   long getPersistentSize();
 
-   void loadInc(long recordInd, int add);
+   void increment(Transaction tx, int add, long persistentSize) throws Exception;
 
-   void applyIncrementOnTX(Transaction tx, long recordID, int add);
+   void loadValue(long recordValueID, long value, long persistentSize);
+
+   void loadInc(long recordInd, int add, long persistentSize);
+
+   void applyIncrementOnTX(Transaction tx, long recordID, int add, long persistentSize);
 
    /**
     * This will process the reload
@@ -43,12 +47,12 @@ public interface PageSubscriptionCounter {
     * @param id
     * @param variance
     */
-   void addInc(long id, int variance);
+   void addInc(long id, int variance, long size);
 
    // used when deleting the counter
    void delete() throws Exception;
 
-   void pendingCounter(Page page, int increment) throws Exception;
+   void pendingCounter(Page page, int increment, long persistentSize) throws Exception;
 
    // used when leaving page mode, so the counters are deleted in batches
    // for each queue on the address

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 42c5423..f5d49cf 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
@@ -59,6 +59,8 @@ public class PagedReferenceImpl extends LinkedListImpl.Node<PagedReferenceImpl>
 
    private final long messageID;
 
+   private long messageSize = -1;
+
    @Override
    public Object getProtocolData() {
       return protocolData;
@@ -104,6 +106,9 @@ public class PagedReferenceImpl extends LinkedListImpl.Node<PagedReferenceImpl>
       this.largeMessage = message.getMessage().isLargeMessage();
       this.transactionID = message.getTransactionID();
       this.messageID = message.getMessage().getMessageID();
+
+      //pre-cache the message size so we don't have to reload the message later if it is GC'd
+      getPersistentSize();
    }
 
    @Override
@@ -191,7 +196,7 @@ public class PagedReferenceImpl extends LinkedListImpl.Node<PagedReferenceImpl>
 
    @Override
    public void handled() {
-      getQueue().referenceHandled();
+      getQueue().referenceHandled(this);
    }
 
    @Override
@@ -280,4 +285,16 @@ public class PagedReferenceImpl extends LinkedListImpl.Node<PagedReferenceImpl>
       return messageID;
    }
 
+   @Override
+   public long getPersistentSize() {
+      if (messageSize == -1) {
+         try {
+            messageSize = getPagedMessage().getPersistentSize();
+         } catch (Throwable e) {
+            ActiveMQServerLogger.LOGGER.errorCalculatePersistentSize(e);
+         }
+      }
+      return messageSize;
+   }
+
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PagePositionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PagePositionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PagePositionImpl.java
index 076f872..52d1c83 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PagePositionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PagePositionImpl.java
@@ -37,6 +37,12 @@ public class PagePositionImpl implements PagePosition {
    private long recordID = -1;
 
    /**
+    * Optional size value that can be set to specify the peristent size of the message
+    * for metrics tracking purposes
+    */
+   private long persistentSize;
+
+   /**
     * @param pageNr
     * @param messageNr
     */
@@ -82,6 +88,22 @@ public class PagePositionImpl implements PagePosition {
       return messageNr;
    }
 
+   /**
+    * @return the persistentSize
+    */
+   @Override
+   public long getPersistentSize() {
+      return persistentSize;
+   }
+
+   /**
+    * @param persistentSize the persistentSize to set
+    */
+   @Override
+   public void setPersistentSize(long persistentSize) {
+      this.persistentSize = persistentSize;
+   }
+
    @Override
    public int compareTo(PagePosition o) {
       if (pageNr > o.getPageNr()) {

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
index 01ad778..3bb56f8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionCounterImpl.java
@@ -21,10 +21,10 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
-import org.apache.activemq.artemis.api.core.Pair;
 import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
 import org.apache.activemq.artemis.core.paging.cursor.PageSubscriptionCounter;
 import org.apache.activemq.artemis.core.paging.impl.Page;
@@ -60,10 +60,13 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
    private final Executor executor;
 
    private final AtomicLong value = new AtomicLong(0);
+   private final AtomicLong persistentSize = new AtomicLong(0);
 
    private final AtomicLong added = new AtomicLong(0);
+   private final AtomicLong addedPersistentSize = new AtomicLong(0);
 
    private final AtomicLong pendingValue = new AtomicLong(0);
+   private final AtomicLong pendingPersistentSize = new AtomicLong(0);
 
    private final LinkedList<Long> incrementRecords = new LinkedList<>();
 
@@ -71,9 +74,9 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
    // we will recount a page case we still see pending records
    // as soon as we close a page we remove these records replacing by a regular page increment record
    // A Map per pageID, each page will have a set of IDs, with the increment on each one
-   private final Map<Long, Pair<Long, AtomicInteger>> pendingCounters = new HashMap<>();
+   private final Map<Long, PendingCounter> pendingCounters = new HashMap<>();
 
-   private LinkedList<Pair<Long, Integer>> loadList;
+   private LinkedList<PendingCounter> loadList;
 
    private final Runnable cleanupCheck = new Runnable() {
       @Override
@@ -104,6 +107,16 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
       return value.get() + pendingValue.get();
    }
 
+   @Override
+   public long getPersistentSizeAdded() {
+      return addedPersistentSize.get() + pendingPersistentSize.get();
+   }
+
+   @Override
+   public long getPersistentSize() {
+      return persistentSize.get() + pendingPersistentSize.get();
+   }
+
    /**
     * This is used only on non transactional paging
     *
@@ -112,24 +125,25 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
     * @throws Exception
     */
    @Override
-   public synchronized void pendingCounter(Page page, int increment) throws Exception {
+   public synchronized void pendingCounter(Page page, int increment, long size) throws Exception {
       if (!persistent) {
          return; // nothing to be done
       }
 
-      Pair<Long, AtomicInteger> pendingInfo = pendingCounters.get((long) page.getPageId());
+      PendingCounter pendingInfo = pendingCounters.get((long) page.getPageId());
       if (pendingInfo == null) {
          // We have to make sure this is sync here
          // not syncing this to disk may cause the page files to be out of sync on pages.
          // we can't afford the case where a page file is written without a record here
          long id = storage.storePendingCounter(this.subscriptionID, page.getPageId(), increment);
-         pendingInfo = new Pair<>(id, new AtomicInteger(1));
+         pendingInfo = new PendingCounter(id, increment, size);
          pendingCounters.put((long) page.getPageId(), pendingInfo);
       } else {
-         pendingInfo.getB().addAndGet(increment);
+         pendingInfo.addAndGet(increment, size);
       }
 
       pendingValue.addAndGet(increment);
+      pendingPersistentSize.addAndGet(size);
 
       page.addPendingCounter(this);
    }
@@ -141,23 +155,25 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
     */
    @Override
    public void cleanupNonTXCounters(final long pageID) throws Exception {
-      Pair<Long, AtomicInteger> pendingInfo;
+      PendingCounter pendingInfo;
       synchronized (this) {
          pendingInfo = pendingCounters.remove(pageID);
       }
 
       if (pendingInfo != null) {
-         final AtomicInteger valueCleaned = pendingInfo.getB();
+         final int valueCleaned = pendingInfo.getCount();
+         final long valueSizeCleaned = pendingInfo.getPersistentSize();
          Transaction tx = new TransactionImpl(storage);
-         storage.deletePendingPageCounter(tx.getID(), pendingInfo.getA());
+         storage.deletePendingPageCounter(tx.getID(), pendingInfo.getId());
 
          // To apply the increment of the value just being cleaned
-         increment(tx, valueCleaned.get());
+         increment(tx, valueCleaned, valueSizeCleaned);
 
          tx.addOperation(new TransactionOperationAbstract() {
             @Override
             public void afterCommit(Transaction tx) {
-               pendingValue.addAndGet(-valueCleaned.get());
+               pendingValue.addAndGet(-valueCleaned);
+               pendingPersistentSize.updateAndGet(val -> val >= valueSizeCleaned ? val - valueSizeCleaned : 0);
             }
          });
 
@@ -166,21 +182,21 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
    }
 
    @Override
-   public void increment(Transaction tx, int add) throws Exception {
+   public void increment(Transaction tx, int add, long size) throws Exception {
       if (tx == null) {
          if (persistent) {
-            long id = storage.storePageCounterInc(this.subscriptionID, add);
-            incrementProcessed(id, add);
+            long id = storage.storePageCounterInc(this.subscriptionID, add, size);
+            incrementProcessed(id, add, size);
          } else {
-            incrementProcessed(-1, add);
+            incrementProcessed(-1, add, size);
          }
       } else {
          if (persistent) {
             tx.setContainsPersistent();
-            long id = storage.storePageCounterInc(tx.getID(), this.subscriptionID, add);
-            applyIncrementOnTX(tx, id, add);
+            long id = storage.storePageCounterInc(tx.getID(), this.subscriptionID, add, size);
+            applyIncrementOnTX(tx, id, add, size);
          } else {
-            applyIncrementOnTX(tx, -1, add);
+            applyIncrementOnTX(tx, -1, add, size);
          }
       }
    }
@@ -193,7 +209,7 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
     * @param add
     */
    @Override
-   public void applyIncrementOnTX(Transaction tx, long recordID1, int add) {
+   public void applyIncrementOnTX(Transaction tx, long recordID1, int add, long size) {
       CounterOperations oper = (CounterOperations) tx.getProperty(TransactionPropertyIndexes.PAGE_COUNT_INC);
 
       if (oper == null) {
@@ -202,22 +218,24 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
          tx.addOperation(oper);
       }
 
-      oper.operations.add(new ItemOper(this, recordID1, add));
+      oper.operations.add(new ItemOper(this, recordID1, add, size));
    }
 
    @Override
-   public synchronized void loadValue(final long recordID1, final long value1) {
+   public synchronized void loadValue(final long recordID1, final long value1, long size) {
       if (this.subscription != null) {
          // it could be null on testcases... which is ok
          this.subscription.notEmpty();
       }
       this.value.set(value1);
       this.added.set(value1);
+      this.persistentSize.set(size);
+      this.addedPersistentSize.set(size);
       this.recordID = recordID1;
    }
 
-   public synchronized void incrementProcessed(long id, int add) {
-      addInc(id, add);
+   public synchronized void incrementProcessed(long id, int add, long size) {
+      addInc(id, add, size);
       if (incrementRecords.size() > FLUSH_COUNTER) {
          executor.execute(cleanupCheck);
       }
@@ -259,12 +277,12 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
    }
 
    @Override
-   public void loadInc(long id, int add) {
+   public void loadInc(long id, int add, long size) {
       if (loadList == null) {
          loadList = new LinkedList<>();
       }
 
-      loadList.add(new Pair<>(id, add));
+      loadList.add(new PendingCounter(id, add, size));
    }
 
    @Override
@@ -275,10 +293,12 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
             subscription.notEmpty();
          }
 
-         for (Pair<Long, Integer> incElement : loadList) {
-            value.addAndGet(incElement.getB());
-            added.addAndGet(incElement.getB());
-            incrementRecords.add(incElement.getA());
+         for (PendingCounter incElement : loadList) {
+            value.addAndGet(incElement.getCount());
+            added.addAndGet(incElement.getCount());
+            persistentSize.addAndGet(incElement.getPersistentSize());
+            addedPersistentSize.addAndGet(incElement.getPersistentSize());
+            incrementRecords.add(incElement.getId());
          }
          loadList.clear();
          loadList = null;
@@ -286,11 +306,15 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
    }
 
    @Override
-   public synchronized void addInc(long id, int variance) {
+   public synchronized void addInc(long id, int variance, long size) {
       value.addAndGet(variance);
+      this.persistentSize.addAndGet(size);
       if (variance > 0) {
          added.addAndGet(variance);
       }
+      if (size > 0) {
+         addedPersistentSize.addAndGet(size);
+      }
       if (id >= 0) {
          incrementRecords.add(id);
       }
@@ -310,11 +334,13 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
       ArrayList<Long> deleteList;
 
       long valueReplace;
+      long sizeReplace;
       synchronized (this) {
          if (incrementRecords.size() <= FLUSH_COUNTER) {
             return;
          }
          valueReplace = value.get();
+         sizeReplace = persistentSize.get();
          deleteList = new ArrayList<>(incrementRecords);
          incrementRecords.clear();
       }
@@ -332,7 +358,7 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
             storage.deletePageCounter(txCleanup, recordID);
          }
 
-         newRecordID = storage.storePageCounter(txCleanup, subscriptionID, valueReplace);
+         newRecordID = storage.storePageCounter(txCleanup, subscriptionID, valueReplace, sizeReplace);
 
          if (logger.isTraceEnabled()) {
             logger.trace("Replacing page-counter record = " + recordID + " by record = " + newRecordID + " on subscriptionID = " + this.subscriptionID + " for queue = " + this.subscription.getQueue().getName());
@@ -354,10 +380,11 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
 
    private static class ItemOper {
 
-      private ItemOper(PageSubscriptionCounterImpl counter, long id, int add) {
+      private ItemOper(PageSubscriptionCounterImpl counter, long id, int add, long persistentSize) {
          this.counter = counter;
          this.id = id;
          this.amount = add;
+         this.persistentSize = persistentSize;
       }
 
       PageSubscriptionCounterImpl counter;
@@ -365,6 +392,8 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
       long id;
 
       int amount;
+
+      long persistentSize;
    }
 
    private static class CounterOperations extends TransactionOperationAbstract implements TransactionOperation {
@@ -374,8 +403,55 @@ public class PageSubscriptionCounterImpl implements PageSubscriptionCounter {
       @Override
       public void afterCommit(Transaction tx) {
          for (ItemOper oper : operations) {
-            oper.counter.incrementProcessed(oper.id, oper.amount);
+            oper.counter.incrementProcessed(oper.id, oper.amount, oper.persistentSize);
          }
       }
    }
+
+   private static class PendingCounter {
+      private static final AtomicIntegerFieldUpdater<PendingCounter> COUNT_UPDATER =
+            AtomicIntegerFieldUpdater.newUpdater(PendingCounter.class, "count");
+
+      private static final AtomicLongFieldUpdater<PendingCounter> SIZE_UPDATER =
+            AtomicLongFieldUpdater.newUpdater(PendingCounter.class, "persistentSize");
+
+      private final long id;
+      private volatile int count;
+      private volatile long persistentSize;
+
+      /**
+       * @param id
+       * @param count
+       * @param size
+       */
+      PendingCounter(long id, int count, long persistentSize) {
+         super();
+         this.id = id;
+         this.count = count;
+         this.persistentSize = persistentSize;
+      }
+      /**
+       * @return the id
+       */
+      public long getId() {
+         return id;
+      }
+      /**
+       * @return the count
+       */
+      public int getCount() {
+         return count;
+      }
+      /**
+       * @return the size
+       */
+      public long getPersistentSize() {
+         return persistentSize;
+      }
+
+      public void addAndGet(int count, long persistentSize) {
+         COUNT_UPDATER.addAndGet(this, count);
+         SIZE_UPDATER.addAndGet(this, persistentSize);
+      }
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 24c69be..924aace 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
@@ -96,6 +96,8 @@ final class PageSubscriptionImpl implements PageSubscription {
 
    private final AtomicLong deliveredCount = new AtomicLong(0);
 
+   private final AtomicLong deliveredSize = new AtomicLong(0);
+
    PageSubscriptionImpl(final PageCursorProvider cursorProvider,
                         final PagingStore pageStore,
                         final StorageManager store,
@@ -178,6 +180,18 @@ final class PageSubscriptionImpl implements PageSubscription {
    }
 
    @Override
+   public long getPersistentSize() {
+      if (empty) {
+         return 0;
+      } else {
+         //A negative value could happen if an old journal was loaded that didn't have
+         //size metrics for old records
+         long messageSize = counter.getPersistentSize() - deliveredSize.get();
+         return messageSize > 0 ? messageSize : 0;
+      }
+   }
+
+   @Override
    public PageSubscriptionCounter getCounter() {
       return counter;
    }
@@ -439,7 +453,7 @@ final class PageSubscriptionImpl implements PageSubscription {
    public void ackTx(final Transaction tx, final PagedReference reference) throws Exception {
       confirmPosition(tx, reference.getPosition());
 
-      counter.increment(tx, -1);
+      counter.increment(tx, -1, -getPersistentSize(reference));
 
       PageTransactionInfo txInfo = getPageTransaction(reference);
       if (txInfo != null) {
@@ -831,6 +845,12 @@ final class PageSubscriptionImpl implements PageSubscription {
       }
 
       PageCursorInfo info = getPageInfo(position);
+      PageCache cache = info.getCache();
+      long size = 0;
+      if (cache != null) {
+         size = getPersistentSize(cache.getMessage(position.getMessageNr()));
+         position.setPersistentSize(size);
+      }
 
       logger.tracef("InstallTXCallback looking up pagePosition %s, result=%s", position, info);
 
@@ -1060,6 +1080,13 @@ final class PageSubscriptionImpl implements PageSubscription {
          }
       }
 
+      /**
+       * @return the cache
+       */
+      public PageCache getCache() {
+         return cache != null ? cache.get() : null;
+      }
+
    }
 
    private final class PageCursorTX extends TransactionOperationAbstract {
@@ -1087,6 +1114,7 @@ final class PageSubscriptionImpl implements PageSubscription {
             for (PagePosition confirmed : positions) {
                cursor.processACK(confirmed);
                cursor.deliveredCount.decrementAndGet();
+               cursor.deliveredSize.addAndGet(-confirmed.getPersistentSize());
             }
 
          }
@@ -1309,4 +1337,43 @@ final class PageSubscriptionImpl implements PageSubscription {
       public void close() {
       }
    }
+
+   /**
+    * @return the deliveredCount
+    */
+   @Override
+   public long getDeliveredCount() {
+      return deliveredCount.get();
+   }
+
+   /**
+    * @return the deliveredSize
+    */
+   @Override
+   public long getDeliveredSize() {
+      return deliveredSize.get();
+   }
+
+   @Override
+   public void incrementDeliveredSize(long size) {
+      deliveredSize.addAndGet(size);
+   }
+
+   private long getPersistentSize(PagedMessage msg) {
+      try {
+         return msg != null && msg.getPersistentSize() > 0 ? msg.getPersistentSize() : 0;
+      } catch (ActiveMQException e) {
+         logger.warn("Error computing persistent size of message: " + msg, e);
+         return 0;
+      }
+   }
+
+   private long getPersistentSize(PagedReference ref) {
+      try {
+         return ref != null && ref.getPersistentSize() > 0 ? ref.getPersistentSize() : 0;
+      } catch (ActiveMQException e) {
+         logger.warn("Error computing persistent size of message: " + ref, e);
+         return 0;
+      }
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java
index d7bd05c..3ef833d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagedMessageImpl.java
@@ -20,6 +20,7 @@ import java.util.Arrays;
 
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+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.core.paging.PagedMessage;
@@ -169,4 +170,9 @@ public class PagedMessageImpl implements PagedMessage {
          message +
          "]";
    }
+
+   @Override
+   public long getPersistentSize() throws ActiveMQException {
+      return message.getPersistentSize();
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 f1beb31..0eec5a0 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
@@ -840,7 +840,8 @@ public class PagingStoreImpl implements PagingStore {
             // the apply counter will make sure we write a record on journal
             // especially on the case for non transactional sends and paging
             // doing this will give us a possibility of recovering the page counters
-            applyPageCounters(tx, getCurrentPage(), listCtx);
+            long persistentSize = pagedMessage.getPersistentSize() > 0 ? pagedMessage.getPersistentSize() : 0;
+            applyPageCounters(tx, getCurrentPage(), listCtx, persistentSize);
 
             currentPage.write(pagedMessage);
 
@@ -906,22 +907,22 @@ public class PagingStoreImpl implements PagingStore {
     * @param ctx
     * @throws Exception
     */
-   private void applyPageCounters(Transaction tx, Page page, RouteContextList ctx) throws Exception {
+   private void applyPageCounters(Transaction tx, Page page, RouteContextList ctx, long size) throws Exception {
       List<org.apache.activemq.artemis.core.server.Queue> durableQueues = ctx.getDurableQueues();
       List<org.apache.activemq.artemis.core.server.Queue> nonDurableQueues = ctx.getNonDurableQueues();
       for (org.apache.activemq.artemis.core.server.Queue q : durableQueues) {
          if (tx == null) {
             // non transactional writes need an intermediate place
             // to avoid the counter getting out of sync
-            q.getPageSubscription().getCounter().pendingCounter(page, 1);
+            q.getPageSubscription().getCounter().pendingCounter(page, 1, size);
          } else {
             // null tx is treated through pending counters
-            q.getPageSubscription().getCounter().increment(tx, 1);
+            q.getPageSubscription().getCounter().increment(tx, 1, size);
          }
       }
 
       for (org.apache.activemq.artemis.core.server.Queue q : nonDurableQueues) {
-         q.getPageSubscription().getCounter().increment(tx, 1);
+         q.getPageSubscription().getCounter().increment(tx, 1, size);
       }
 
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
index 6defb1e..f9793d8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
@@ -336,7 +336,7 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent {
    /**
     * @return The ID with the stored counter
     */
-   long storePageCounter(long txID, long queueID, long value) throws Exception;
+   long storePageCounter(long txID, long queueID, long value, long persistentSize) throws Exception;
 
    long storePendingCounter(long queueID, long pageID, int inc) throws Exception;
 
@@ -350,13 +350,13 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent {
     * @return the ID with the increment record
     * @throws Exception
     */
-   long storePageCounterInc(long txID, long queueID, int add) throws Exception;
+   long storePageCounterInc(long txID, long queueID, int add, long persistentSize) throws Exception;
 
    /**
     * @return the ID with the increment record
     * @throws Exception
     */
-   long storePageCounterInc(long queueID, int add) throws Exception;
+   long storePageCounterInc(long queueID, int add, long size) throws Exception;
 
    /**
     * @return the bindings journal

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index 34d249e..ada5b90 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -16,7 +16,13 @@
  */
 package org.apache.activemq.artemis.core.persistence.impl.journal;
 
-import javax.transaction.xa.Xid;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.DUPLICATE_ID;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_INC;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
+
 import java.io.File;
 import java.io.FileInputStream;
 import java.security.DigestInputStream;
@@ -37,6 +43,8 @@ import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import javax.transaction.xa.Xid;
+
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
 import org.apache.activemq.artemis.api.core.Message;
@@ -109,13 +117,6 @@ import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
 import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
 import org.jboss.logging.Logger;
 
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.DUPLICATE_ID;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_INC;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
-
 /**
  * Controls access to the journals and other storage files such as the ones used to store pages and
  * large messages.  This class must control writing of any non-transient data, as it is the key point
@@ -1084,7 +1085,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                   PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
 
                   if (sub != null) {
-                     sub.getCounter().loadValue(record.id, encoding.getValue());
+                     sub.getCounter().loadValue(record.id, encoding.getValue(), encoding.getPersistentSize());
                   } else {
                      ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPage(encoding.getQueueID());
                      messageJournal.appendDeleteRecord(record.id, false);
@@ -1101,7 +1102,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                   PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
 
                   if (sub != null) {
-                     sub.getCounter().loadInc(record.id, encoding.getValue());
+                     sub.getCounter().loadInc(record.id, encoding.getValue(), encoding.getPersistentSize());
                   } else {
                      ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPageCursor(encoding.getQueueID());
                      messageJournal.appendDeleteRecord(record.id, false);
@@ -1136,6 +1137,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: {
 
                   PageCountPendingImpl pendingCountEncoding = new PageCountPendingImpl();
+
                   pendingCountEncoding.decode(buff);
                   pendingCountEncoding.setID(record.id);
 
@@ -1143,6 +1145,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                   if (pendingNonTXPageCounter != null) {
                      pendingNonTXPageCounter.add(pendingCountEncoding);
                   }
+
                   break;
                }
 
@@ -1349,11 +1352,11 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
    }
 
    @Override
-   public long storePageCounterInc(long txID, long queueID, int value) throws Exception {
+   public long storePageCounterInc(long txID, long queueID, int value, long persistentSize) throws Exception {
       readLock();
       try {
          long recordID = idGenerator.generateID();
-         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value));
+         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value, persistentSize));
          return recordID;
       } finally {
          readUnLock();
@@ -1361,11 +1364,11 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
    }
 
    @Override
-   public long storePageCounterInc(long queueID, int value) throws Exception {
+   public long storePageCounterInc(long queueID, int value, long persistentSize) throws Exception {
       readLock();
       try {
          final long recordID = idGenerator.generateID();
-         messageJournal.appendAddRecord(recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value), true, getContext());
+         messageJournal.appendAddRecord(recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value, persistentSize), true, getContext());
          return recordID;
       } finally {
          readUnLock();
@@ -1373,11 +1376,11 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
    }
 
    @Override
-   public long storePageCounter(long txID, long queueID, long value) throws Exception {
+   public long storePageCounter(long txID, long queueID, long value, long persistentSize) throws Exception {
       readLock();
       try {
          final long recordID = idGenerator.generateID();
-         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE, new PageCountRecord(queueID, value));
+         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE, new PageCountRecord(queueID, value, persistentSize));
          return recordID;
       } finally {
          readUnLock();
@@ -1789,7 +1792,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
                   PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
 
                   if (sub != null) {
-                     sub.getCounter().applyIncrementOnTX(tx, record.id, encoding.getValue());
+                     sub.getCounter().applyIncrementOnTX(tx, record.id, encoding.getValue(), encoding.getPersistentSize());
                      sub.notEmpty();
                   } else {
                      ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingACK(encoding.getQueueID());

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
index bfabc25..6cd417b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/DescribeJournal.java
@@ -16,7 +16,29 @@
  */
 package org.apache.activemq.artemis.core.persistence.impl.journal;
 
-import javax.transaction.xa.Xid;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_REF;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADDRESS_BINDING_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADDRESS_SETTING_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_MESSAGE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_MESSAGE_PROTOCOL;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_REF;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.DUPLICATE_ID;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.HEURISTIC_COMPLETION;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ID_COUNTER_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COMPLETE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_INC;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_TRANSACTION;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_BINDING_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_STATUS_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SECURITY_RECORD;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.UPDATE_DELIVERY_COUNT;
+
 import java.io.File;
 import java.io.PrintStream;
 import java.util.HashMap;
@@ -24,6 +46,8 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
+import javax.transaction.xa.Xid;
+
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
 import org.apache.activemq.artemis.api.core.Message;
@@ -58,29 +82,6 @@ import org.apache.activemq.artemis.spi.core.protocol.MessagePersister;
 import org.apache.activemq.artemis.utils.Base64;
 import org.apache.activemq.artemis.utils.XidCodecSupport;
 
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_REF;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADDRESS_SETTING_RECORD;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_MESSAGE;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_MESSAGE_PROTOCOL;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_REF;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.DUPLICATE_ID;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.HEURISTIC_COMPLETION;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ID_COUNTER_RECORD;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COMPLETE;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_INC;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_TRANSACTION;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_BINDING_RECORD;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.QUEUE_STATUS_RECORD;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SECURITY_RECORD;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.UPDATE_DELIVERY_COUNT;
-import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADDRESS_BINDING_RECORD;
-
 
 /**
  * Outputs a String description of the Journals contents.
@@ -217,9 +218,9 @@ public final class DescribeJournal {
                      out.println("####### Counter replace wrongly on queue " + queueIDForCounter + " oldValue=" + subsCounter.getValue() + " newValue=" + encoding.getValue());
                   }
 
-                  subsCounter.loadValue(info.id, encoding.getValue());
+                  subsCounter.loadValue(info.id, encoding.getValue(), encoding.getPersistentSize());
                   subsCounter.processReload();
-                  out.print("#Counter queue " + queueIDForCounter + " value=" + subsCounter.getValue() + ", result=" + subsCounter.getValue());
+                  out.print("#Counter queue " + queueIDForCounter + " value=" + subsCounter.getValue() + " persistentSize=" + subsCounter.getPersistentSize() + ", result=" + subsCounter.getValue());
                   if (subsCounter.getValue() < 0) {
                      out.println(" #NegativeCounter!!!!");
                   } else {
@@ -232,9 +233,9 @@ public final class DescribeJournal {
 
                   PageSubscriptionCounterImpl subsCounter = lookupCounter(counters, queueIDForCounter);
 
-                  subsCounter.loadInc(info.id, encoding.getValue());
+                  subsCounter.loadInc(info.id, encoding.getValue(), encoding.getPersistentSize());
                   subsCounter.processReload();
-                  out.print("#Counter queue " + queueIDForCounter + " value=" + subsCounter.getValue() + " increased by " + encoding.getValue());
+                  out.print("#Counter queue " + queueIDForCounter + " value=" + subsCounter.getValue() + " persistentSize=" + subsCounter.getPersistentSize() + " increased by " + encoding.getValue());
                   if (subsCounter.getValue() < 0) {
                      out.println(" #NegativeCounter!!!!");
                   } else {
@@ -321,7 +322,7 @@ public final class DescribeJournal {
 
             subsCounter = lookupCounter(counters, queueIDForCounter);
 
-            subsCounter.loadValue(info.id, encoding.getValue());
+            subsCounter.loadValue(info.id, encoding.getValue(), encoding.getPersistentSize());
             subsCounter.processReload();
          } else if (info.getUserRecordType() == JournalRecordIds.PAGE_CURSOR_COUNTER_INC) {
             PageCountRecordInc encoding = (PageCountRecordInc) o;
@@ -329,7 +330,7 @@ public final class DescribeJournal {
 
             subsCounter = lookupCounter(counters, queueIDForCounter);
 
-            subsCounter.loadInc(info.id, encoding.getValue());
+            subsCounter.loadInc(info.id, encoding.getValue(), encoding.getPersistentSize());
             subsCounter.processReload();
          }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
index 433031c..dabb039 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
@@ -19,7 +19,6 @@ package org.apache.activemq.artemis.core.persistence.impl.journal;
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import io.netty.buffer.Unpooled;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
@@ -35,6 +34,8 @@ import org.apache.activemq.artemis.utils.DataConstants;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
 import org.jboss.logging.Logger;
 
+import io.netty.buffer.Unpooled;
+
 public final class LargeServerMessageImpl extends CoreMessage implements LargeServerMessage {
 
    // Constants -----------------------------------------------------
@@ -345,6 +346,13 @@ public final class LargeServerMessageImpl extends CoreMessage implements LargeSe
    }
 
    @Override
+   public long getPersistentSize() throws ActiveMQException {
+      long size = super.getPersistentSize();
+      size += getBodyEncoder().getLargeBodySize();
+
+      return size;
+   }
+   @Override
    public String toString() {
       return "LargeServerMessage[messageID=" + messageID + ",durable=" + isDurable() + ",userID=" + getUserID() + ",priority=" + this.getPriority() +
          ", timestamp=" + toDate(getTimestamp()) + ",expiration=" + toDate(getExpiration()) +

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountPendingImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountPendingImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountPendingImpl.java
index 56e8c87..e600d46 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountPendingImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountPendingImpl.java
@@ -23,9 +23,10 @@ import org.apache.activemq.artemis.utils.DataConstants;
 
 public class PageCountPendingImpl implements EncodingSupport, PageCountPending {
 
+
    @Override
    public String toString() {
-      return "PageCountPending [queueID=" + queueID + ", pageID=" + pageID + "]";
+      return "PageCountPendingImpl [queueID=" + queueID + ", pageID=" + pageID + "]";
    }
 
    public PageCountPendingImpl() {
@@ -64,7 +65,7 @@ public class PageCountPendingImpl implements EncodingSupport, PageCountPending {
 
    @Override
    public int getEncodeSize() {
-      return DataConstants.SIZE_LONG * 2;
+      return DataConstants.SIZE_LONG * 3;
    }
 
    @Override

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecord.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecord.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecord.java
index 642feb2..af9e135 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecord.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecord.java
@@ -26,18 +26,21 @@ public class PageCountRecord implements EncodingSupport {
 
    private long value;
 
+   private long persistentSize;
+
    @Override
    public String toString() {
-      return "PageCountRecord [queueID=" + queueID + ", value=" + value + "]";
+      return "PageCountRecord [queueID=" + queueID + ", value=" + value + ", persistentSize=" + persistentSize + "]";
    }
 
    public PageCountRecord() {
 
    }
 
-   public PageCountRecord(long queueID, long value) {
+   public PageCountRecord(long queueID, long value, long persistentSize) {
       this.queueID = queueID;
       this.value = value;
+      this.persistentSize = persistentSize;
    }
 
    public long getQueueID() {
@@ -48,21 +51,30 @@ public class PageCountRecord implements EncodingSupport {
       return value;
    }
 
+   public long getPersistentSize() {
+      return persistentSize;
+   }
+
    @Override
    public int getEncodeSize() {
-      return DataConstants.SIZE_LONG * 2;
+      return DataConstants.SIZE_LONG * 3;
    }
 
    @Override
    public void encode(ActiveMQBuffer buffer) {
       buffer.writeLong(queueID);
       buffer.writeLong(value);
+      buffer.writeLong(persistentSize);
    }
 
    @Override
    public void decode(ActiveMQBuffer buffer) {
       queueID = buffer.readLong();
       value = buffer.readLong();
+
+      if (buffer.readableBytes() > 0) {
+         persistentSize = buffer.readLong();
+      }
    }
 
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecordInc.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecordInc.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecordInc.java
index e427d68..c174155 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecordInc.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/codec/PageCountRecordInc.java
@@ -26,17 +26,20 @@ public class PageCountRecordInc implements EncodingSupport {
 
    private int value;
 
+   private long persistentSize;
+
    @Override
    public String toString() {
-      return "PageCountRecordInc [queueID=" + queueID + ", value=" + value + "]";
+      return "PageCountRecordInc [queueID=" + queueID + ", value=" + value + ", persistentSize=" + persistentSize + "]";
    }
 
    public PageCountRecordInc() {
    }
 
-   public PageCountRecordInc(long queueID, int value) {
+   public PageCountRecordInc(long queueID, int value, long persistentSize) {
       this.queueID = queueID;
       this.value = value;
+      this.persistentSize = persistentSize;
    }
 
    public long getQueueID() {
@@ -47,21 +50,30 @@ public class PageCountRecordInc implements EncodingSupport {
       return value;
    }
 
+   public long getPersistentSize() {
+      return persistentSize;
+   }
+
    @Override
    public int getEncodeSize() {
-      return DataConstants.SIZE_LONG + DataConstants.SIZE_INT;
+      return 2 * DataConstants.SIZE_LONG + DataConstants.SIZE_INT;
    }
 
    @Override
    public void encode(ActiveMQBuffer buffer) {
       buffer.writeLong(queueID);
       buffer.writeInt(value);
+      buffer.writeLong(persistentSize);
    }
 
    @Override
    public void decode(ActiveMQBuffer buffer) {
       queueID = buffer.readLong();
       value = buffer.readInt();
+
+      if (buffer.readableBytes() > 0) {
+         persistentSize = buffer.readLong();
+      }
    }
 
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
index 32f9010..8c5e11c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -467,7 +467,7 @@ public class NullStorageManager implements StorageManager {
    }
 
    @Override
-   public long storePageCounter(final long txID, final long queueID, final long value) throws Exception {
+   public long storePageCounter(final long txID, final long queueID, final long value, final long size) throws Exception {
       return 0;
    }
 
@@ -489,12 +489,12 @@ public class NullStorageManager implements StorageManager {
    }
 
    @Override
-   public long storePageCounterInc(final long txID, final long queueID, final int add) throws Exception {
+   public long storePageCounterInc(final long txID, final long queueID, final int add, final long size) throws Exception {
       return 0;
    }
 
    @Override
-   public long storePageCounterInc(final long queueID, final int add) throws Exception {
+   public long storePageCounterInc(final long queueID, final int add, final long size) throws Exception {
       return 0;
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index 12c722a..73d6953 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -1480,7 +1480,7 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
       public void afterPrepare(final Transaction tx) {
          for (MessageReference ref : refs) {
             if (ref.isAlreadyAcked()) {
-               ref.getQueue().referenceHandled();
+               ref.getQueue().referenceHandled(ref);
                ref.getQueue().incrementMesssagesAdded();
             }
          }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index 7ae1ee4..10c827e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -1911,4 +1911,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
    @LogMessage(level = Logger.Level.ERROR)
    @Message(id = 224088, value = "Timeout ({0} seconds) while handshaking has occurred.", format = Message.Format.MESSAGE_FORMAT)
    void handshakeTimeout(int timeout);
+
+   @LogMessage(level = Logger.Level.WARN)
+   @Message(id = 224089, value = "Failed to calculate persistent size", format = Message.Format.MESSAGE_FORMAT)
+   void errorCalculatePersistentSize(@Cause Throwable e);
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 906ea7e..d9145b1 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
@@ -17,6 +17,7 @@
 package org.apache.activemq.artemis.core.server;
 
 
+import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.core.server.impl.AckReason;
 import org.apache.activemq.artemis.core.server.impl.MessageReferenceImpl;
@@ -99,4 +100,14 @@ public interface MessageReference {
    void setAlreadyAcked();
 
    boolean isAlreadyAcked();
+
+   /**
+    * This is the size of the message when persisted on disk which is used for metrics tracking
+    * Note that even if the message itself is not persisted on disk (ie non-durable) this value is
+    * still used for metrics tracking for the amount of data on a queue
+    *
+    * @return
+    * @throws ActiveMQException
+    */
+   long getPersistentSize() throws ActiveMQException;
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/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 ff4e82b..c355dbf 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
@@ -138,12 +138,42 @@ public interface Queue extends Bindable,CriticalComponent {
 
    long getMessageCount();
 
+   /**
+    * This is the size of the messages in the queue when persisted on disk which is used for metrics tracking
+    * to give an idea of the amount of data on the queue to be consumed
+    *
+    * Note that this includes all messages on the queue, even messages that are non-durable which may only be in memory
+    */
+   long getPersistentSize();
+
+   /**
+    * This is the number of the durable messages in the queue
+    */
+   long getDurableMessageCount();
+
+   /**
+    * This is the persistent size of all the durable messages in the queue
+    */
+   long getDurablePersistentSize();
+
    int getDeliveringCount();
 
-   void referenceHandled();
+   long getDeliveringSize();
+
+   int getDurableDeliveringCount();
+
+   long getDurableDeliveringSize();
+
+   void referenceHandled(MessageReference ref);
 
    int getScheduledCount();
 
+   long getScheduledSize();
+
+   int getDurableScheduledCount();
+
+   long getDurableScheduledSize();
+
    List<MessageReference> getScheduledMessages();
 
    /**
@@ -314,8 +344,6 @@ public interface Queue extends Bindable,CriticalComponent {
     */
    SimpleString getUser();
 
-   void decDelivering(int size);
-
    /** This is to perform a check on the counter again */
    void recheckRefCount(OperationContext context);
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ea70af15/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ScheduledDeliveryHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ScheduledDeliveryHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ScheduledDeliveryHandler.java
index 62fad5e..1dc2eda 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ScheduledDeliveryHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ScheduledDeliveryHandler.java
@@ -27,6 +27,12 @@ public interface ScheduledDeliveryHandler {
 
    int getScheduledCount();
 
+   long getScheduledSize();
+
+   int getDurableScheduledCount();
+
+   long getDurableScheduledSize();
+
    List<MessageReference> getScheduledReferences();
 
    List<MessageReference> cancel(Filter filter) throws ActiveMQException;


Mime
View raw message