[#2160] Fix dispatch starvation when network consumers have full prefetch

Queue.doPageInForDispatch() used pagedInPendingSize < maxPageSize as the
sole gate for paging in messages from the store. When low-priority network
consumers had saturated prefetch (not acking), their undispatched messages
accumulated in dispatchPendingList, counting against maxPageSize and
blocking page-in for all consumers — including higher-priority app
consumers with available prefetch capacity.

Two changes:
1. When pagedInPendingSize >= maxPageSize, cap toPageIn to what consumers
   can actually accept (getConsumerMessageCountBeforeFull)
2. Allow page-in when any consumer has available prefetch capacity, even
   if pagedInPendingSize >= maxPageSize

Author: mattrpav

activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -2113,8 +2113,13 @@ private PendingList doPageInForDispatch(boolean force, boolean processExpired, i
             pagedInPendingDispatchLock.readLock().unlock();
         }
         if (isLazyDispatch() && !force) {
-            // Only page in the minimum number of messages which can be
-            // dispatched immediately.
+            toPageIn = Math.min(toPageIn, getConsumerMessageCountBeforeFull());
+        }
+        if (pagedInPendingSize >= maxPageSize && !force) {
+            // When the pending dispatch list is full, only page in what
+            // consumers can actually accept. This prevents unbounded growth
+            // while still allowing consumers with available prefetch capacity
+            // to receive messages even when other consumers are full.
             toPageIn = Math.min(toPageIn, getConsumerMessageCountBeforeFull());
         }
 
@@ -2132,7 +2137,8 @@ private PendingList doPageInForDispatch(boolean force, boolean processExpired, i
                     maxPageSize);
         }
 
-        if (toPageIn > 0 && (force || (haveRealConsumer() && pagedInPendingSize < maxPageSize))) {
+        if (toPageIn > 0 && (force || (haveRealConsumer() && (pagedInPendingSize < maxPageSize
+                || getConsumerMessageCountBeforeFull() > 0)))) {
             int count = 0;
             result = new ArrayList<QueueMessageReference>(toPageIn);
             messagesLock.writeLock().lock();