Fix

Author: wu-sheng

docs/en/changes/changes.md

@@ -23,13 +23,13 @@
 
   | Queue | Old threads | Old channels | Old buffer slots | New threads | New partitions | New buffer slots | New policy |
   |-------|-------------|--------------|------------------|-------------|----------------|------------------|------------|
-  | L1 Aggregation (OAL) | 24 | ~1,240 | ~12.4M | 8 (unified) | ~460 adaptive | ~9.2M | `cpuCores(1.0)` |
+  | L1 Aggregation (OAL) | 24 | ~1,240 | ~12.4M | 8 (unified) | ~330 adaptive | ~6.6M | `cpuCores(1.0)` |
   | L1 Aggregation (MAL) | 2 | ~100 | ~100K | (unified above) | | | |
-  | L2 Persistence (OAL) | 2 | ~620 | ~1.24M | 3 (unified) | ~460 adaptive | ~920K | `cpuCoresWithBase(1, 0.25)` |
+  | L2 Persistence (OAL) | 2 | ~620 | ~1.24M | 3 (unified) | ~330 adaptive | ~660K | `cpuCoresWithBase(1, 0.25)` |
   | L2 Persistence (MAL) | 1 | ~100 | ~100K | (unified above) | | | |
   | TopN Persistence | 4 | 4 | 4K | 1 | 4 adaptive | 4K | `fixed(1)` |
   | Exporters (gRPC/Kafka) | 3 | 6 | 120K | 3 (1 per exporter) | — | 60K | `fixed(1)` each |
-  | **Total** | **36** | **~2,070** | **~13.9M** | **15** | **~924** | **~10.2M** | |
+  | **Total** | **36** | **~2,070** | **~13.9M** | **15** | **~664** | **~7.3M** | |
 
 * Remove `library-datacarrier-queue` module. All usages have been replaced by `library-batch-queue`.
 * Enable throughput-weighted drain rebalancing for L1 aggregation and L2 persistence queues (10s interval).
@@ -57,8 +57,8 @@
 
   | Catalog | Thread Name | Count | Policy | Partitions |
   |---------|-------------|-------|--------|------------|
-  | Data Pipeline | `BatchQueue-METRICS_L1_AGGREGATION-N` | 8 | `cpuCores(1.0)` | ~460 adaptive |
-  | Data Pipeline | `BatchQueue-METRICS_L2_PERSISTENCE-N` | 3 | `cpuCoresWithBase(1, 0.25)` | ~460 adaptive |
+  | Data Pipeline | `BatchQueue-METRICS_L1_AGGREGATION-N` | 8 | `cpuCores(1.0)` | ~330 adaptive |
+  | Data Pipeline | `BatchQueue-METRICS_L2_PERSISTENCE-N` | 3 | `cpuCoresWithBase(1, 0.25)` | ~330 adaptive |
   | Data Pipeline | `BatchQueue-TOPN_PERSISTENCE-N` | 1 | `fixed(1)` | ~4 adaptive |
   | Data Pipeline | `BatchQueue-GRPC_REMOTE_{host}_{port}-N` | 1 per peer | `fixed(1)` | `fixed(1)` |
   | Data Pipeline | `BatchQueue-EXPORTER_GRPC_METRICS-N` | 1 | `fixed(1)` | `fixed(1)` |

oap-server/server-library/library-batch-queue/CLAUDE.md

@@ -146,7 +146,7 @@ Silently ignored for single-thread queues (nothing to rebalance).
 ### L1 Metrics Aggregation (`MetricsAggregateWorker`)
 ```
 threads:    cpuCores(1.0)        -- 8 threads on 8-core
-partitions: adaptive()           -- grows with metric types (~460 for typical OAL+MAL)
+partitions: adaptive()           -- grows with metric types (~330 for typical OAL+MAL on 8 threads)
 bufferSize: 20,000 per partition
 strategy:   IF_POSSIBLE
 idleMs:     1..50

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueue.java

@@ -213,6 +213,11 @@ public class BatchQueue<T> {
     private final int[] consecutiveIdleCycles;
 
     /** Set to false on {@link #shutdown()} to stop drain loops and reject new data. */
+    /**
+     * Whether the queue is currently accepting produces and running drain loops.
+     *
+     * @return true if the queue is running
+     */
     @Getter
     private volatile boolean running;
 
@@ -424,6 +429,9 @@ private void scheduleDrain(final int taskIndex) {
      * count and grows the partition array if needed. For non-adaptive policies the
      * resolved count never changes, so this is a no-op beyond the registration.
      * Drain loop threads pick up new partitions on their next cycle via volatile reads.
+     *
+     * @param type the class of items to route to this handler
+     * @param handler the consumer that processes batches of the given type
      */
     @SuppressWarnings("unchecked")
     public void addHandler(final Class<? extends T> type, final HandlerConsumer<T> handler) {
@@ -522,6 +530,7 @@ private void enableRebalancing(final long intervalMs) {
      *   <li>IF_POSSIBLE — returns false immediately if the partition is full (data dropped)</li>
      * </ul>
      *
+     * @param data the item to enqueue
      * @return true if data was accepted, false if dropped or queue is stopped
      */
     public boolean produce(final T data) {
@@ -844,6 +853,8 @@ int getTaskCount() {
 
     /**
      * Take a point-in-time snapshot of queue usage across all partitions.
+     *
+     * @return a stats snapshot containing per-partition usage and capacity
      */
     public BatchQueueStats stats() {
         final ArrayBlockingQueue<T>[] currentPartitions = this.partitions;

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfig.java

@@ -116,6 +116,13 @@ void validate() {
      * Builder customizations: convenience methods for setting paired fields together.
      */
     public static class BatchQueueConfigBuilder<T> {
+        /**
+         * Configure the queue to use a shared scheduler instead of a dedicated one.
+         *
+         * @param name the shared scheduler name (queues with the same name share a pool)
+         * @param threads the thread policy for the shared scheduler
+         * @return this builder
+         */
         public BatchQueueConfigBuilder<T> sharedScheduler(final String name, final ThreadPolicy threads) {
             this.sharedSchedulerName = name;
             this.sharedSchedulerThreads = threads;
@@ -127,6 +134,7 @@ public BatchQueueConfigBuilder<T> sharedScheduler(final String name, final Threa
          *
          * @param balancer rebalancing strategy (e.g. {@link DrainBalancer#throughputWeighted()})
          * @param intervalMs rebalance interval in milliseconds
+         * @return this builder
          */
         public BatchQueueConfigBuilder<T> balancer(final DrainBalancer balancer, final long intervalMs) {
             this.balancer = balancer;

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueStats.java

@@ -48,13 +48,17 @@ public class BatchQueueStats {
 
     /**
      * Total capacity across all partitions: {@code partitionCount * bufferSize}.
+     *
+     * @return total capacity in item slots
      */
     public long totalCapacity() {
         return (long) partitionCount * bufferSize;
     }
 
     /**
      * Total number of items currently queued across all partitions.
+     *
+     * @return sum of items across all partitions
      */
     public int totalUsed() {
         int sum = 0;
@@ -66,6 +70,8 @@ public int totalUsed() {
 
     /**
      * Overall queue usage as a percentage (0.0–100.0).
+     *
+     * @return usage percentage across all partitions
      */
     public double totalUsedPercentage() {
         final long capacity = totalCapacity();
@@ -77,13 +83,19 @@ public double totalUsedPercentage() {
 
     /**
      * Number of items currently queued in the given partition.
+     *
+     * @param index the partition index
+     * @return number of items in the partition
      */
     public int partitionUsed(final int index) {
         return partitionUsed[index];
     }
 
     /**
      * Usage of the given partition as a percentage (0.0–100.0).
+     *
+     * @param index the partition index
+     * @return usage percentage for the partition
      */
     public double partitionUsedPercentage(final int index) {
         if (bufferSize == 0) {
@@ -94,7 +106,10 @@ public double partitionUsedPercentage(final int index) {
 
     /**
      * Return the top {@code n} most-loaded partitions, sorted by usage descending.
-     * If {@code n >= partitionCount}, all partitions are returned.
+     * If {@code n >= partitionCount}, all partitions are returned.
+     *
+     * @param n the maximum number of partitions to return
+     * @return list of partition usage snapshots sorted by usage descending
      */
     public List<PartitionUsage> topN(final int n) {
         final Integer[] indices = new Integer[partitionCount];

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/DrainBalancer.java

@@ -49,6 +49,8 @@ public interface DrainBalancer {
      * Sorts partitions by throughput descending, assigns each to the least-loaded thread.
      * Zero-throughput partitions keep their current owner to avoid unnecessary moves.
      * Skips rebalancing when load is already balanced (max/min ratio < 1.15).
+     *
+     * @return a throughput-weighted drain balancer
      */
     static DrainBalancer throughputWeighted() {
         return new ThroughputWeightedBalancer();

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/HandlerConsumer.java

@@ -27,6 +27,8 @@
 public interface HandlerConsumer<T> {
     /**
      * Process a batch of data belonging to this handler's type.
+     *
+     * @param data the batch of items to process
      */
     void consume(List<T> data);
 

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicy.java

@@ -53,6 +53,8 @@ private PartitionPolicy(final int fixedCount, final int multiplier,
     /**
      * Fixed number of partitions.
      *
+     * @param count the exact number of partitions
+     * @return a PartitionPolicy with a fixed partition count
      * @throws IllegalArgumentException if count < 1
      */
     public static PartitionPolicy fixed(final int count) {
@@ -65,6 +67,8 @@ public static PartitionPolicy fixed(final int count) {
     /**
      * Partitions = multiplier * resolved thread count.
      *
+     * @param multiplier factor applied to thread count
+     * @return a PartitionPolicy that scales with thread count
      * @throws IllegalArgumentException if multiplier < 1
      */
     public static PartitionPolicy threadMultiply(final int multiplier) {
@@ -95,6 +99,8 @@ public static PartitionPolicy threadMultiply(final int multiplier) {
      *  1000 handlers → 600 partitions  (200 + 800/2)
      *  2000 handlers → 1100 partitions (200 + 1800/2)
      * </pre>
+     *
+     * @return an adaptive PartitionPolicy with default threshold multiplier
      */
     public static PartitionPolicy adaptive() {
         return new PartitionPolicy(0, DEFAULT_ADAPTIVE_MULTIPLIER, true);
@@ -108,6 +114,7 @@ public static PartitionPolicy adaptive() {
      * at 1:2 ratio: {@code threshold + (handlerCount - threshold) / 2}.
      *
      * @param multiplier threshold per thread (default 25)
+     * @return a PartitionPolicy that grows with handler registrations
      * @throws IllegalArgumentException if multiplier &lt; 1
      */
     public static PartitionPolicy adaptive(final int multiplier) {
@@ -128,6 +135,10 @@ public static PartitionPolicy adaptive(final int multiplier) {
      *       &lt;= threshold, returns handlerCount (1:1). If above, returns
      *       threshold + (handlerCount - threshold) / 2.</li>
      * </ul>
+     *
+     * @param resolvedThreadCount the resolved number of drain threads
+     * @param handlerCount the current number of registered type handlers
+     * @return the resolved partition count, always &gt;= 1
      */
     public int resolve(final int resolvedThreadCount, final int handlerCount) {
         if (fixedCount > 0) {

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionSelector.java

@@ -47,6 +47,9 @@ public interface PartitionSelector<T> {
      * Default selector: routes by {@code data.getClass().hashCode()}.
      * Same type always hits the same partition, so each consumer thread
      * drains pre-grouped batches — dispatch grouping is effectively a no-op.
+     *
+     * @param <T> the queue element type
+     * @return a selector that partitions by item class hash
      */
     static <T> PartitionSelector<T> typeHash() {
         return (data, count) -> (data.getClass().hashCode() & 0x7FFFFFFF) % count;

oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicy.java

@@ -44,7 +44,9 @@ private ThreadPolicy(final int fixedCount, final int base, final double cpuMulti
     /**
      * Fixed number of threads. Count must be >= 1.
      *
-     * @throws IllegalArgumentException if count < 1
+     * @param count the exact number of threads
+     * @return a ThreadPolicy with a fixed thread count
+     * @throws IllegalArgumentException if count &lt; 1
      */
     public static ThreadPolicy fixed(final int count) {
         if (count < 1) {
@@ -57,7 +59,9 @@ public static ThreadPolicy fixed(final int count) {
      * Threads = multiplier * available CPU cores, rounded, min 1.
      * Multiplier must be &gt; 0.
      *
-     * @throws IllegalArgumentException if multiplier <= 0
+     * @param multiplier factor applied to available CPU core count
+     * @return a ThreadPolicy proportional to CPU cores
+     * @throws IllegalArgumentException if multiplier &lt;= 0
      */
     public static ThreadPolicy cpuCores(final double multiplier) {
         if (multiplier <= 0) {
@@ -72,7 +76,10 @@ public static ThreadPolicy cpuCores(final double multiplier) {
      *
      * Example: cpuCoresWithBase(2, 0.25) on 8-core = 2 + 2 = 4, on 16-core = 2 + 4 = 6, on 24-core = 2 + 6 = 8.
      *
-     * @throws IllegalArgumentException if base < 0 or multiplier <= 0
+     * @param base fixed base thread count added to the CPU-proportional portion
+     * @param multiplier factor applied to available CPU core count
+     * @return a ThreadPolicy that combines a fixed base with a CPU-proportional count
+     * @throws IllegalArgumentException if base &lt; 0 or multiplier &lt;= 0
      */
     public static ThreadPolicy cpuCoresWithBase(final int base, final double multiplier) {
         if (base < 0) {
@@ -86,6 +93,8 @@ public static ThreadPolicy cpuCoresWithBase(final int base, final double multipl
 
     /**
      * Resolve the actual thread count. Always returns &gt;= 1.
+     *
+     * @return the resolved thread count, at least 1
      */
     public int resolve() {
         if (fixedCount > 0) {