[#2275] Switch usage counter to AtomicLong

A three-way benchmark of the counter primitive (TopicRawThroughputBenchmark,
topic-per-producer, direct dispatch, 0 consumers, JDK 25, 11-core Apple M3
Pro, msgs/sec) showed the striped LongAdder buys nothing measurable over a
plain AtomicLong at 1-22 producer threads:

  threads   RW-lock (pre-#2275)   AtomicLong      LongAdder
     1        5.12M +/- 0.10      5.85M +/- 0.11  5.79M +/- 0.10
     4        1.48M +/- 0.12      7.03M +/- 0.94  6.93M +/- 1.04
     8        1.28M +/- 0.11      5.01M +/- 0.41  5.92M +/- 0.27
    22        1.17M +/- 0.09      4.98M +/- 1.53  4.85M +/- 0.59

The 4-6x win over the old ReentrantReadWriteLock comes from removing
blocking (park/unpark convoys), not from striping; at the broker's achieved
message rates (~2 CAS per message on the shared parent) a single AtomicLong
cache line is nowhere near saturation on this class of hardware. The lone
LongAdder edge (8 threads) does not persist at 22 and is within run-to-run
wobble of that regime.

AtomicLong is perferrable over LongAdder for simplicity.

Validation: MemoryUsageConcurrencyTest (4, incl. the concurrent setUsage
drift aggressive test and the 150-round untimed waitForSpace liveness soak)
and MemoryUsageTest (5) pass; ProducerFlowControlTest,
ProducerFlowControlSendFailTest and TopicProducerFlowControlTest (18) pass.

Author: mattrpav

activemq-client/src/main/java/org/apache/activemq/usage/MemoryUsage.java

@@ -17,7 +17,7 @@
 package org.apache.activemq.usage;
 
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Used to keep track of how much of something is being used so that a
@@ -29,11 +29,14 @@
  */
 public class MemoryUsage extends Usage<MemoryUsage> {
 
-    // PROTOTYPE: lock-free usage accounting. The counter is a striped LongAdder so
-    // increase/decrease never take an exclusive lock; the usageLock is only taken when the
-    // rounded percentUsage actually changes (at most ~100/percentUsageMinDelta times per
-    // limit traversal), which preserves listener events and waitForSpace signalling.
-    private final LongAdder usage = new LongAdder();
+    // Lock-free usage accounting: the counter is an AtomicLong so increase/decrease never
+    // take an exclusive lock; the usageLock is only taken when the rounded percentUsage
+    // actually changes (at most ~100/percentUsageMinDelta times per limit traversal), which
+    // preserves listener events and waitForSpace signalling. AtomicLong was chosen over a
+    // striped LongAdder after benchmarking showed equal throughput at 1-22 producer threads
+    // on an 11-core machine, while AtomicLong keeps get() exact, makes setUsage() a plain
+    // atomic set, and avoids per-instance cell inflation.
+    private final AtomicLong usage = new AtomicLong();
 
     public MemoryUsage() {
         this(null, null);
@@ -166,7 +169,7 @@ public void increaseUsage(long value) {
         // recomputes the percent from a sum that includes all completed updates, so a lasting
         // 100% -> <100% transition always reaches the locked setPercentUsage() path, which
         // signals waitForSpaceCondition. Breaking this ordering can strand waiters forever.
-        usage.add(value);
+        usage.addAndGet(value);
         maybeUpdatePercent();
 
         if (parent != null) {
@@ -186,7 +189,7 @@ public void decreaseUsage(long value) {
 
         // INVARIANT: add() must be followed unconditionally by maybeUpdatePercent()
         // (see increaseUsage for the full liveness rationale).
-        usage.add(-value);
+        usage.addAndGet(-value);
         maybeUpdatePercent();
 
         if (parent != null) {
@@ -214,24 +217,21 @@ private void maybeUpdatePercent() {
 
     @Override
     protected long retrieveUsage() {
-        return usage.sum();
+        return usage.get();
     }
 
     @Override
     public long getUsage() {
-        return usage.sum();
+        return usage.get();
     }
 
     /**
-     * Sets the usage to the given value. Implemented as a delta adjustment because
-     * LongAdder.reset() is only safe when there are no concurrent updates - a racing
-     * increase/decrease could be lost outright. With a delta add, a concurrent update is
-     * always preserved, equivalent to it linearizing after the set. Note: as with the
-     * historical field assignment, this does not propagate an adjustment to the parent
-     * usage.
+     * Sets the usage to the given value as a single atomic store; a concurrent
+     * increase/decrease linearizes cleanly before or after it. Note: as with the historical
+     * field assignment, this does not propagate an adjustment to the parent usage.
      */
     public void setUsage(long value) {
-        this.usage.add(value - this.usage.sum());
+        this.usage.set(value);
         maybeUpdatePercent();
     }