You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/02/13 02:26:29 UTC
(skywalking) branch library-batch-queue created (now 19184e2996)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch library-batch-queue
in repository https://gitbox.apache.org/repos/asf/skywalking.git
at 19184e2996 Add library-batch-queue module with partitioned, self-draining queue
This branch includes the following new commits:
new 19184e2996 Add library-batch-queue module with partitioned, self-draining queue
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking) 01/01: Add library-batch-queue module with partitioned, self-draining queue
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch library-batch-queue
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 19184e29964132b64a039d02a6964c7ba843c79a
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Feb 13 10:25:16 2026 +0800
Add library-batch-queue module with partitioned, self-draining queue
Introduce BatchQueue as a new queue implementation for metrics aggregation
with type-based dispatch, configurable partition selection, and adaptive
backoff. Includes comprehensive unit tests, throughput benchmarks, and
DataCarrier comparison benchmarks. Also migrates existing microbench
benchmarks out of the JMH-based microbench module into standard JUnit tests.
---
.../benchmark/EBPFProfilingAnalyzerBenchmark.java | 177 +++
.../EndpointGrouping4OpenapiBenchmark.java | 81 ++
.../core/benchmark/RegexVSQuickMatchBenchmark.java | 116 ++
.../server-library/library-batch-queue/DESIGN.md | 1204 ++++++++++++++++++++
.../{ => library-batch-queue}/pom.xml | 37 +-
.../oap/server/library/batchqueue/BatchQueue.java | 467 ++++++++
.../library/batchqueue/BatchQueueConfig.java | 110 ++
.../library/batchqueue/BatchQueueManager.java | 179 +++
.../server/library/batchqueue/BufferStrategy.java | 31 +
.../server/library/batchqueue/HandlerConsumer.java | 39 +
.../server/library/batchqueue/PartitionPolicy.java | 84 ++
.../library/batchqueue/PartitionSelector.java | 54 +
.../library/batchqueue/QueueErrorHandler.java | 26 +
.../server/library/batchqueue/ThreadPolicy.java | 83 ++
.../library/batchqueue/BatchQueueBenchmark.java | 400 +++++++
.../library/batchqueue/BatchQueueConfigTest.java | 110 ++
.../library/batchqueue/BatchQueueManagerTest.java | 162 +++
.../server/library/batchqueue/BatchQueueTest.java | 573 ++++++++++
.../library/batchqueue/BenchmarkMetricTypes.java | 649 +++++++++++
.../library/batchqueue/PartitionPolicyTest.java | 73 ++
.../library/batchqueue/ThreadPolicyTest.java | 78 ++
.../src/test/resources/log4j2-test.xml | 32 +
.../library/datacarrier/DataCarrierBenchmark.java | 417 +++++++
.../util/benchmark/StringFormatGroupBenchmark.java | 79 ++
oap-server/server-library/pom.xml | 1 +
25 files changed, 5246 insertions(+), 16 deletions(-)
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EBPFProfilingAnalyzerBenchmark.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EBPFProfilingAnalyzerBenchmark.java
new file mode 100644
index 0000000000..4c14aa19c4
--- /dev/null
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EBPFProfilingAnalyzerBenchmark.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.benchmark;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.analyze.EBPFProfilingAnalyzer;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.analyze.EBPFProfilingStack;
+import org.apache.skywalking.oap.server.core.profiling.ebpf.storage.EBPFProfilingStackType;
+import org.apache.skywalking.oap.server.core.query.type.EBPFProfilingAnalyzation;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+public class EBPFProfilingAnalyzerBenchmark {
+
+ private static final Random RANDOM = new Random(System.currentTimeMillis());
+ private static final int SYMBOL_LENGTH = 10;
+ private static final char[] SYMBOL_TABLE = "abcdefgABCDEFG1234567890_[]<>.".toCharArray();
+ private static final EBPFProfilingStackType[] STACK_TYPES = new EBPFProfilingStackType[] {
+ EBPFProfilingStackType.KERNEL_SPACE, EBPFProfilingStackType.USER_SPACE};
+
+ private static final int WARMUP_ITERATIONS = 3;
+ private static final int MEASURE_ITERATIONS = 5;
+
+ private static List<EBPFProfilingStack> generateStacks(final int totalStackCount,
+ final int perStackMinDepth,
+ final int perStackMaxDepth,
+ final double[] stackSymbolDuplicateRate,
+ final double stackDuplicateRate) {
+ final int uniqStackCount = (int) (100 / stackDuplicateRate);
+ final List<EBPFProfilingStack> stacks = new ArrayList<>(totalStackCount);
+ final StackSymbolGenerator stackSymbolGenerator =
+ new StackSymbolGenerator(stackSymbolDuplicateRate, perStackMaxDepth);
+ for (int inx = 0; inx < uniqStackCount; inx++) {
+ final EBPFProfilingStack s = generateStack(perStackMinDepth, perStackMaxDepth, stackSymbolGenerator);
+ stacks.add(s);
+ }
+ for (int inx = uniqStackCount; inx < totalStackCount; inx++) {
+ stacks.add(stacks.get(RANDOM.nextInt(uniqStackCount)));
+ }
+ return stacks;
+ }
+
+ private static class StackSymbolGenerator {
+ private final Map<Integer, Integer> stackDepthSymbolCount;
+ private final Map<Integer, List<String>> existingSymbolMap;
+
+ StackSymbolGenerator(final double[] stackSymbolDuplicateRate, final int maxDepth) {
+ this.stackDepthSymbolCount = new HashMap<>();
+ for (int depth = 0; depth < maxDepth; depth++) {
+ double rate = stackSymbolDuplicateRate[stackSymbolDuplicateRate.length - 1];
+ if (stackSymbolDuplicateRate.length > depth) {
+ rate = stackSymbolDuplicateRate[depth];
+ }
+ final int uniqStackCount = (int) (100 / rate);
+ stackDepthSymbolCount.put(depth, uniqStackCount);
+ }
+ this.existingSymbolMap = new HashMap<>();
+ }
+
+ String generate(final int depth) {
+ List<String> symbols = existingSymbolMap.get(depth);
+ if (symbols == null) {
+ symbols = new ArrayList<>();
+ existingSymbolMap.put(depth, symbols);
+ }
+ final Integer needCount = this.stackDepthSymbolCount.get(depth);
+ if (symbols.size() < needCount) {
+ final StringBuilder sb = new StringBuilder(SYMBOL_LENGTH);
+ for (int j = 0; j < SYMBOL_LENGTH; j++) {
+ sb.append(SYMBOL_TABLE[RANDOM.nextInt(SYMBOL_TABLE.length)]);
+ }
+ final String s = sb.toString();
+ symbols.add(s);
+ return s;
+ } else {
+ return symbols.get(RANDOM.nextInt(symbols.size()));
+ }
+ }
+ }
+
+ private static EBPFProfilingStack generateStack(final int stackMinDepth, final int stackMaxDepth,
+ final StackSymbolGenerator stackSymbolGenerator) {
+ final int stackDepth = stackMinDepth + RANDOM.nextInt(stackMaxDepth - stackMinDepth);
+ final List<EBPFProfilingStack.Symbol> symbols = new ArrayList<>(stackDepth);
+ for (int i = 0; i < stackDepth; i++) {
+ final EBPFProfilingStack.Symbol symbol = new EBPFProfilingStack.Symbol(
+ stackSymbolGenerator.generate(i), buildStackType(i, stackDepth));
+ symbols.add(symbol);
+ }
+ final EBPFProfilingStack stack = new EBPFProfilingStack();
+ stack.setDumpCount(RANDOM.nextInt(100));
+ stack.setSymbols(symbols);
+ return stack;
+ }
+
+ private static EBPFProfilingStackType buildStackType(final int currentDepth, final int totalDepth) {
+ final int partition = totalDepth / STACK_TYPES.length;
+ for (int i = 1; i <= STACK_TYPES.length; i++) {
+ if (currentDepth < i * partition) {
+ return STACK_TYPES[i - 1];
+ }
+ }
+ return STACK_TYPES[STACK_TYPES.length - 1];
+ }
+
+ private static int calculateStackCount(final int stackReportPeriodSecond,
+ final int totalTimeMinute,
+ final int combineInstanceCount) {
+ return (int) (TimeUnit.MINUTES.toSeconds(totalTimeMinute)
+ / stackReportPeriodSecond * combineInstanceCount);
+ }
+
+ private static void analyze(final List<EBPFProfilingStack> stacks) {
+ new EBPFProfilingAnalyzer(null, 100, 5)
+ .generateTrees(new EBPFProfilingAnalyzation(), stacks.parallelStream());
+ }
+
+ @Test
+ public void analyzeLowDataSource() {
+ final List<EBPFProfilingStack> stacks = generateStacks(
+ calculateStackCount(5, 60, 10), 15, 30,
+ new double[] {100, 50, 45, 40, 35, 30, 15, 10, 5}, 5);
+ runBenchmark("low", stacks);
+ }
+
+ @Test
+ public void analyzeMedianDataSource() {
+ final List<EBPFProfilingStack> stacks = generateStacks(
+ calculateStackCount(5, 100, 200), 15, 30,
+ new double[] {50, 40, 35, 30, 20, 10, 7, 5, 2}, 3);
+ runBenchmark("median", stacks);
+ }
+
+ @Test
+ public void analyzeHighDataSource() {
+ final List<EBPFProfilingStack> stacks = generateStacks(
+ calculateStackCount(5, 2 * 60, 2000), 15, 40,
+ new double[] {30, 27, 25, 20, 17, 15, 10, 7, 5, 2, 1}, 1);
+ runBenchmark("high", stacks);
+ }
+
+ private void runBenchmark(final String label, final List<EBPFProfilingStack> stacks) {
+ for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+ analyze(stacks);
+ }
+ final long start = System.nanoTime();
+ for (int i = 0; i < MEASURE_ITERATIONS; i++) {
+ analyze(stacks);
+ }
+ final long elapsed = System.nanoTime() - start;
+ log.info("{}: {} iterations, {} ms/op",
+ label, MEASURE_ITERATIONS,
+ elapsed / MEASURE_ITERATIONS / 1_000_000);
+ }
+}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EndpointGrouping4OpenapiBenchmark.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EndpointGrouping4OpenapiBenchmark.java
new file mode 100644
index 0000000000..d3d0dae010
--- /dev/null
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/EndpointGrouping4OpenapiBenchmark.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.benchmark;
+
+import java.util.Collections;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.config.group.openapi.EndpointGroupingRule4Openapi;
+import org.apache.skywalking.oap.server.core.config.group.openapi.EndpointGroupingRuleReader4Openapi;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+public class EndpointGrouping4OpenapiBenchmark {
+
+ private static final String APT_TEST_DATA = " /products1/{id}/%d:\n" + " get:\n" + " post:\n"
+ + " /products2/{id}/%d:\n" + " get:\n" + " post:\n"
+ + " /products3/{id}/%d:\n" + " get:\n";
+
+ private static final int WARMUP_ITERATIONS = 5000;
+ private static final int MEASURE_ITERATIONS = 100_000;
+
+ private static Map<String, String> createTestFile(final int size) {
+ final StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("paths:\n");
+ for (int i = 0; i <= size; i++) {
+ stringBuilder.append(String.format(APT_TEST_DATA, i, i, i));
+ }
+ return Collections.singletonMap("whatever", stringBuilder.toString());
+ }
+
+ @Test
+ public void formatEndpointNameMatchedPaths20() {
+ final EndpointGroupingRule4Openapi rule =
+ new EndpointGroupingRuleReader4Openapi(createTestFile(3)).read();
+ runBenchmark("paths20", rule);
+ }
+
+ @Test
+ public void formatEndpointNameMatchedPaths50() {
+ final EndpointGroupingRule4Openapi rule =
+ new EndpointGroupingRuleReader4Openapi(createTestFile(9)).read();
+ runBenchmark("paths50", rule);
+ }
+
+ @Test
+ public void formatEndpointNameMatchedPaths200() {
+ final EndpointGroupingRule4Openapi rule =
+ new EndpointGroupingRuleReader4Openapi(createTestFile(39)).read();
+ runBenchmark("paths200", rule);
+ }
+
+ private void runBenchmark(final String label, final EndpointGroupingRule4Openapi rule) {
+ for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+ rule.format("serviceA", "GET:/products1/123");
+ }
+ final long start = System.nanoTime();
+ for (int i = 0; i < MEASURE_ITERATIONS; i++) {
+ rule.format("serviceA", "GET:/products1/123");
+ }
+ final long elapsed = System.nanoTime() - start;
+ log.info("{}: {} ops, {} ns/op, {} ops/s",
+ label, MEASURE_ITERATIONS, elapsed / MEASURE_ITERATIONS,
+ MEASURE_ITERATIONS * 1_000_000_000L / elapsed);
+ }
+}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/RegexVSQuickMatchBenchmark.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/RegexVSQuickMatchBenchmark.java
new file mode 100644
index 0000000000..0ef96aa5e3
--- /dev/null
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/benchmark/RegexVSQuickMatchBenchmark.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.benchmark;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.config.group.EndpointGroupingRule;
+import org.apache.skywalking.oap.server.core.config.group.uri.quickmatch.QuickUriGroupingRule;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+public class RegexVSQuickMatchBenchmark {
+
+ private static final int WARMUP_ITERATIONS = 5000;
+ private static final int MEASURE_ITERATIONS = 100_000;
+
+ private final EndpointGroupingRule regexRule = new EndpointGroupingRule();
+ private final QuickUriGroupingRule quickRule = new QuickUriGroupingRule();
+
+ {
+ regexRule.addRule("service1", "/products/{var}", "/products/.+");
+ regexRule.addRule("service1", "/products/{var}/detail", "/products/.+/detail");
+ regexRule.addRule("service1", "/sales/{var}/1", "/sales/.+/1");
+ regexRule.addRule("service1", "/sales/{var}/2", "/sales/.+/2");
+ regexRule.addRule("service1", "/sales/{var}/3", "/sales/.+/3");
+ regexRule.addRule("service1", "/sales/{var}/4", "/sales/.+/4");
+ regexRule.addRule("service1", "/sales/{var}/5", "/sales/.+/5");
+ regexRule.addRule("service1", "/sales/{var}/6", "/sales/.+/6");
+ regexRule.addRule("service1", "/sales/{var}/7", "/sales/.+/7");
+ regexRule.addRule("service1", "/sales/{var}/8", "/sales/.+/8");
+ regexRule.addRule("service1", "/sales/{var}/9", "/sales/.+/9");
+ regexRule.addRule("service1", "/sales/{var}/10", "/sales/.+/10");
+ regexRule.addRule("service1", "/sales/{var}/11", "/sales/.+/11");
+ regexRule.addRule("service1", "/employees/{var}/profile", "/employees/.+/profile");
+
+ quickRule.addRule("service1", "/products/{var}");
+ quickRule.addRule("service1", "/products/{var}/detail");
+ quickRule.addRule("service1", "/sales/{var}/1");
+ quickRule.addRule("service1", "/sales/{var}/2");
+ quickRule.addRule("service1", "/sales/{var}/3");
+ quickRule.addRule("service1", "/sales/{var}/4");
+ quickRule.addRule("service1", "/sales/{var}/5");
+ quickRule.addRule("service1", "/sales/{var}/6");
+ quickRule.addRule("service1", "/sales/{var}/7");
+ quickRule.addRule("service1", "/sales/{var}/8");
+ quickRule.addRule("service1", "/sales/{var}/9");
+ quickRule.addRule("service1", "/sales/{var}/10");
+ quickRule.addRule("service1", "/sales/{var}/11");
+ quickRule.addRule("service1", "/employees/{var}/profile");
+ }
+
+ @Test
+ public void matchFirstRegex() {
+ runBenchmark("matchFirstRegex",
+ () -> regexRule.format("service1", "/products/123"));
+ }
+
+ @Test
+ public void matchFirstQuickUriGrouping() {
+ runBenchmark("matchFirstQuick",
+ () -> quickRule.format("service1", "/products/123"));
+ }
+
+ @Test
+ public void matchFourthRegex() {
+ runBenchmark("matchFourthRegex",
+ () -> regexRule.format("service1", "/sales/123/2"));
+ }
+
+ @Test
+ public void matchFourthQuickUriGrouping() {
+ runBenchmark("matchFourthQuick",
+ () -> quickRule.format("service1", "/sales/123/2"));
+ }
+
+ @Test
+ public void notMatchRegex() {
+ runBenchmark("notMatchRegex",
+ () -> regexRule.format("service1", "/employees/123"));
+ }
+
+ @Test
+ public void notMatchQuickUriGrouping() {
+ runBenchmark("notMatchQuick",
+ () -> quickRule.format("service1", "/employees/123"));
+ }
+
+ private void runBenchmark(final String label, final Runnable action) {
+ for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+ action.run();
+ }
+ final long start = System.nanoTime();
+ for (int i = 0; i < MEASURE_ITERATIONS; i++) {
+ action.run();
+ }
+ final long elapsed = System.nanoTime() - start;
+ log.info("{}: {} ops, {} ns/op, {} ops/s",
+ label, MEASURE_ITERATIONS, elapsed / MEASURE_ITERATIONS,
+ MEASURE_ITERATIONS * 1_000_000_000L / elapsed);
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/DESIGN.md b/oap-server/server-library/library-batch-queue/DESIGN.md
new file mode 100644
index 0000000000..4094f56f29
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/DESIGN.md
@@ -0,0 +1,1204 @@
+# library-batch-queue Design Proposal
+
+## Goal
+
+Replace `library-datacarrier-queue` with a unified, simpler batch queue library that reduces thread
+usage while preserving all required capabilities.
+
+## Problem Statement
+
+DataCarrier has two consumption modes with separate code paths:
+
+1. **Simple mode** (`ConsumeDriver`): Each DataCarrier gets dedicated consumer thread(s).
+ Used by TopNWorker, GRPCRemoteClient, JDBCBatchDAO, exporters.
+2. **Pool mode** (`BulkConsumePool`): Multiple DataCarriers share a thread pool.
+ Used by MetricsAggregateWorker (L1) and MetricsPersistentMinWorker (L2).
+
+On an 8-core production machine, this creates **47+ DataCarrier consumer threads**:
+
+| Source | Threads | Mode |
+|--------------------------------|---------|--------|
+| L1 OAL aggregation pool | 24 | Pool |
+| L2 OAL persistent pool | 2 | Pool |
+| L1 MAL aggregation pool | 2 | Pool |
+| L2 MAL persistent pool | 1 | Pool |
+| TopNWorker (per type) | 5-10 | Simple |
+| GRPCRemoteClient (per peer) | 2-4 | Simple |
+| JDBCBatchDAO | 2-4 | Simple |
+| Exporters (gRPC/Kafka) | 0-3 | Simple |
+| **Total** | **~38-48** | |
+
+Key issues:
+- Simple mode wastes threads: each queue gets a dedicated thread even though most are idle
+ (sleeping in 200ms polling loop).
+- Pool mode creates one DataCarrier (with Channels) per metric type. With 100+ metric types,
+ pool threads must iterate through all assigned channels even when most are empty — wasted CPU.
+- Two completely separate code paths for the same produce-consume pattern.
+- Pool mode has static assignment — no rebalancing after initial allocation.
+
+## Design
+
+### Architecture Overview
+
+```
+ BatchQueueManager (global singleton registry + lazy shared schedulers)
+ │
+ │ Shared schedulers (created lazily on first queue reference):
+ │ ├── "IO_POOL" ──> ScheduledExecutorService (cpuCores(0.5) → 4 threads on 8-core)
+ │ │ Created when first queue calls sharedScheduler("IO_POOL", cpuCores(0.5)).
+ │ │ Shared by all I/O queues:
+ │ │ - GRPCRemoteClient.* (gRPC to peer OAP nodes)
+ │ │ - GRPCMetricsExporter (gRPC metrics export)
+ │ │ - KafkaLogExporter (Kafka log export)
+ │ │ - KafkaTraceExporter (Kafka trace export)
+ │ │ - JDBCBatchDAO (JDBC batch writes)
+ │ │
+ │
+ │ Queues:
+ │
+ ├── "METRICS_L1_AGGREGATION" ──> BatchQueue<Metrics>
+ │ │ threads: cpuCores(1.0) → 8 on 8-core
+ │ │ partitions: threadMultiply(2) → 16 on 8-core
+ │ │ strategy: IF_POSSIBLE
+ │ │ handlerMap: { ServiceRespTimeMetrics.class -> handler-A (OAL),
+ │ │ ServiceCpmMetrics.class -> handler-B (OAL),
+ │ │ MeterMetrics_xxx.class -> handler-C (MAL), ... }
+ │ └── OAL and MAL metrics share the same L1 queue and thread pool.
+ │
+ ├── "METRICS_L2_PERSISTENT" ──> BatchQueue<Metrics>
+ │ │ threads: cpuCores(0.25) → 2 on 8-core
+ │ │ partitions: threadMultiply(2) → 4 on 8-core
+ │ │ strategy: BLOCKING
+ │ │ handlerMap: { ServiceRespTimeMetrics.class -> handler-D (OAL),
+ │ │ MeterMetrics_xxx.class -> handler-E (MAL), ... }
+ │ └── OAL and MAL metrics share the same L2 queue and thread pool.
+ │
+ ├── "TOPN_WORKER" ──> BatchQueue<TopN>
+ │ │ threads: fixed(1)
+ │ │ partitions: fixed(1)
+ │ │ strategy: BLOCKING
+ │ │ handlerMap: { DatabaseSlowStatement.class -> handler-F,
+ │ │ DatabaseSlowSql.class -> handler-G, ... }
+ │ └── drain: drainTo → groupBy(class) → dispatch to handler
+ │ (TopN is in-memory ranking computation — all types share one thread)
+ │
+ ├── "GRPCRemoteClient.peer1" ──> BatchQueue<RemoteMessage> (shared="IO_POOL", partitions=1, BLOCKING)
+ │ │ scheduler: shared "IO_POOL"
+ │ │ consumer: direct consumer for RemoteMessage
+ │ └── drain: drainTo → direct consumer
+ │
+ └── "JDBCBatchDAO" ──> BatchQueue<PrepareRequest> (shared="IO_POOL", partitions=1, BLOCKING)
+ │ scheduler: shared "IO_POOL"
+ │ consumer: direct consumer for PrepareRequest
+ └── same
+```
+
+### Core Concept
+
+**One queue type, one config, two scheduler modes.**
+
+- **`BatchQueueManager`** is the global singleton registry. It also manages named shared
+ schedulers for low-throughput queues. Users call `createIfAbsent(name, config)` to
+ get a named `BatchQueue`.
+- **`BatchQueue<T>`** has N partitions (configurable, default 1) and a handler map.
+ Producers round-robin data into partitions. On drain, each batch is **grouped by
+ message class** and dispatched to the handler registered for that class.
+- **Handler registration** via `queue.addHandler(Class, HandlerConsumer)`.
+ Each worker provides its own handler instance for its specific type.
+
+The handler map pattern works the same way regardless of partition count:
+- `threadMultiply(2)` with 100+ handlers → metrics aggregation (many types, shared partitions)
+- `partitions=1` with N handlers → TopN (multiple types, low throughput, shared 1 thread)
+- `partitions=1` with 1 consumer → I/O queue (gRPC client, exporter, JDBC)
+
+No need for separate queue classes. The difference is just configuration.
+
+### Why Shared Partitions + Handler Map
+
+In the old BulkConsumePool model with 100+ metric types:
+
+```
+Pool Thread-0 assigned: [service_resp_time channels, service_cpm channels, ...]
+Pool Thread-1 assigned: [endpoint_resp_time channels, endpoint_cpm channels, ...]
+...
+Each thread iterates ALL assigned channels per cycle, even if most are empty.
+```
+
+In the new model:
+
+```
+Partition-0: mixed data from all metric types (round-robin)
+Partition-1: mixed data from all metric types
+...
+Partition-N: mixed data from all metric types
+
+On drain of Partition-K:
+ batch = drainTo(list) // all data, mixed types
+ grouped = batch.groupBy(item.getClass()) // group by metric class
+ for each (class, items) in grouped:
+ handler = handlerMap.get(class) // lookup registered handler
+ handler.consume(items) // dispatch to the right worker
+```
+
+Benefits:
+- Partitions are created based on parallelism needs, not metric count.
+ 16 partitions (8 threads * 2) serve 100+ metric types.
+- No empty channel iteration — every partition gets data.
+- Handlers are registered on-demand. Adding a new metric type is just
+ `addHandler(NewMetrics.class, handlerInstance)`.
+- Each handler still processes only its own metric type's data in isolation.
+- I/O queues use the same structure with `partitions=1` and a direct consumer.
+
+### API
+
+```java
+// ── Metrics aggregation (dedicated pool, many types, handler map dispatch) ──
+
+BatchQueue<Metrics> l1Queue = BatchQueueManager.createIfAbsent(
+ "METRICS_L1_AGGREGATION",
+ BatchQueueConfig.<Metrics>builder()
+ .threads(ThreadPolicy.cpuCores(1.0)) // 1x CPU cores (e.g. 8 on 8-core)
+ .partitions(PartitionPolicy.threadMultiply(2)) // 2x resolved threads = 16 on 8-core
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.IF_POSSIBLE)
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+);
+
+// Each MetricsAggregateWorker registers its inner class handler for its metric class.
+// Called per metric type in MetricsStreamProcessor.create() (100+ times):
+l1Queue.addHandler(metricsClass, new L1Handler()); // L1Handler is worker's inner class
+
+// Produce — data goes into a partition by round-robin
+// Adaptive backoff ensures fast re-poll (minIdleMs) when data is flowing.
+l1Queue.produce(metricsData);
+
+// ── TopN (shared queue — all TopN types share one thread, handler map dispatch) ──
+
+BatchQueue<TopN> topNQueue = BatchQueueManager.createIfAbsent(
+ "TOPN_WORKER",
+ BatchQueueConfig.<TopN>builder()
+ .threads(ThreadPolicy.fixed(1)) // all TopN types share 1 thread
+ .partitions(PartitionPolicy.fixed(1))
+ .bufferSize(1000)
+ .strategy(BufferStrategy.BLOCKING)
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+);
+
+// Each TopNWorker registers its handler for its specific TopN class (5-10 types):
+topNQueue.addHandler(topNClass, new TopNHandler()); // TopNHandler is worker's inner class
+
+topNQueue.produce(topNData);
+
+// ── I/O queues (shared scheduler, single consumer) ──
+// sharedScheduler() specifies both name and ThreadPolicy. BatchQueueManager creates the
+// shared ScheduledExecutorService on first reference, reuses it for subsequent queues.
+// No separate createSharedScheduler() call needed — no startup ordering dependency.
+//
+// All these queues share the same "IO_POOL" scheduler:
+// GRPCRemoteClient.* — gRPC streaming to peer OAP nodes
+// GRPCMetricsExporter — gRPC metrics export
+// KafkaLogExporter — Kafka log export
+// KafkaTraceExporter — Kafka trace export
+// JDBCBatchDAO — JDBC batch writes to database
+
+BatchQueue<RemoteMessage> grpcQueue = BatchQueueManager.createIfAbsent(
+ "GRPCRemoteClient.peer1",
+ BatchQueueConfig.<RemoteMessage>builder()
+ .sharedScheduler("IO_POOL", ThreadPolicy.cpuCores(0.5)) // creates IO_POOL on first use
+ .partitions(PartitionPolicy.fixed(1))
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.BLOCKING)
+ .consumer(new RemoteMessageHandler())
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+);
+
+// Another queue referencing the same "IO_POOL" — reuses existing scheduler.
+// If ThreadPolicy differs from the first creator, logs a warning (first one wins).
+BatchQueue<PrepareRequest> jdbcQueue = BatchQueueManager.createIfAbsent(
+ "JDBCBatchDAO",
+ BatchQueueConfig.<PrepareRequest>builder()
+ .sharedScheduler("IO_POOL", ThreadPolicy.cpuCores(0.5)) // reuses existing IO_POOL
+ .partitions(PartitionPolicy.fixed(1))
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.BLOCKING)
+ .consumer(new JDBCBatchHandler())
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+);
+
+// ── Lifecycle ──
+
+BatchQueueManager.shutdown("METRICS_L1_AGGREGATION");
+BatchQueueManager.shutdownAll(); // shuts down all queues + shared schedulers
+```
+
+### Classes
+
+```
+library-batch-queue/
+ src/main/java/org/apache/skywalking/oap/server/library/batchqueue/
+ BatchQueueManager.java — Global singleton registry + named shared schedulers
+ BatchQueue.java — Partitioned queue with dedicated or shared scheduler + handler map dispatch
+ BatchQueueConfig.java — Builder: threads/sharedScheduler, partitions, bufferSize, strategy
+ ThreadPolicy.java — Fixed or CPU-relative thread count: fixed(N) / cpuCores(double)
+ PartitionPolicy.java — Fixed or thread-relative partition count: fixed(N) / threadMultiply(N)
+ HandlerConsumer.java — Interface: void consume(List<T>), default void onIdle()
+ BufferStrategy.java — BLOCKING / IF_POSSIBLE
+ QueueErrorHandler.java — Functional interface: void onError(List<T>, Throwable)
+```
+
+### BatchQueueManager
+
+```java
+/**
+ * Global registry for batch queues and shared schedulers.
+ * Thread-safe. Queues are created by name and shared across modules.
+ *
+ * Two scheduler modes:
+ * - Dedicated: queue creates its own ScheduledExecutorService (for high-throughput queues).
+ * Configured via BatchQueueConfig.threads(ThreadPolicy).
+ * - Shared: queue uses a named shared scheduler managed by this manager (for low-throughput queues).
+ * Configured via BatchQueueConfig.sharedScheduler(name, ThreadPolicy).
+ * Multiple queues referencing the same name share one ScheduledExecutorService.
+ *
+ * Shared schedulers are created lazily on first queue reference — no separate
+ * setup step needed. This eliminates startup ordering dependencies.
+ */
+public class BatchQueueManager {
+ private static final ConcurrentHashMap<String, BatchQueue<?>> queues = new ConcurrentHashMap<>();
+ private static final ConcurrentHashMap<String, ScheduledExecutorService> sharedSchedulers = new ConcurrentHashMap<>();
+
+ /**
+ * Get or create a shared scheduler. Called internally by BatchQueue constructor
+ * when config specifies sharedScheduler(name, threads).
+ *
+ * - First call with a given name: creates the ScheduledExecutorService using
+ * threads.resolve() and caches it.
+ * - Subsequent calls with the same name: returns the cached scheduler.
+ * If the ThreadPolicy differs, logs a warning (first one wins).
+ *
+ * Thread-safe (ConcurrentHashMap.computeIfAbsent).
+ *
+ * Shared schedulers are owned by BatchQueueManager, NOT by any individual queue.
+ * They are destroyed only by shutdownAll() — never by individual queue shutdown.
+ */
+ static ScheduledExecutorService getOrCreateSharedScheduler(String name, ThreadPolicy threads);
+
+ /**
+ * Create a new named queue. Throws if name already exists.
+ */
+ public static <T> BatchQueue<T> create(String name, BatchQueueConfig<T> config);
+
+ /**
+ * Create if not present. Returns existing queue if name is taken.
+ *
+ * If the queue already exists, validates consistency:
+ * - Consumption mode conflict: throws IllegalStateException if the existing queue
+ * uses direct consumer mode (config.consumer set) but the new config does not,
+ * or vice versa. These two modes are mutually exclusive per queue.
+ * - Infrastructure settings: logs a warning if threads, partitions,
+ * bufferSize, or strategy differ between the existing and new config.
+ */
+ public static <T> BatchQueue<T> createIfAbsent(String name, BatchQueueConfig<T> config);
+
+ /**
+ * Get an existing queue by name. Returns null if not found.
+ */
+ public static <T> BatchQueue<T> get(String name);
+
+ /**
+ * Shutdown and remove a single queue by name.
+ * - Dedicated scheduler: shut down together with the queue.
+ * - Shared scheduler: NOT shut down. It is owned by BatchQueueManager
+ * and may still be used by other queues.
+ */
+ public static void shutdown(String name);
+
+ /**
+ * Shutdown all queues and all shared schedulers. Called during OAP server shutdown.
+ *
+ * Order:
+ * 1. Set running=false on all queues (stops drain loops from rescheduling)
+ * 2. Final drain of remaining data in each queue
+ * 3. Shut down all dedicated schedulers
+ * 4. Shut down all shared schedulers
+ * 5. Clear both registries
+ */
+ public static void shutdownAll();
+}
+```
+
+### BatchQueue
+
+```java
+/**
+ * A partitioned queue with handler-map-based dispatch.
+ *
+ * The scheduler is either dedicated (owned by this queue) or shared
+ * (managed by BatchQueueManager, shared with other queues).
+ *
+ * Partitions are created based on configured parallelism (default 1).
+ * Producers round-robin data across partitions.
+ * On drain, each batch is grouped by message class and dispatched to the
+ * registered handler for that class.
+ *
+ * Works uniformly for all use cases:
+ * - shared scheduler, partitions=1, one consumer → I/O queue (gRPC, Kafka, JDBC)
+ * - dedicated fixed(1), partitions=1, many handlers → TopN (all types share 1 thread)
+ * - dedicated cpuCores(1.0), threadMultiply(2), many handlers → metrics aggregation
+ */
+public class BatchQueue<T> {
+ private final String name;
+ private final ScheduledExecutorService scheduler;
+ private final boolean dedicatedScheduler; // true = owned by this queue, false = shared
+ private final ArrayBlockingQueue<T>[] partitions;
+ private final ConcurrentHashMap<Class<? extends T>, HandlerConsumer<T>> handlerMap;
+ private final BatchQueueConfig<T> config;
+ private final AtomicInteger roundRobinIndex = new AtomicInteger(0);
+ private volatile boolean running;
+
+ /**
+ * Partition assignment per drain task. Each drain task owns a set of partition indices.
+ *
+ * Dedicated mode: one drain task per thread, partitions assigned round-robin.
+ * threads=4, partitions=8: task[0]→[0,4], task[1]→[1,5], task[2]→[2,6], task[3]→[3,7]
+ *
+ * Shared mode: single drain task covering ALL partitions (partitions typically = 1).
+ * assignedPartitions = { [0] } (one task, one partition)
+ */
+ private final int[][] assignedPartitions;
+
+ /**
+ * Per-task count of consecutive idle cycles (all assigned partitions empty).
+ * Used to compute adaptive backoff sleep interval.
+ */
+ private final int[] consecutiveIdleCycles;
+
+ BatchQueue(String name, BatchQueueConfig<T> config) {
+ this.name = name;
+ this.config = config;
+ this.handlerMap = new ConcurrentHashMap<>();
+
+ int taskCount;
+ if (config.getSharedSchedulerName() != null) {
+ // Shared scheduler mode: get-or-create shared scheduler from BatchQueueManager.
+ ScheduledExecutorService sharedScheduler = BatchQueueManager.getOrCreateSharedScheduler(
+ config.getSharedSchedulerName(), config.getSharedSchedulerThreads());
+
+ int partitionCount = config.getPartitions().resolve(1);
+ this.partitions = new ArrayBlockingQueue[partitionCount];
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i] = new ArrayBlockingQueue<>(config.getBufferSize());
+ }
+
+ this.scheduler = sharedScheduler;
+ this.dedicatedScheduler = false;
+ taskCount = 1;
+ this.assignedPartitions = new int[][] {
+ java.util.stream.IntStream.range(0, partitions.length).toArray()
+ };
+ } else {
+ // Dedicated scheduler mode: resolve threads and partitions.
+ int threadCount = config.getThreads().resolve(); // cpuCores(1.0) → 8 on 8-core
+ int partitionCount = config.getPartitions().resolve(threadCount);
+
+ // Validation: if partitions < threads, cut threads to match and warn.
+ if (partitionCount < threadCount) {
+ log.warn("BatchQueue[{}]: partitions({}) < threads({}), "
+ + "reducing threads to {}",
+ name, partitionCount, threadCount, partitionCount);
+ threadCount = partitionCount;
+ }
+
+ this.partitions = new ArrayBlockingQueue[partitionCount];
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i] = new ArrayBlockingQueue<>(config.getBufferSize());
+ }
+
+ this.scheduler = Executors.newScheduledThreadPool(
+ threadCount,
+ new ThreadFactoryBuilder().setNameFormat("BatchQueue-" + name + "-%d").build()
+ );
+ this.dedicatedScheduler = true;
+ taskCount = threadCount;
+
+ // Assign partitions to threads by round-robin.
+ // threads=4, partitions=8: task[0]→[0,4], task[1]→[1,5], ...
+ // threads=4, threadMultiply(2)=8: same
+ // threads=8, partitions=8: task[0]→[0], task[1]→[1], ...
+ this.assignedPartitions = new int[taskCount][];
+ List<List<Integer>> assignment = new ArrayList<>();
+ for (int t = 0; t < taskCount; t++) {
+ assignment.add(new ArrayList<>());
+ }
+ for (int p = 0; p < partitions.length; p++) {
+ assignment.get(p % taskCount).add(p);
+ }
+ for (int t = 0; t < taskCount; t++) {
+ assignedPartitions[t] = assignment.get(t).stream().mapToInt(Integer::intValue).toArray();
+ }
+ }
+
+ // Kick off one self-rescheduling drain task per assignment.
+ this.consecutiveIdleCycles = new int[taskCount];
+ this.running = true;
+ for (int t = 0; t < taskCount; t++) {
+ scheduleDrain(t);
+ }
+ }
+
+ /**
+ * Schedule the next drain with adaptive backoff.
+ *
+ * Idle count 0 (just had data): sleep = minIdleMs (e.g. 5ms)
+ * Idle count 1: sleep = minIdleMs * 2 (e.g. 10ms)
+ * Idle count 2: sleep = minIdleMs * 4 (e.g. 20ms)
+ * ...
+ * Idle count N: sleep = min(minIdleMs * 2^N, maxIdleMs)
+ */
+ private void scheduleDrain(int taskIndex) {
+ int idleCount = consecutiveIdleCycles[taskIndex];
+ long delay = Math.min(
+ config.getMinIdleMs() * (1L << Math.min(idleCount, 20)),
+ config.getMaxIdleMs()
+ );
+ scheduler.schedule(
+ () -> drainLoop(taskIndex),
+ delay,
+ TimeUnit.MILLISECONDS
+ );
+ }
+
+ /**
+ * Register a handler for a specific message class type.
+ * Multiple metric types can each register their own handler instance.
+ */
+ public void addHandler(Class<? extends T> type, HandlerConsumer<T> handler);
+
+ /**
+ * Produce data into a partition (round-robin).
+ * BLOCKING: waits if the selected partition is full (queue.put).
+ * IF_POSSIBLE: returns false if full (queue.offer).
+ */
+ public boolean produce(T data) {
+ int index = Math.abs(roundRobinIndex.getAndIncrement() % partitions.length);
+ if (config.getStrategy() == BufferStrategy.BLOCKING) {
+ partitions[index].put(data); // blocks until space available
+ return true;
+ } else {
+ return partitions[index].offer(data); // returns false if full
+ }
+ }
+
+ /**
+ * Drain loop for one thread: iterates ALL assigned partitions in a round-robin loop.
+ * Only stops when ALL assigned partitions are empty in a full cycle.
+ *
+ * Each cycle:
+ * 1. Drain ALL assigned partitions into one combined batch
+ * 2. If combined batch is empty → all partitions empty → onIdle, break
+ * 3. dispatch(combined batch) → handlers get ALL data of their type as one list
+ * 4. Loop back to step 1 (more data may have arrived during dispatch)
+ *
+ * Example: Thread-0 owns partitions [0, 4]
+ * cycle 1: drain(0)→[A1,B1,A2] drain(4)→[C1,A3]
+ * combined = [A1,B1,A2,C1,A3]
+ * dispatch → groupBy class:
+ * handlerA.consume([A1,A2,A3]) ← all A's in one call
+ * handlerB.consume([B1])
+ * handlerC.consume([C1])
+ * cycle 2: drain(0)→[A4] drain(4)→[]
+ * combined = [A4] (not empty → dispatch)
+ * cycle 3: drain(0)→[] drain(4)→[]
+ * combined = [] → onIdle, reschedule
+ */
+ void drainLoop(int taskIndex) {
+ int[] myPartitions = assignedPartitions[taskIndex];
+ try {
+ while (running) {
+ // Step 1: drain ALL assigned partitions into one batch
+ List<T> combined = new ArrayList<>();
+ for (int partitionIndex : myPartitions) {
+ partitions[partitionIndex].drainTo(combined);
+ }
+
+ // Step 2: if nothing across all partitions, we are idle
+ if (combined.isEmpty()) {
+ consecutiveIdleCycles[taskIndex]++;
+ notifyIdle();
+ break; // reschedule with backoff
+ }
+
+ // Had data → reset backoff so next reschedule is fast
+ consecutiveIdleCycles[taskIndex] = 0;
+
+ // Step 3: dispatch the combined batch
+ dispatch(combined);
+ // Step 4: loop immediately — more data may have arrived
+ }
+ } finally {
+ if (running) {
+ scheduleDrain(taskIndex);
+ }
+ }
+ }
+
+ void shutdown() {
+ running = false;
+ // Final drain of remaining data across all partitions
+ List<T> combined = new ArrayList<>();
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i].drainTo(combined);
+ }
+ if (!combined.isEmpty()) {
+ dispatch(combined);
+ }
+ // Only shut down the scheduler if this queue owns it.
+ // Shared schedulers are shut down by BatchQueueManager.shutdownAll().
+ if (dedicatedScheduler) {
+ scheduler.shutdown();
+ }
+ }
+}
+```
+
+**Dispatch and idle notification:**
+
+```java
+private void dispatch(List<T> batch) {
+ // Direct consumer mode: pass the whole batch, no groupBy overhead.
+ if (config.getConsumer() != null) {
+ try {
+ config.getConsumer().consume(batch);
+ } catch (Throwable t) {
+ config.getErrorHandler().onError(batch, t);
+ }
+ return;
+ }
+
+ // Handler map mode: group by class type and dispatch to registered handlers.
+ Map<Class<?>, List<T>> grouped = new HashMap<>();
+ for (T item : batch) {
+ grouped.computeIfAbsent(item.getClass(), k -> new ArrayList<>()).add(item);
+ }
+
+ for (Map.Entry<Class<?>, List<T>> entry : grouped.entrySet()) {
+ HandlerConsumer<T> handler = handlerMap.get(entry.getKey());
+ if (handler != null) {
+ try {
+ handler.consume(entry.getValue());
+ } catch (Throwable t) {
+ config.getErrorHandler().onError(entry.getValue(), t);
+ }
+ }
+ }
+}
+
+private void notifyIdle() {
+ if (config.getConsumer() != null) {
+ config.getConsumer().onIdle();
+ } else {
+ handlerMap.values().forEach(HandlerConsumer::onIdle);
+ }
+}
+```
+
+**Consumer workflow (end-to-end):**
+
+```
+Producer threads Consumer thread (Thread-0, owns partitions [0, 4])
+────────────── ─────────────────────────────────────────────────
+
+produce(A1) ──offer/put──► Partition-0: [A1, C1, B1]
+produce(B1) ──offer/put──► Partition-4: [A2, A3]
+produce(C1) ──┘ (round-robin)
+
+ ┌─── scheduleDrain(0) after adaptive backoff delay
+ │
+ ▼
+ drainLoop(taskIndex=0)
+ │
+ ┌────────┴───────────────────────────────────────────┐
+ │ while (running): │
+ │ │
+ │ ── Step 1: drain ALL assigned partitions ── │
+ │ combined = [] │
+ │ Partition-0.drainTo(combined) → [A1, C1, B1] │
+ │ Partition-4.drainTo(combined) → [A1,C1,B1,A2,A3]│
+ │ │
+ │ ── Step 2: check if empty ── │
+ │ combined not empty → dispatch │
+ │ │
+ │ ── Step 3: dispatch(combined) ── │
+ │ ┌─ config.consumer set? │
+ │ │ YES → consumer.consume(combined) ─── done │
+ │ │ NO → groupBy class: │
+ │ │ MetricA.class → [A1, A2, A3] │
+ │ │ MetricB.class → [B1] │
+ │ │ MetricC.class → [C1] │
+ │ │ for each group: │
+ │ │ handler = handlerMap.get(class) │
+ │ │ handler.consume(group) │
+ │ │ ↓ │
+ │ │ L1Handler_A.consume([A1, A2, A3]) │
+ │ │ → workerA.onWork([A1, A2, A3]) │
+ │ │ → mergeDataCache.accept(A1) │
+ │ │ → mergeDataCache.accept(A2) │
+ │ │ → mergeDataCache.accept(A3) │
+ │ │ → flush() if period elapsed │
+ │ │ L1Handler_B.consume([B1]) │
+ │ │ → workerB.onWork([B1]) │
+ │ │ L1Handler_C.consume([C1]) │
+ │ │ → workerC.onWork([C1]) │
+ │ └───────────────────────────────────────────── │
+ │ │
+ │ ── Step 4: loop immediately (more may have come)──│
+ │ │
+ │ ── next cycle ── │
+ │ combined = [] │
+ │ Partition-0.drainTo(combined) → [] │
+ │ Partition-4.drainTo(combined) → [] │
+ │ combined is empty │
+ │ → notifyIdle() │
+ │ handlerMap.values().forEach(::onIdle) │
+ │ ↓ │
+ │ L1Handler_A.onIdle() │
+ │ → workerA.flush() (force flush cache) │
+ │ L1Handler_B.onIdle() │
+ │ → workerB.flush() │
+ │ ... │
+ │ consecutiveIdleCycles[0]++ (e.g. now = 1) │
+ │ break │
+ └────────────────────────────────────────────────────┘
+ │
+ ▼ (finally block)
+ scheduleDrain(0)
+ │
+ ├─ idleCount=0 (just had data): wait 5ms ← fast re-poll
+ ├─ idleCount=1: wait 10ms
+ ├─ idleCount=2: wait 20ms
+ ├─ idleCount=3: wait 40ms
+ ├─ idleCount=4: wait 80ms
+ ├─ idleCount=5: wait 160ms
+ ├─ idleCount=6+: wait 200ms ← capped at maxIdleMs
+ │
+ ▼
+ drainLoop again
+ │
+ (if data found → idleCount resets to 0 → back to fast polling)
+```
+
+**Key points:**
+- Each cycle drains ALL assigned partitions into one combined batch before dispatching.
+- `dispatch()` is called once per cycle with all data from all partitions combined.
+- In handler map mode, `groupBy(class)` collects all items of the same type across all
+ partitions into one list. The handler receives ALL available data of its type in a single
+ `consume()` call — e.g., `[A1, A2, A3]` not three separate calls.
+- The handler's `consume()` runs **synchronously** inside the drain thread. The handler
+ (an inner class of the worker) directly accesses the worker's fields — merge cache,
+ counters, flush logic — with no extra threading.
+- If any partition had data, loop immediately to check all partitions again.
+- `onIdle()` fires only when ALL assigned partitions are empty in a full cycle, giving
+ handlers a chance to flush periodic caches (e.g., L1 aggregation merge cache → nextWorker).
+- **Adaptive backoff**: after data, re-poll in `minIdleMs` (5ms). Each consecutive empty
+ cycle doubles the sleep, capping at `maxIdleMs` (200ms). Data resets to fast polling.
+
+**Two consumption modes, same queue class:**
+- **Direct consumer** (`config.consumer` set) — whole batch goes to one handler, no groupBy.
+ Use for I/O queues where all data is the same type (gRPC, Kafka, JDBC).
+- **Handler map** (`addHandler` called) — batch grouped by class, dispatched per type.
+ Use for metrics aggregation (L1/L2) and TopN with many types sharing partitions.
+
+If both are set, direct consumer takes priority (handler map is ignored).
+If neither is set, data is drained but silently dropped.
+
+### BatchQueueConfig
+
+```java
+@Builder
+public class BatchQueueConfig<T> {
+ /**
+ * Thread policy for a dedicated ScheduledExecutorService.
+ * Resolved at queue construction time. Examples:
+ * ThreadPolicy.fixed(8) → always 8 threads
+ * ThreadPolicy.cpuCores(1.0) → 1x available CPU cores (8 on 8-core)
+ * ThreadPolicy.cpuCores(0.25) → 0.25x CPU cores (2 on 8-core, min 1)
+ *
+ * When set, the queue creates its own scheduler.
+ * When null, sharedScheduler must be set — the queue uses a shared scheduler.
+ *
+ * Use dedicated pools for high-throughput queues (metrics L1/L2 aggregation)
+ * where you need guaranteed thread capacity.
+ */
+ private ThreadPolicy threads;
+
+ /**
+ * Shared scheduler name and its ThreadPolicy. Set via the builder method
+ * sharedScheduler(name, threads) which populates both fields together.
+ * Mutually exclusive with the threads field above.
+ *
+ * When set, the queue registers its drain tasks on a shared
+ * ScheduledExecutorService managed by BatchQueueManager.
+ * The shared scheduler is created lazily on first queue reference —
+ * no separate setup step needed. Subsequent queues with the same name
+ * reuse the existing scheduler (ThreadPolicy mismatch logs a warning).
+ *
+ * Use for low-throughput I/O queues (gRPC, Kafka, JDBC) to reduce OS thread count.
+ * Multiple queues sharing the same scheduler name share the same thread pool.
+ *
+ * Exactly one of threads or sharedScheduler must be set.
+ */
+ private String sharedSchedulerName;
+ private ThreadPolicy sharedSchedulerThreads;
+
+ /**
+ * Number of partitions, or a policy to derive it from resolved thread count.
+ *
+ * Can be set as:
+ * - Absolute: fixed(8) → exactly 8 partitions.
+ * - Relative: threadMultiply(2) → 2x resolved thread count.
+ * e.g. cpuCores(0.5) on 8-core = 4 threads, threadMultiply(2) → 8 partitions.
+ *
+ * Default: fixed(1).
+ *
+ * Validation (applied at construction time for dedicated scheduler mode):
+ * - If partitions < resolved thread count, thread count is reduced to match
+ * partitions and a warning is logged. No point having more threads than partitions.
+ */
+ @Builder.Default
+ private PartitionPolicy partitions = PartitionPolicy.fixed(1);
+
+ /**
+ * Buffer size per partition.
+ */
+ @Builder.Default
+ private int bufferSize = 10_000;
+
+ /**
+ * BLOCKING: producer waits when buffer full.
+ * IF_POSSIBLE: producer gets false when buffer full.
+ */
+ @Builder.Default
+ private BufferStrategy strategy = BufferStrategy.BLOCKING;
+
+ /**
+ * Direct consumer for the whole batch. When set, all drained data goes to this
+ * handler without class-based grouping. Takes priority over handler map.
+ *
+ * Use this for I/O queues where all data is the same type (gRPC, Kafka, JDBC).
+ * Leave null to use handler map dispatch via addHandler().
+ */
+ private HandlerConsumer<T> consumer;
+
+ /**
+ * Called when a handler throws during consume.
+ */
+ private QueueErrorHandler<T> errorHandler;
+
+ /**
+ * Minimum idle sleep in milliseconds. Default 5ms.
+ * Used as the base interval when data was recently consumed.
+ * The thread re-polls quickly to catch new data with low latency.
+ */
+ @Builder.Default
+ private long minIdleMs = 5;
+
+ /**
+ * Maximum idle sleep in milliseconds. Default 200ms.
+ * After several consecutive empty cycles, the sleep interval backs off
+ * exponentially from minIdleMs up to this cap.
+ *
+ * Backoff: sleep = min(minIdleMs * 2^consecutiveIdleCycles, maxIdleMs)
+ * idle 0: 5ms → idle 1: 10ms → idle 2: 20ms → idle 3: 40ms
+ * → idle 4: 80ms → idle 5: 160ms → idle 6+: 200ms (capped)
+ */
+ @Builder.Default
+ private long maxIdleMs = 200;
+}
+```
+
+### HandlerConsumer
+
+```java
+/**
+ * Handler for processing a batch of data for a specific type.
+ * Each metric type (or I/O queue user) provides its own handler instance.
+ */
+public interface HandlerConsumer<T> {
+ /**
+ * Process a batch of data belonging to this handler's type.
+ */
+ void consume(List<T> data);
+
+ /**
+ * Called when there is nothing to consume. Can be used as a timer trigger
+ * (e.g. flush L1 aggregation cache periodically).
+ */
+ default void onIdle() {
+ }
+}
+```
+
+### ThreadPolicy
+
+```java
+/**
+ * Determines the number of threads for a BatchQueue's dedicated scheduler
+ * or for a shared scheduler created via BatchQueueManager.
+ *
+ * Two modes:
+ * - fixed(N): exactly N threads, regardless of hardware.
+ * - cpuCores(multiplier): multiplier * Runtime.availableProcessors(), rounded.
+ *
+ * Resolved value is always >= 1 — every pool must have at least one thread.
+ * fixed() requires count >= 1 at construction. cpuCores() applies max(1, ...) at resolution.
+ */
+public class ThreadPolicy {
+ private final int fixedCount; // > 0 for fixed mode, 0 for cpuCores mode
+ private final double cpuMultiplier; // > 0 for cpuCores mode, 0 for fixed mode
+
+ /**
+ * Fixed number of threads. Count must be >= 1.
+ * Example: fixed(1) → always 1 thread.
+ * fixed(8) → always 8 threads.
+ * Throws IllegalArgumentException if count < 1.
+ */
+ public static ThreadPolicy fixed(int count);
+
+ /**
+ * Threads = multiplier * available CPU cores, rounded, min 1.
+ * Multiplier must be > 0.
+ * Example on 8-core machine:
+ * cpuCores(1.0) → 8 threads
+ * cpuCores(0.5) → 4 threads
+ * cpuCores(0.25) → 2 threads
+ * cpuCores(2.0) → 16 threads
+ * Example on 2-core machine:
+ * cpuCores(0.25) → 1 thread (min 1, never 0)
+ */
+ public static ThreadPolicy cpuCores(double multiplier);
+
+ /**
+ * Resolve the actual thread count. Always returns >= 1.
+ * For fixed mode, returns fixedCount.
+ * For cpuCores mode, returns max(1, round(cpuMultiplier * availableProcessors())).
+ */
+ public int resolve();
+}
+```
+
+### PartitionPolicy
+
+```java
+/**
+ * Determines the number of partitions for a BatchQueue.
+ *
+ * Two modes:
+ * - fixed(N): exactly N partitions, regardless of thread count.
+ * - threadMultiply(N): N * resolved thread count.
+ *
+ * At queue construction time, the resolved partition count is validated:
+ * if partitions < resolved thread count, thread count is reduced to match
+ * partitions and a warning is logged.
+ */
+public class PartitionPolicy {
+ private final int fixedCount; // > 0 for fixed mode
+ private final int multiplier; // > 0 for threadMultiply mode
+
+ /**
+ * Fixed number of partitions.
+ * Example: fixed(8) → always 8 partitions.
+ */
+ public static PartitionPolicy fixed(int count);
+
+ /**
+ * Partitions = multiplier * resolved thread count.
+ * Example: threadMultiply(2) with 4 resolved threads → 8 partitions.
+ */
+ public static PartitionPolicy threadMultiply(int multiplier);
+
+ /**
+ * Resolve the actual partition count.
+ * For fixed mode, returns fixedCount.
+ * For threadMultiply mode, returns multiplier * resolvedThreadCount.
+ */
+ public int resolve(int resolvedThreadCount);
+}
+```
+
+### Implementing HandlerConsumer
+
+Each worker creates a handler as an **inner class** that directly accesses the worker's
+fields (merge cache, telemetry counters, flush logic, etc.). The handler instance is
+registered per metric class — one handler per worker, one worker per metric type.
+
+#### L1 Aggregation (MetricsAggregateWorker)
+
+Current code uses an inner `AggregatorConsumer` that calls `onWork()` and `flush()`:
+
+```java
+// Current: inner class IConsumer accesses outer worker fields
+private class AggregatorConsumer implements IConsumer<Metrics> {
+ public void consume(List<Metrics> data) {
+ MetricsAggregateWorker.this.onWork(data); // accesses mergeDataCache
+ }
+ public void nothingToConsume() {
+ flush(); // accesses lastSendTime, nextWorker
+ }
+}
+```
+
+New code — same pattern, just implements `HandlerConsumer` instead of `IConsumer`.
+Each `MetricsAggregateWorker` instance creates its own handler and registers it for
+its specific metric class:
+
+```java
+public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
+ private final MergableBufferedData<Metrics> mergeDataCache;
+ private final AbstractWorker<Metrics> nextWorker;
+ private final BatchQueue<Metrics> l1Queue;
+
+ MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder,
+ AbstractWorker<Metrics> nextWorker,
+ String modelName,
+ Class<? extends Metrics> metricsClass,
+ ...) {
+ this.nextWorker = nextWorker;
+ this.mergeDataCache = new MergableBufferedData<>();
+
+ // Get or create the shared L1 queue (idempotent)
+ this.l1Queue = BatchQueueManager.createIfAbsent(
+ "METRICS_L1_AGGREGATION",
+ BatchQueueConfig.<Metrics>builder()
+ .threads(ThreadPolicy.cpuCores(1.0)) // 1x CPU cores
+ .partitions(PartitionPolicy.threadMultiply(2)) // 2x resolved threads
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.IF_POSSIBLE)
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+ );
+
+ // Register this worker's handler for its specific metric class.
+ // The inner class directly accesses mergeDataCache, nextWorker, etc.
+ l1Queue.addHandler(metricsClass, new L1Handler());
+ }
+
+ @Override
+ public void in(Metrics metrics) {
+ l1Queue.produce(metrics);
+ }
+
+ private void onWork(List<Metrics> data) {
+ data.forEach(mergeDataCache::accept);
+ flush();
+ }
+
+ private void flush() {
+ if (System.currentTimeMillis() - lastSendTime > l1FlushPeriod) {
+ mergeDataCache.read().forEach(nextWorker::in);
+ lastSendTime = System.currentTimeMillis();
+ }
+ }
+
+ // Inner class handler — accesses worker fields directly
+ private class L1Handler implements HandlerConsumer<Metrics> {
+ @Override
+ public void consume(List<Metrics> data) {
+ MetricsAggregateWorker.this.onWork(data);
+ }
+
+ @Override
+ public void onIdle() {
+ MetricsAggregateWorker.this.flush();
+ }
+ }
+}
+```
+
+Key point: 100+ `MetricsAggregateWorker` instances are created (one per metric type,
+both OAL and MAL), each registers its own `L1Handler` inner class instance on the
+same shared queue. Handler map dispatch routes each metric class to its own worker's
+`mergeDataCache` — OAL and MAL handlers coexist in the same queue without interference.
+
+#### L2 Persistent (MetricsPersistentMinWorker)
+
+Same pattern — inner class handler accesses the worker's `onWork()` method:
+
+```java
+public class MetricsPersistentMinWorker extends MetricsPersistentWorker {
+ private final BatchQueue<Metrics> l2Queue;
+
+ MetricsPersistentMinWorker(..., Class<? extends Metrics> metricsClass, ...) {
+ super(...);
+
+ this.l2Queue = BatchQueueManager.createIfAbsent(
+ "METRICS_L2_PERSISTENT",
+ BatchQueueConfig.<Metrics>builder()
+ .threads(ThreadPolicy.cpuCores(0.25)) // 0.25x CPU cores (2 on 8-core)
+ .partitions(PartitionPolicy.threadMultiply(2)) // 2x resolved threads
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.BLOCKING)
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+ );
+
+ l2Queue.addHandler(metricsClass, new L2Handler());
+ }
+
+ @Override
+ public void in(Metrics metrics) {
+ l2Queue.produce(metrics);
+ }
+
+ // Inner class handler — accesses worker's onWork, queuePercentageGauge, etc.
+ private class L2Handler implements HandlerConsumer<Metrics> {
+ @Override
+ public void consume(List<Metrics> data) {
+ queuePercentageGauge.setValue(...);
+ MetricsPersistentMinWorker.this.onWork(data);
+ }
+ }
+}
+```
+
+#### TopN (TopNWorker — shared queue, handler map dispatch)
+
+All TopN types share one queue. Same handler map pattern as L1/L2 — each TopNWorker
+registers its inner class handler for its specific TopN class:
+
+```java
+public class TopNWorker extends PersistenceWorker<TopN> {
+ private final BatchQueue<TopN> topNQueue;
+
+ TopNWorker(..., Class<? extends TopN> topNClass, ...) {
+ // Get or create the shared TopN queue (idempotent)
+ this.topNQueue = BatchQueueManager.createIfAbsent(
+ "TOPN_WORKER",
+ BatchQueueConfig.<TopN>builder()
+ .threads(ThreadPolicy.fixed(1)) // all TopN types share 1 thread
+ .partitions(PartitionPolicy.fixed(1))
+ .bufferSize(1000)
+ .strategy(BufferStrategy.BLOCKING)
+ .errorHandler((data, t) -> log.error(t.getMessage(), t))
+ .build()
+ );
+
+ // Register this worker's handler for its specific TopN class
+ topNQueue.addHandler(topNClass, new TopNHandler());
+ }
+
+ @Override
+ public void in(TopN topN) {
+ topNQueue.produce(topN);
+ }
+
+ // Inner class — accesses worker's limitedSizeBufferedData, reportPeriod, etc.
+ private class TopNHandler implements HandlerConsumer<TopN> {
+ @Override
+ public void consume(List<TopN> data) {
+ TopNWorker.this.onWork(data);
+ }
+
+ @Override
+ public void onIdle() {
+ TopNWorker.this.flushIfNeeded();
+ }
+ }
+}
+```
+
+### Thread Reduction
+
+Thread counts scale with CPU cores via ThreadPolicy. OAL and MAL share the same L1/L2 pools.
+Example on 8-core machine:
+
+| Before (DataCarrier) | Threads | After (BatchQueue) | Threads (8-core) |
+|---------------------------------|---------|-------------------------------------------------|------------------|
+| L1 OAL pool | 24 | METRICS_L1_AGGREGATION (cpuCores(1.0)) | 8 |
+| L1 MAL pool | 2 | *(shared with L1 above)* | *(shared)* |
+| L2 OAL pool | 2 | METRICS_L2_PERSISTENT (cpuCores(0.25)) | 2 |
+| L2 MAL pool | 1 | *(shared with L2 above)* | *(shared)* |
+| TopNWorker (5-10 types) | 5-10 | TOPN_WORKER (fixed(1), handler map) | 1 |
+| GRPCRemoteClient (2-4 peers) | 2-4 | GRPCRemoteClient.* (shared IO_POOL) | |
+| JDBCBatchDAO | 2-4 | JDBCBatchDAO (shared IO_POOL) | |
+| Exporters (gRPC/Kafka) | 0-3 | Exporter.* (shared IO_POOL) | |
+| | | **IO_POOL shared scheduler (cpuCores(0.5))** | **4** |
+| **Total** | **~38-48** | **Total** | **~15** |
+
+On different hardware:
+
+| Machine | L1 (OAL+MAL) | L2 (OAL+MAL) | TopN | IO_POOL | Total |
+|-----------|---------------|---------------|------|---------|-------|
+| 2-core | 2 | 1 | 1 | 1 | 5 |
+| 4-core | 4 | 1 | 1 | 2 | 8 |
+| 8-core | 8 | 2 | 1 | 4 | 15 |
+| 16-core | 16 | 4 | 1 | 8 | 29 |
+
+Savings (8-core):
+- L1: 24+2 → 8 threads (OAL+MAL share one pool, CPU-relative, no empty channel iteration)
+- L2: 2+1 → 2 threads (OAL+MAL share one pool)
+- TopN: 5-10 → 1 thread (all types share one queue with handler map dispatch)
+- I/O queues: 4-11 → 4 threads (shared IO_POOL for gRPC, Kafka, JDBC)
+- Total: from ~38-48 OS threads down to ~15
+
+IO_POOL queues (all do network or database I/O, low-throughput, bursty):
+
+| Queue Name | Current Source | I/O Type | Threads Before |
+|-------------------------|------------------------|---------------|----------------|
+| GRPCRemoteClient.* | GRPCRemoteClient.java | gRPC network | 2-4 (per peer) |
+| GRPCMetricsExporter | GRPCMetricsExporter.java| gRPC network | 1 |
+| KafkaLogExporter | KafkaLogExporter.java | Kafka network | 1 |
+| KafkaTraceExporter | KafkaTraceExporter.java| Kafka network | 1 |
+| JDBCBatchDAO | JDBCBatchDAO.java | JDBC database | 2-4 |
+| **Subtotal** | | | **7-14** |
+| **After (IO_POOL)** | | | **cpuCores(0.5)** |
+
+Number of partitions/buffers:
+
+| Before | Count | After | Count |
+|--------------------------------|----------|------------------------------------|-------|
+| L1 OAL channels (100+ * 2ch) | 200+ | L1 partitions (8 threads * 2) | 16 |
+| L1 MAL channels (N * 1ch) | N | *(shared with L1 above)* | *(0)* |
+| L2 OAL channels (100+ * 1ch) | 100+ | L2 partitions (2 threads * 2) | 4 |
+| L2 MAL channels (N * 1ch) | N | *(shared with L2 above)* | *(0)* |
+| TopN buffers (5-10 types) | 5-10 | TOPN_WORKER partitions | 1 |
+| I/O buffers (gRPC, JDBC, etc.) | 5-8 | I/O queue partitions | 5-8 |
+| **Total buffers** | **300+** | **Total buffers** | **~28** |
+
+### What Gets Dropped
+
+| DataCarrier Feature | Status | Reason |
+|-------------------------------------|-----------|-------------------------------------------------------|
+| One queue per metric type | Dropped | Shared partitions + handler map instead |
+| Separate OAL / MAL pools | Dropped | OAL and MAL share L1/L2 queues (handler map dispatch) |
+| One thread per TopN type | Dropped | All TopN types share one TOPN_WORKER queue |
+| Multi-channel per DataCarrier | Dropped | Single partition array replaces multi-channel |
+| IDataPartitioner | Dropped | Simple round-robin on partition array |
+| Consumer instantiation by class | Dropped | All callers use instance-based handlers |
+| Consumer init(Properties) | Dropped | Not used by any production consumer |
+| EnvUtil override | Dropped | Configuration via application.yml |
+| Two separate queue classes | Dropped | One `BatchQueue` with configurable scheduler modes |
+| BulkConsumePool / ConsumerPoolFactory | Dropped | Dedicated/shared ScheduledExecutorService replaces pool|
+| Fixed thread counts | Dropped | ThreadPolicy: CPU-relative (cpuCores) or fixed |
+| Signal-driven consumption | Dropped | Adaptive backoff replaces explicit notify |
+| Separate createSharedScheduler step | Dropped | Shared schedulers created lazily on first queue ref |
+
+### What Gets Preserved
+
+| Feature | How |
+|-----------------------|--------------------------------------------------------------|
+| Named queue management| `BatchQueueManager.create/createIfAbsent/get` by name |
+| Per-type isolation | `handlerMap` dispatches each class to its own handler |
+| Bounded buffer | ArrayBlockingQueue per partition |
+| BLOCKING strategy | `queue.put()` — producer blocks when full |
+| IF_POSSIBLE strategy | `queue.offer()` — returns false when full, data dropped |
+| Batch consumption | `drainTo(list)` — same as current |
+| Error handling | `errorHandler.onError(batch, throwable)` |
+| Nothing-to-consume | `handler.onIdle()` — called when all partitions empty |
+| Fast data response | Adaptive backoff (minIdleMs=5ms) replaces signal-driven mode |
+| Drain on shutdown | Manager shutdown drains all queues, then schedulers |
+| Produce-gate | `produce()` returns false if queue is shut down |
+| Hardware scaling | ThreadPolicy.cpuCores() scales threads with available cores |
diff --git a/oap-server/server-library/pom.xml b/oap-server/server-library/library-batch-queue/pom.xml
similarity index 64%
copy from oap-server/server-library/pom.xml
copy to oap-server/server-library/library-batch-queue/pom.xml
index 84eeae3260..455b77f447 100644
--- a/oap-server/server-library/pom.xml
+++ b/oap-server/server-library/library-batch-queue/pom.xml
@@ -19,25 +19,30 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>oap-server</artifactId>
+ <artifactId>server-library</artifactId>
<groupId>org.apache.skywalking</groupId>
<version>${revision}</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>server-library</artifactId>
- <packaging>pom</packaging>
- <modules>
- <module>library-module</module>
- <module>library-server</module>
- <module>library-util</module>
- <module>library-client</module>
- <module>library-elasticsearch-client</module>
- <module>library-datacarrier-queue</module>
- <module>library-kubernetes-support</module>
- <module>library-async-profiler-jfr-parser</module>
- <module>library-pprof-parser</module>
- <module>library-integration-test</module>
- <module>library-banyandb-client</module>
- </modules>
+ <artifactId>library-batch-queue</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
</project>
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueue.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueue.java
new file mode 100644
index 0000000000..6f2d04b79e
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueue.java
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * A partitioned, self-draining queue with type-based dispatch.
+ *
+ * <h3>Produce workflow</h3>
+ * <pre>
+ * produce(data)
+ * |
+ * +-- queue stopped? --> return false
+ * |
+ * +-- N == 1? --> partition[0] (skip selector)
+ * +-- N > 1? --> partition[selector.select(data, N)]
+ * |
+ * +-- BLOCKING strategy?
+ * | yes --> ArrayBlockingQueue.put(data) // blocks until space available
+ * | no --> ArrayBlockingQueue.offer(data) // returns false if full (drop)
+ * |
+ * +-- return true/false
+ * </pre>
+ *
+ * <h3>Consume workflow (drain loop, runs on scheduler threads)</h3>
+ * <pre>
+ * scheduleDrain(taskIndex) // schedule with adaptive backoff delay
+ * |
+ * +-- scheduler.schedule(drainLoop, delay)
+ * |
+ * +-- drainTo(combined) from each assigned partition
+ * |
+ * +-- combined is empty?
+ * | yes --> consecutiveIdleCycles++, notify onIdle(), break
+ * | no --> consecutiveIdleCycles = 0, dispatch(combined)
+ * | |
+ * | +-- single consumer set?
+ * | | yes --> consumer.consume(batch)
+ * | | no --> group by item.getClass()
+ * | | for each (class, subBatch):
+ * | | handlerMap.get(class).consume(subBatch)
+ * | |
+ * | +-- loop back to drainTo (drain until empty)
+ * |
+ * +-- finally: scheduleDrain(taskIndex) // re-schedule self
+ * </pre>
+ *
+ * <h3>Adaptive backoff</h3>
+ * Delay doubles on each consecutive idle cycle: {@code minIdleMs * 2^idleCount},
+ * capped at {@code maxIdleMs}. Resets to {@code minIdleMs} on first non-empty drain.
+ *
+ * <h3>Scheduler modes</h3>
+ * <ul>
+ * <li><b>Dedicated</b> — this queue owns its own ScheduledThreadPool. Each thread
+ * is assigned a fixed subset of partitions (round-robin). Thread count and
+ * partition count are configured independently.</li>
+ * <li><b>Shared</b> — the queue borrows a scheduler from {@link BatchQueueManager},
+ * shared with other queues. Only 1 drain task is submitted (drains all partitions).
+ * The shared scheduler is reference-counted and shut down when the last queue
+ * releases it.</li>
+ * </ul>
+ *
+ * <h3>Use case examples</h3>
+ * <pre>
+ * shared scheduler, partitions=1, one consumer --> I/O queue (gRPC, Kafka, JDBC)
+ * dedicated fixed(1), partitions=1, many handlers --> TopN (all types share 1 thread)
+ * dedicated cpuCores(1.0), threadMultiply(2),
+ * many handlers --> metrics aggregation
+ * </pre>
+ */
+@Slf4j
+public class BatchQueue<T> {
+ private final String name;
+
+ /** The thread pool that executes drain tasks. Either dedicated or shared. */
+ private final ScheduledExecutorService scheduler;
+
+ /** True if this queue owns the scheduler and should shut it down. */
+ private final boolean dedicatedScheduler;
+
+ /** Non-null only for shared schedulers; used to release the ref count on shutdown. */
+ private final String sharedSchedulerName;
+
+ /**
+ * Fixed-size partitions. Producers select a partition via {@link PartitionSelector}
+ * to reduce contention. Each partition is an {@link ArrayBlockingQueue} with
+ * capacity = bufferSize.
+ */
+ private final ArrayBlockingQueue<T>[] partitions;
+
+ /**
+ * Cached partition selector from config. Only used when {@code partitions.length > 1};
+ * single-partition queues bypass the selector entirely.
+ */
+ private final PartitionSelector<T> partitionSelector;
+
+ /**
+ * Type-based handler registry. When no single consumer is configured,
+ * drained items are grouped by {@code item.getClass()} and dispatched
+ * to the matching handler. Allows multiple metric/record types to share
+ * one queue while each type has its own processing logic.
+ */
+ private final ConcurrentHashMap<Class<?>, HandlerConsumer<T>> handlerMap;
+
+ private final BatchQueueConfig<T> config;
+
+ /** Set to false on {@link #shutdown()} to stop drain loops and reject new data. */
+ private volatile boolean running;
+
+ /**
+ * Which partitions each drain task is responsible for.
+ * {@code assignedPartitions[taskIndex] = int[] of partition indices}.
+ *
+ * <p>Dedicated mode: partitions are distributed across threads by round-robin.
+ * Example with 4 partitions and 2 threads:
+ * <pre>
+ * task 0 --> partitions [0, 2]
+ * task 1 --> partitions [1, 3]
+ * </pre>
+ *
+ * <p>Shared mode: a single task drains all partitions.
+ */
+ private final int[][] assignedPartitions;
+
+ /**
+ * Per-task count of consecutive idle cycles (no data drained).
+ * Used for adaptive exponential backoff in {@link #scheduleDrain}.
+ */
+ private final int[] consecutiveIdleCycles;
+
+ @SuppressWarnings("unchecked")
+ BatchQueue(final String name, final BatchQueueConfig<T> config) {
+ this.name = name;
+ this.config = config;
+ this.partitionSelector = config.getPartitionSelector();
+ this.handlerMap = new ConcurrentHashMap<>();
+
+ final int taskCount;
+ if (config.getSharedSchedulerName() != null) {
+ // ---- Shared scheduler mode ----
+ // Borrow a scheduler from BatchQueueManager (ref-counted).
+ // Only 1 drain task; it drains all partitions sequentially.
+ final ScheduledExecutorService sharedScheduler =
+ BatchQueueManager.getOrCreateSharedScheduler(
+ config.getSharedSchedulerName(), config.getSharedSchedulerThreads());
+
+ final int partitionCount = config.getPartitions().resolve(1);
+ this.partitions = new ArrayBlockingQueue[partitionCount];
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i] = new ArrayBlockingQueue<>(config.getBufferSize());
+ }
+
+ this.scheduler = sharedScheduler;
+ this.dedicatedScheduler = false;
+ this.sharedSchedulerName = config.getSharedSchedulerName();
+ taskCount = 1;
+ final int[] allPartitions = new int[partitions.length];
+ for (int i = 0; i < allPartitions.length; i++) {
+ allPartitions[i] = i;
+ }
+ this.assignedPartitions = new int[][] {allPartitions};
+ } else {
+ // ---- Dedicated scheduler mode ----
+ // Create a private ScheduledThreadPool. Each thread drains
+ // a fixed subset of partitions (no cross-thread contention).
+ int threadCount = config.getThreads().resolve();
+ final int partitionCount = config.getPartitions().resolve(threadCount);
+
+ if (partitionCount < threadCount) {
+ log.warn("BatchQueue[{}]: partitions({}) < threads({}), reducing threads to {}",
+ name, partitionCount, threadCount, partitionCount);
+ threadCount = partitionCount;
+ }
+
+ this.partitions = new ArrayBlockingQueue[partitionCount];
+ for (int i = 0; i < partitions.length; i++) {
+ partitions[i] = new ArrayBlockingQueue<>(config.getBufferSize());
+ }
+
+ this.scheduler = Executors.newScheduledThreadPool(threadCount, r -> {
+ final Thread t = new Thread(r);
+ t.setName("BatchQueue-" + name + "-" + t.getId());
+ t.setDaemon(true);
+ return t;
+ });
+ this.dedicatedScheduler = true;
+ this.sharedSchedulerName = null;
+ taskCount = threadCount;
+
+ // Distribute partitions across threads by round-robin
+ this.assignedPartitions = new int[taskCount][];
+ final List<List<Integer>> assignment = new ArrayList<>();
+ for (int t = 0; t < taskCount; t++) {
+ assignment.add(new ArrayList<>());
+ }
+ for (int p = 0; p < partitions.length; p++) {
+ assignment.get(p % taskCount).add(p);
+ }
+ for (int t = 0; t < taskCount; t++) {
+ final List<Integer> parts = assignment.get(t);
+ assignedPartitions[t] = new int[parts.size()];
+ for (int i = 0; i < parts.size(); i++) {
+ assignedPartitions[t][i] = parts.get(i);
+ }
+ }
+ }
+
+ this.consecutiveIdleCycles = new int[taskCount];
+ this.running = true;
+ // Kick off the drain loop for each task
+ for (int t = 0; t < taskCount; t++) {
+ scheduleDrain(t);
+ }
+ }
+
+ /**
+ * Schedule the next drain for the given task. The delay uses adaptive exponential
+ * backoff: {@code minIdleMs * 2^consecutiveIdleCycles}, capped at maxIdleMs.
+ * When data is flowing, consecutiveIdleCycles is 0, so delay = minIdleMs.
+ */
+ private void scheduleDrain(final int taskIndex) {
+ if (!running) {
+ return;
+ }
+ final int idleCount = consecutiveIdleCycles[taskIndex];
+ final long delay = Math.min(
+ config.getMinIdleMs() * (1L << Math.min(idleCount, 20)),
+ config.getMaxIdleMs()
+ );
+ try {
+ scheduler.schedule(() -> drainLoop(taskIndex), delay, TimeUnit.MILLISECONDS);
+ } catch (final Exception e) {
+ if (running) {
+ log.error("BatchQueue[{}]: failed to schedule drain task", name, e);
+ }
+ }
+ }
+
+ /**
+ * Register a type-based handler. Items whose {@code getClass()} matches the given
+ * type will be batched together and dispatched to this handler.
+ * Only used when no single consumer is set in config.
+ */
+ public void addHandler(final Class<? extends T> type, final HandlerConsumer<T> handler) {
+ handlerMap.put(type, handler);
+ }
+
+ /**
+ * Produce data into a partition selected by the configured {@link PartitionSelector}.
+ *
+ * <p>Single-partition queues bypass the selector entirely (always index 0).
+ * Multi-partition queues default to {@link PartitionSelector#typeHash()}, which
+ * uses {@code data.getClass().hashCode()} so all items of the same type land on
+ * the same partition — the HashMap grouping in {@link #dispatch} becomes
+ * effectively a no-op.
+ *
+ * <p>Behavior depends on {@link BufferStrategy}:
+ * <ul>
+ * <li>BLOCKING — blocks the caller until space is available</li>
+ * <li>IF_POSSIBLE — returns false immediately if the partition is full (data dropped)</li>
+ * </ul>
+ *
+ * @return true if data was accepted, false if dropped or queue is stopped
+ */
+ public boolean produce(final T data) {
+ if (!running) {
+ return false;
+ }
+ final int index = partitions.length == 1
+ ? 0 : partitionSelector.select(data, partitions.length);
+ if (config.getStrategy() == BufferStrategy.BLOCKING) {
+ try {
+ partitions[index].put(data);
+ return true;
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ } else {
+ return partitions[index].offer(data);
+ }
+ }
+
+ /**
+ * Core drain loop. Runs on a scheduler thread. Drains all assigned partitions
+ * into a single list, then dispatches. Loops until partitions are empty,
+ * then breaks to re-schedule with backoff.
+ *
+ * <p>The loop structure ensures that a burst of data is fully drained before
+ * the task yields. This prevents data piling up across reschedule gaps.
+ */
+ void drainLoop(final int taskIndex) {
+ final int[] myPartitions = assignedPartitions[taskIndex];
+ try {
+ while (running) {
+ // Drain all assigned partitions into one batch
+ final List<T> combined = new ArrayList<>();
+ for (final int partitionIndex : myPartitions) {
+ partitions[partitionIndex].drainTo(combined);
+ }
+
+ if (combined.isEmpty()) {
+ // Nothing to drain — increase backoff and notify idle
+ consecutiveIdleCycles[taskIndex]++;
+ notifyIdle();
+ break;
+ }
+
+ // Data found — reset backoff and dispatch
+ consecutiveIdleCycles[taskIndex] = 0;
+ dispatch(combined);
+ }
+ } catch (final Throwable t) {
+ log.error("BatchQueue[{}]: drain loop error", name, t);
+ } finally {
+ // Always re-schedule unless shutdown
+ if (running) {
+ scheduleDrain(taskIndex);
+ }
+ }
+ }
+
+ /**
+ * Dispatch a batch to the appropriate consumer(s).
+ *
+ * <p>Two modes:
+ * <ol>
+ * <li><b>Single consumer</b> (config.consumer != null): the entire batch goes
+ * to one consumer, regardless of item types.</li>
+ * <li><b>Handler map</b>: items are grouped by {@code item.getClass()}, then
+ * each group is dispatched to its registered handler. Unregistered types
+ * are silently dropped.</li>
+ * </ol>
+ */
+ private void dispatch(final List<T> batch) {
+ if (config.getConsumer() != null) {
+ try {
+ config.getConsumer().consume(batch);
+ } catch (final Throwable t) {
+ handleError(batch, t);
+ }
+ return;
+ }
+
+ // Group by concrete class, then dispatch each group to its handler
+ final Map<Class<?>, List<T>> grouped = new HashMap<>();
+ for (final T item : batch) {
+ grouped.computeIfAbsent(item.getClass(), k -> new ArrayList<>()).add(item);
+ }
+
+ for (final Map.Entry<Class<?>, List<T>> entry : grouped.entrySet()) {
+ final HandlerConsumer<T> handler = handlerMap.get(entry.getKey());
+ if (handler != null) {
+ try {
+ handler.consume(entry.getValue());
+ } catch (final Throwable t) {
+ handleError(entry.getValue(), t);
+ }
+ }
+ }
+ }
+
+ /**
+ * Notify consumer/handlers that a drain cycle found no data.
+ * Useful for flush-on-idle semantics (e.g. flush partial batches to storage).
+ */
+ private void notifyIdle() {
+ if (config.getConsumer() != null) {
+ try {
+ config.getConsumer().onIdle();
+ } catch (final Throwable t) {
+ log.error("BatchQueue[{}]: onIdle error in consumer", name, t);
+ }
+ } else {
+ for (final HandlerConsumer<T> handler : handlerMap.values()) {
+ try {
+ handler.onIdle();
+ } catch (final Throwable t) {
+ log.error("BatchQueue[{}]: onIdle error in handler", name, t);
+ }
+ }
+ }
+ }
+
+ private void handleError(final List<T> data, final Throwable t) {
+ if (config.getErrorHandler() != null) {
+ try {
+ config.getErrorHandler().onError(data, t);
+ } catch (final Throwable inner) {
+ log.error("BatchQueue[{}]: error handler threw", name, inner);
+ }
+ } else {
+ log.error("BatchQueue[{}]: unhandled dispatch error", name, t);
+ }
+ }
+
+ /**
+ * Stop the queue: reject new produces, perform a final drain of all partitions,
+ * and release the scheduler (dedicated: shutdown; shared: decrement ref count).
+ */
+ void shutdown() {
+ running = false;
+ // Final drain — flush any remaining data to consumers
+ final List<T> combined = new ArrayList<>();
+ for (final ArrayBlockingQueue<T> partition : partitions) {
+ partition.drainTo(combined);
+ }
+ if (!combined.isEmpty()) {
+ dispatch(combined);
+ }
+ if (dedicatedScheduler) {
+ scheduler.shutdown();
+ } else if (sharedSchedulerName != null) {
+ BatchQueueManager.releaseSharedScheduler(sharedSchedulerName);
+ }
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public boolean isRunning() {
+ return running;
+ }
+
+ int getPartitionCount() {
+ return partitions.length;
+ }
+
+ int getTaskCount() {
+ return assignedPartitions.length;
+ }
+
+ int[][] getAssignedPartitions() {
+ return assignedPartitions;
+ }
+
+ boolean isDedicatedScheduler() {
+ return dedicatedScheduler;
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfig.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfig.java
new file mode 100644
index 0000000000..32ca86c2bf
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfig.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import lombok.Builder;
+import lombok.Getter;
+
+@Getter
+@Builder
+public class BatchQueueConfig<T> {
+ /**
+ * Thread policy for a dedicated ScheduledExecutorService.
+ * When set, the queue creates its own scheduler.
+ * When null, sharedSchedulerName/sharedSchedulerThreads must be set.
+ */
+ private ThreadPolicy threads;
+
+ /**
+ * Shared scheduler name. Set via the builder method
+ * {@code sharedScheduler(name, threads)}.
+ */
+ private String sharedSchedulerName;
+
+ /**
+ * Thread policy for the shared scheduler. Set together with sharedSchedulerName.
+ */
+ private ThreadPolicy sharedSchedulerThreads;
+
+ @Builder.Default
+ private PartitionPolicy partitions = PartitionPolicy.fixed(1);
+
+ /**
+ * Partition selector for multi-partition queues. Determines which partition
+ * a produced item is placed into. Ignored when the queue has only 1 partition.
+ * Defaults to {@link PartitionSelector#typeHash()}.
+ */
+ @Builder.Default
+ private PartitionSelector<T> partitionSelector = PartitionSelector.typeHash();
+
+ @Builder.Default
+ private int bufferSize = 10_000;
+
+ @Builder.Default
+ private BufferStrategy strategy = BufferStrategy.BLOCKING;
+
+ /**
+ * Direct consumer for the whole batch. When set, all drained data goes to this
+ * handler without class-based grouping. Takes priority over handler map.
+ */
+ private HandlerConsumer<T> consumer;
+
+ private QueueErrorHandler<T> errorHandler;
+
+ @Builder.Default
+ private long minIdleMs = 5;
+
+ @Builder.Default
+ private long maxIdleMs = 200;
+
+ void validate() {
+ final boolean hasDedicated = threads != null;
+ final boolean hasShared = sharedSchedulerName != null;
+ if (hasDedicated == hasShared) {
+ throw new IllegalArgumentException(
+ "Exactly one of threads or sharedScheduler must be set. " +
+ "threads=" + threads + ", sharedSchedulerName=" + sharedSchedulerName);
+ }
+ if (hasShared && sharedSchedulerThreads == null) {
+ throw new IllegalArgumentException(
+ "sharedSchedulerThreads must be set when sharedSchedulerName is set");
+ }
+ if (bufferSize < 1) {
+ throw new IllegalArgumentException("bufferSize must be >= 1, got: " + bufferSize);
+ }
+ if (minIdleMs < 1) {
+ throw new IllegalArgumentException("minIdleMs must be >= 1, got: " + minIdleMs);
+ }
+ if (maxIdleMs < minIdleMs) {
+ throw new IllegalArgumentException(
+ "maxIdleMs must be >= minIdleMs, got maxIdleMs=" + maxIdleMs + " minIdleMs=" + minIdleMs);
+ }
+ }
+
+ /**
+ * Builder customization: convenience method for setting shared scheduler fields together.
+ */
+ public static class BatchQueueConfigBuilder<T> {
+ public BatchQueueConfigBuilder<T> sharedScheduler(final String name, final ThreadPolicy threads) {
+ this.sharedSchedulerName = name;
+ this.sharedSchedulerThreads = threads;
+ return this;
+ }
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManager.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManager.java
new file mode 100644
index 0000000000..061daf6dee
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManager.java
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Global registry for batch queues and shared schedulers.
+ * Thread-safe. Queues are created by name and shared across modules.
+ *
+ * <p>Shared schedulers are created lazily on first queue reference — no separate
+ * setup step needed. This eliminates startup ordering dependencies.
+ *
+ * <p>Shared schedulers are reference-counted: each queue that uses a shared scheduler
+ * increments the count on creation and decrements on shutdown. When the count reaches
+ * zero, the scheduler is shut down automatically.
+ *
+ * <p>Internal maps:
+ * <pre>
+ * QUEUES: queueName -> BatchQueue instance
+ * SHARED_SCHEDULERS: schedulerName -> ScheduledExecutorService
+ * SHARED_SCHEDULER_POLICIES: schedulerName -> ThreadPolicy (first-wins)
+ * SHARED_SCHEDULER_REF_COUNTS: schedulerName -> AtomicInteger (reference count)
+ * </pre>
+ */
+@Slf4j
+public class BatchQueueManager {
+ /**
+ * queueName -> BatchQueue instance. Each queue has a unique name.
+ */
+ private static final ConcurrentHashMap<String, BatchQueue<?>> QUEUES = new ConcurrentHashMap<>();
+ /**
+ * schedulerName -> ScheduledExecutorService. Multiple queues can share one scheduler by
+ * referencing the same scheduler name in their config.
+ */
+ private static final ConcurrentHashMap<String, ScheduledExecutorService> SHARED_SCHEDULERS =
+ new ConcurrentHashMap<>();
+ /**
+ * schedulerName -> ThreadPolicy. Tracks the first-wins policy for each shared scheduler
+ * to detect mismatched configs.
+ */
+ private static final ConcurrentHashMap<String, ThreadPolicy> SHARED_SCHEDULER_POLICIES =
+ new ConcurrentHashMap<>();
+ /**
+ * schedulerName -> reference count. Incremented when a queue acquires the scheduler,
+ * decremented when a queue releases it. Scheduler is shut down when count reaches 0.
+ */
+ private static final ConcurrentHashMap<String, AtomicInteger> SHARED_SCHEDULER_REF_COUNTS =
+ new ConcurrentHashMap<>();
+
+ /**
+ * Get or create a shared scheduler and increment its reference count.
+ * Called internally by BatchQueue constructor.
+ * First call creates the scheduler; subsequent calls reuse it.
+ * If ThreadPolicy differs from the first creator, logs a warning (first one wins).
+ */
+ static ScheduledExecutorService getOrCreateSharedScheduler(final String name,
+ final ThreadPolicy threads) {
+ SHARED_SCHEDULER_POLICIES.compute(name, (k, existing) -> {
+ if (existing != null) {
+ if (!existing.toString().equals(threads.toString())) {
+ log.warn("Shared scheduler [{}]: ThreadPolicy mismatch. "
+ + "Existing={}, requested={}. Using existing.",
+ name, existing, threads);
+ }
+ return existing;
+ }
+ return threads;
+ });
+
+ SHARED_SCHEDULER_REF_COUNTS.computeIfAbsent(name, k -> new AtomicInteger(0)).incrementAndGet();
+
+ return SHARED_SCHEDULERS.computeIfAbsent(name, k -> {
+ final int threadCount = threads.resolve();
+ log.info("Creating shared scheduler [{}] with {} threads ({})",
+ name, threadCount, threads);
+ return Executors.newScheduledThreadPool(threadCount, r -> {
+ final Thread t = new Thread(r);
+ t.setName("SharedScheduler-" + name + "-" + t.getId());
+ t.setDaemon(true);
+ return t;
+ });
+ });
+ }
+
+ /**
+ * Decrement the reference count for a shared scheduler.
+ * When the count reaches zero, the scheduler is shut down and removed.
+ */
+ static void releaseSharedScheduler(final String name) {
+ final AtomicInteger refCount = SHARED_SCHEDULER_REF_COUNTS.get(name);
+ if (refCount == null) {
+ return;
+ }
+ if (refCount.decrementAndGet() <= 0) {
+ SHARED_SCHEDULER_REF_COUNTS.remove(name);
+ SHARED_SCHEDULER_POLICIES.remove(name);
+ final ScheduledExecutorService scheduler = SHARED_SCHEDULERS.remove(name);
+ if (scheduler != null) {
+ log.info("Shutting down shared scheduler [{}] (ref count reached 0)", name);
+ scheduler.shutdown();
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> BatchQueue<T> create(final String name, final BatchQueueConfig<T> config) {
+ config.validate();
+ return (BatchQueue<T>) QUEUES.computeIfAbsent(name, k -> new BatchQueue<>(name, config));
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> BatchQueue<T> get(final String name) {
+ return (BatchQueue<T>) QUEUES.get(name);
+ }
+
+ public static void shutdown(final String name) {
+ final BatchQueue<?> queue = QUEUES.remove(name);
+ if (queue != null) {
+ queue.shutdown();
+ }
+ }
+
+ /**
+ * Shutdown all queues and all shared schedulers. Called during OAP server shutdown.
+ */
+ public static void shutdownAll() {
+ final List<BatchQueue<?>> allQueues = new ArrayList<>(QUEUES.values());
+ QUEUES.clear();
+
+ for (final BatchQueue<?> queue : allQueues) {
+ try {
+ queue.shutdown();
+ } catch (final Throwable t) {
+ log.error("Error shutting down queue: {}", queue.getName(), t);
+ }
+ }
+
+ for (final ScheduledExecutorService scheduler : SHARED_SCHEDULERS.values()) {
+ try {
+ scheduler.shutdown();
+ } catch (final Throwable t) {
+ log.error("Error shutting down shared scheduler", t);
+ }
+ }
+ SHARED_SCHEDULERS.clear();
+ SHARED_SCHEDULER_POLICIES.clear();
+ SHARED_SCHEDULER_REF_COUNTS.clear();
+ }
+
+ /**
+ * Reset for testing purposes only.
+ */
+ static void reset() {
+ shutdownAll();
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BufferStrategy.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BufferStrategy.java
new file mode 100644
index 0000000000..66a8e5df2a
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/BufferStrategy.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+public enum BufferStrategy {
+ /**
+ * Producer blocks when buffer is full (queue.put).
+ */
+ BLOCKING,
+
+ /**
+ * Producer gets false when buffer is full (queue.offer). Data is dropped.
+ */
+ IF_POSSIBLE
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/HandlerConsumer.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/HandlerConsumer.java
new file mode 100644
index 0000000000..0b9e583654
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/HandlerConsumer.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.List;
+
+/**
+ * Handler for processing a batch of data for a specific type.
+ * Each metric type (or I/O queue user) provides its own handler instance.
+ */
+public interface HandlerConsumer<T> {
+ /**
+ * Process a batch of data belonging to this handler's type.
+ */
+ void consume(List<T> data);
+
+ /**
+ * Called when there is nothing to consume. Can be used as a timer trigger
+ * (e.g. flush L1 aggregation cache periodically).
+ */
+ default void onIdle() {
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicy.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicy.java
new file mode 100644
index 0000000000..ac03e9f75d
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicy.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+/**
+ * Determines the number of partitions for a BatchQueue.
+ *
+ * Two modes:
+ * - fixed(N): exactly N partitions, regardless of thread count.
+ * - threadMultiply(N): N * resolved thread count.
+ *
+ * At queue construction time, the resolved partition count is validated:
+ * if partitions less than resolved thread count, thread count is reduced to match
+ * partitions and a warning is logged.
+ */
+public class PartitionPolicy {
+ private final int fixedCount;
+ private final int multiplier;
+
+ private PartitionPolicy(final int fixedCount, final int multiplier) {
+ this.fixedCount = fixedCount;
+ this.multiplier = multiplier;
+ }
+
+ /**
+ * Fixed number of partitions.
+ *
+ * @throws IllegalArgumentException if count < 1
+ */
+ public static PartitionPolicy fixed(final int count) {
+ if (count < 1) {
+ throw new IllegalArgumentException("Partition count must be >= 1, got: " + count);
+ }
+ return new PartitionPolicy(count, 0);
+ }
+
+ /**
+ * Partitions = multiplier * resolved thread count.
+ *
+ * @throws IllegalArgumentException if multiplier < 1
+ */
+ public static PartitionPolicy threadMultiply(final int multiplier) {
+ if (multiplier < 1) {
+ throw new IllegalArgumentException("Partition multiplier must be >= 1, got: " + multiplier);
+ }
+ return new PartitionPolicy(0, multiplier);
+ }
+
+ /**
+ * Resolve the actual partition count.
+ * For fixed mode, returns fixedCount.
+ * For threadMultiply mode, returns multiplier * resolvedThreadCount (min 1).
+ */
+ public int resolve(final int resolvedThreadCount) {
+ if (fixedCount > 0) {
+ return fixedCount;
+ }
+ return Math.max(1, multiplier * resolvedThreadCount);
+ }
+
+ @Override
+ public String toString() {
+ if (fixedCount > 0) {
+ return "fixed(" + fixedCount + ")";
+ }
+ return "threadMultiply(" + multiplier + ")";
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionSelector.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionSelector.java
new file mode 100644
index 0000000000..d30d74a05e
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionSelector.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+/**
+ * Strategy for selecting which partition a produced item should be placed into.
+ *
+ * <p>The default implementation ({@link #typeHash()}) uses the item's class hash,
+ * ensuring all items of the same type land on the same partition. This eliminates
+ * the need for HashMap grouping during dispatch and avoids contention on shared
+ * counters with many concurrent producers.
+ *
+ * <p>Only used when the queue has multiple partitions. Single-partition queues
+ * bypass the selector entirely.
+ *
+ * @param <T> the queue element type
+ */
+@FunctionalInterface
+public interface PartitionSelector<T> {
+
+ /**
+ * Select a partition index for the given data item.
+ *
+ * @param data the item being produced
+ * @param partitionCount total number of partitions (always > 1)
+ * @return a partition index in [0, partitionCount)
+ */
+ int select(T data, int partitionCount);
+
+ /**
+ * 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.
+ */
+ static <T> PartitionSelector<T> typeHash() {
+ return (data, count) -> (data.getClass().hashCode() & 0x7FFFFFFF) % count;
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/QueueErrorHandler.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/QueueErrorHandler.java
new file mode 100644
index 0000000000..c122d4c761
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/QueueErrorHandler.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.List;
+
+@FunctionalInterface
+public interface QueueErrorHandler<T> {
+ void onError(List<T> data, Throwable t);
+}
diff --git a/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicy.java b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicy.java
new file mode 100644
index 0000000000..6b6585debe
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/main/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicy.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+/**
+ * Determines the number of threads for a BatchQueue's dedicated scheduler
+ * or for a shared scheduler.
+ *
+ * Two modes:
+ * - fixed(N): exactly N threads, regardless of hardware.
+ * - cpuCores(multiplier): multiplier * Runtime.availableProcessors(), rounded.
+ *
+ * Resolved value is always >= 1 — every pool must have at least one thread.
+ * fixed() requires count >= 1 at construction. cpuCores() applies max(1, ...) at resolution.
+ */
+public class ThreadPolicy {
+ private final int fixedCount;
+ private final double cpuMultiplier;
+
+ private ThreadPolicy(final int fixedCount, final double cpuMultiplier) {
+ this.fixedCount = fixedCount;
+ this.cpuMultiplier = cpuMultiplier;
+ }
+
+ /**
+ * Fixed number of threads. Count must be >= 1.
+ *
+ * @throws IllegalArgumentException if count < 1
+ */
+ public static ThreadPolicy fixed(final int count) {
+ if (count < 1) {
+ throw new IllegalArgumentException("Thread count must be >= 1, got: " + count);
+ }
+ return new ThreadPolicy(count, 0);
+ }
+
+ /**
+ * Threads = multiplier * available CPU cores, rounded, min 1.
+ * Multiplier must be > 0.
+ *
+ * @throws IllegalArgumentException if multiplier <= 0
+ */
+ public static ThreadPolicy cpuCores(final double multiplier) {
+ if (multiplier <= 0) {
+ throw new IllegalArgumentException("CPU multiplier must be > 0, got: " + multiplier);
+ }
+ return new ThreadPolicy(0, multiplier);
+ }
+
+ /**
+ * Resolve the actual thread count. Always returns >= 1.
+ */
+ public int resolve() {
+ if (fixedCount > 0) {
+ return fixedCount;
+ }
+ return Math.max(1, (int) Math.round(cpuMultiplier * Runtime.getRuntime().availableProcessors()));
+ }
+
+ @Override
+ public String toString() {
+ if (fixedCount > 0) {
+ return "fixed(" + fixedCount + ")";
+ }
+ return "cpuCores(" + cpuMultiplier + ")";
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueBenchmark.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueBenchmark.java
new file mode 100644
index 0000000000..e4e8ceee25
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueBenchmark.java
@@ -0,0 +1,400 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Throughput benchmark for BatchQueue consuming with handler-map dispatch.
+ *
+ * <p>Measures how many metric items per second the queue can consume
+ * when there are 40 or 200 distinct metric types, each routed
+ * to its own handler via class-based dispatch.
+ *
+ * <p>Run with: mvn test -pl oap-server/server-library/library-batch-queue
+ * -Dtest=BatchQueueBenchmark -DfailIfNoTests=false
+ *
+ * <h3>Reference results (Apple M3 Max, 128 GB RAM, macOS 26.2, JDK 17)</h3>
+ * <pre>
+ * Benchmark Types Threads Partitions IF_POSSIBLE BLOCKING
+ * ---------------------------- ------ -------- --------------- ------------- -------------
+ * 40-types-1thread 40 fixed(1) fixed(1) ~33,200,000 ~32,200,000
+ * 200-types-1thread 200 fixed(1) fixed(1) ~34,600,000 ~33,900,000
+ * direct-consumer-4threads 1 fixed(4) multiply(8) ~19,200,000 ~18,800,000
+ * 40-types-4threads 40 fixed(4) multiply(8) ~31,200,000 ~44,300,000
+ * 200-types-4threads 200 fixed(4) multiply(8) ~43,900,000 ~43,400,000
+ * 200reg/50hot, 100 producers 200 fixed(4) multiply(8) ~26,900,000 ~26,100,000
+ *
+ * All runs: bufferSize=50,000, 0% drop rate.
+ * Single-partition queues (fixed(1)) bypass PartitionSelector entirely.
+ * Multi-partition queues default to PartitionSelector.typeHash() —
+ * same type always hits the same partition, so each consumer thread
+ * drains pre-grouped batches. Custom selectors can be set via config.
+ * </pre>
+ *
+ * <h3>Comparison with DataCarrier — production case (4 consumer threads)</h3>
+ * <pre>
+ * Scenario DataCarrier BatchQueue
+ * (N carriers (1 queue +
+ * + BulkPool) handler map)
+ * ------------------------------- ----------- -----------
+ * 40 types, 4 producers ~42,000,000 ~31,200,000
+ * 200 types, 4 producers ~25,100,000 ~43,900,000
+ * 200 reg / 50 hot, 100 producers ~17,600,000 ~26,900,000 (IF_POSSIBLE)
+ * 200 reg / 50 hot, 100 producers ~23,800,000 ~26,100,000 (BLOCKING)
+ *
+ * BatchQueue uses configurable PartitionSelector (default: typeHash).
+ * Same-type items land on the same partition, so dispatch() grouping is
+ * effectively a no-op. The production-realistic case (200reg/50hot/100
+ * producers) shows BatchQueue at ~27M vs DataCarrier at ~18M, a 1.5x
+ * advantage.
+ * </pre>
+ */
+@Slf4j
+@SuppressWarnings("all")
+public class BatchQueueBenchmark {
+
+ private static final int WARMUP_SECONDS = 2;
+ private static final int MEASURE_SECONDS = 5;
+ private static final int PRODUCER_THREADS = 4;
+
+ @AfterEach
+ public void cleanup() {
+ BatchQueueManager.reset();
+ }
+
+ // ---- IF_POSSIBLE (non-blocking) benchmarks ----
+
+ @Test
+ public void benchmarkHandlerMapDispatch40Types() throws Exception {
+ runBenchmark("40-types-4t-ifpossible", 40, ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch200Types() throws Exception {
+ runBenchmark("200-types-4t-ifpossible", 200, ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkDirectConsumer() throws Exception {
+ runDirectBenchmark("direct-4t-ifpossible", ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch40TypesSingleThread() throws Exception {
+ runBenchmark("40-types-1t-ifpossible", 40, ThreadPolicy.fixed(1),
+ PartitionPolicy.fixed(1), BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch200TypesSingleThread() throws Exception {
+ runBenchmark("200-types-1t-ifpossible", 200, ThreadPolicy.fixed(1),
+ PartitionPolicy.fixed(1), BufferStrategy.IF_POSSIBLE);
+ }
+
+ // ---- BLOCKING benchmarks ----
+
+ @Test
+ public void benchmarkHandlerMapDispatch40TypesBlocking() throws Exception {
+ runBenchmark("40-types-4t-blocking", 40, ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch200TypesBlocking() throws Exception {
+ runBenchmark("200-types-4t-blocking", 200, ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmarkDirectConsumerBlocking() throws Exception {
+ runDirectBenchmark("direct-4t-blocking", ThreadPolicy.fixed(4),
+ PartitionPolicy.threadMultiply(8), BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch40TypesSingleThreadBlocking() throws Exception {
+ runBenchmark("40-types-1t-blocking", 40, ThreadPolicy.fixed(1),
+ PartitionPolicy.fixed(1), BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmarkHandlerMapDispatch200TypesSingleThreadBlocking() throws Exception {
+ runBenchmark("200-types-1t-blocking", 200, ThreadPolicy.fixed(1),
+ PartitionPolicy.fixed(1), BufferStrategy.BLOCKING);
+ }
+
+ // ---- Production-realistic: 200 types, 50 hot, 100 gRPC producers ----
+
+ @Test
+ public void benchmarkHotTypes200x50x100producers() throws Exception {
+ runHotTypesBenchmark("200reg-50hot-100p-ifpossible", 200, 50, 100,
+ ThreadPolicy.fixed(4), PartitionPolicy.threadMultiply(8),
+ BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkHotTypes200x50x100producersBlocking() throws Exception {
+ runHotTypesBenchmark("200reg-50hot-100p-blocking", 200, 50, 100,
+ ThreadPolicy.fixed(4), PartitionPolicy.threadMultiply(8),
+ BufferStrategy.BLOCKING);
+ }
+
+ private void runHotTypesBenchmark(final String label, final int totalTypes,
+ final int hotTypes, final int producerCount,
+ final ThreadPolicy threads,
+ final PartitionPolicy partitions,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final BatchQueue<BenchmarkMetricTypes.TypedMetric> queue = BatchQueueManager.create(
+ "bench-" + label,
+ BatchQueueConfig.<BenchmarkMetricTypes.TypedMetric>builder()
+ .threads(threads)
+ .partitions(partitions)
+ .bufferSize(50_000)
+ .strategy(strategy)
+ .minIdleMs(1)
+ .maxIdleMs(50)
+ .build());
+
+ // Register handlers for ALL 200 types
+ for (int t = 0; t < totalTypes; t++) {
+ queue.addHandler(BenchmarkMetricTypes.CLASSES[t],
+ (HandlerConsumer<BenchmarkMetricTypes.TypedMetric>) data ->
+ consumed.addAndGet(data.size()));
+ }
+
+ // Warmup — only hot types get traffic
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducersHot(queue, hotTypes, producerCount, warmupEnd);
+ Thread.sleep(200);
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducersHot(queue, hotTypes, producerCount, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ log.info("\n=== BatchQueue Benchmark: {} ===\n"
+ + " Total types: {} (handlers registered)\n"
+ + " Hot types: {} (receiving traffic)\n"
+ + " Threads: {}\n"
+ + " Partitions: {}\n"
+ + " Producers: {} (simulating gRPC connections)\n"
+ + " Duration: {} ms\n"
+ + " Produced: {}\n"
+ + " Consumed: {}\n"
+ + " Produce rate: {} items/sec\n"
+ + " Consume rate: {} items/sec\n"
+ + " Drop rate: {}%\n",
+ label, totalTypes, hotTypes, threads, partitions, producerCount,
+ measureDuration,
+ String.format("%,d", produced), String.format("%,d", totalConsumed),
+ String.format("%,.0f", produced * 1000.0 / measureDuration),
+ String.format("%,.0f", totalConsumed * 1000.0 / measureDuration),
+ String.format("%.2f", produced > 0
+ ? (produced - totalConsumed) * 100.0 / produced : 0));
+ }
+
+ private long runProducersHot(final BatchQueue<BenchmarkMetricTypes.TypedMetric> queue,
+ final int hotTypes, final int producerCount,
+ final long endTimeMs) throws InterruptedException {
+ final AtomicLong totalProduced = new AtomicLong(0);
+ final CountDownLatch done = new CountDownLatch(producerCount);
+
+ for (int p = 0; p < producerCount; p++) {
+ final int producerIndex = p;
+ final Thread thread = new Thread(() -> {
+ long count = 0;
+ int typeIndex = producerIndex;
+ while (System.currentTimeMillis() < endTimeMs) {
+ for (int batch = 0; batch < 100; batch++) {
+ final int type = typeIndex % hotTypes;
+ if (queue.produce(BenchmarkMetricTypes.FACTORIES[type].create(count))) {
+ count++;
+ }
+ typeIndex++;
+ }
+ }
+ totalProduced.addAndGet(count);
+ done.countDown();
+ });
+ thread.setName("gRPC-" + producerIndex);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ done.await(MEASURE_SECONDS + 10, TimeUnit.SECONDS);
+ return totalProduced.get();
+ }
+
+ private void runBenchmark(final String label, final int typeCount,
+ final ThreadPolicy threads,
+ final PartitionPolicy partitions,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final BatchQueue<BenchmarkMetricTypes.TypedMetric> queue = BatchQueueManager.create(
+ "bench-" + label,
+ BatchQueueConfig.<BenchmarkMetricTypes.TypedMetric>builder()
+ .threads(threads)
+ .partitions(partitions)
+ .bufferSize(50_000)
+ .strategy(strategy)
+ .minIdleMs(1)
+ .maxIdleMs(50)
+ .build());
+
+ for (int t = 0; t < typeCount; t++) {
+ queue.addHandler(BenchmarkMetricTypes.CLASSES[t],
+ (HandlerConsumer<BenchmarkMetricTypes.TypedMetric>) data ->
+ consumed.addAndGet(data.size()));
+ }
+
+ // Warmup
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducers(queue, typeCount, warmupEnd);
+ Thread.sleep(200);
+ final long warmupConsumed = consumed.get();
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducers(queue, typeCount, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ printResults(label, typeCount, threads, partitions,
+ produced, totalConsumed, measureDuration, warmupConsumed);
+ }
+
+ private void runDirectBenchmark(final String label,
+ final ThreadPolicy threads,
+ final PartitionPolicy partitions,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final BatchQueue<BenchmarkMetricTypes.TypedMetric> queue = BatchQueueManager.create(
+ "bench-" + label,
+ BatchQueueConfig.<BenchmarkMetricTypes.TypedMetric>builder()
+ .threads(threads)
+ .partitions(partitions)
+ .bufferSize(50_000)
+ .strategy(strategy)
+ .consumer(data -> consumed.addAndGet(data.size()))
+ .minIdleMs(1)
+ .maxIdleMs(50)
+ .build());
+
+ // Warmup
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducers(queue, 1, warmupEnd);
+ Thread.sleep(200);
+ final long warmupConsumed = consumed.get();
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducers(queue, 1, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ printResults(label, 1, threads, partitions,
+ produced, totalConsumed, measureDuration, warmupConsumed);
+ }
+
+ private long runProducers(final BatchQueue<BenchmarkMetricTypes.TypedMetric> queue,
+ final int typeCount,
+ final long endTimeMs) throws InterruptedException {
+ final AtomicLong totalProduced = new AtomicLong(0);
+ final CountDownLatch done = new CountDownLatch(PRODUCER_THREADS);
+
+ for (int p = 0; p < PRODUCER_THREADS; p++) {
+ final int producerIndex = p;
+ final Thread thread = new Thread(() -> {
+ long count = 0;
+ int typeIndex = producerIndex;
+ while (System.currentTimeMillis() < endTimeMs) {
+ for (int batch = 0; batch < 100; batch++) {
+ final int type = typeIndex % typeCount;
+ if (queue.produce(BenchmarkMetricTypes.FACTORIES[type].create(count))) {
+ count++;
+ }
+ typeIndex++;
+ }
+ }
+ totalProduced.addAndGet(count);
+ done.countDown();
+ });
+ thread.setName("Producer-" + producerIndex);
+ thread.start();
+ }
+
+ done.await(MEASURE_SECONDS + 10, TimeUnit.SECONDS);
+ return totalProduced.get();
+ }
+
+ private void printResults(final String label, final int typeCount,
+ final ThreadPolicy threads,
+ final PartitionPolicy partitions,
+ final long produced, final long consumed,
+ final long durationMs, final long warmupConsumed) {
+ final double producedPerSec = produced * 1000.0 / durationMs;
+ final double consumedPerSec = consumed * 1000.0 / durationMs;
+ final double dropRate = produced > 0 ? (produced - consumed) * 100.0 / produced : 0;
+
+ log.info("\n=== BatchQueue Benchmark: {} ===\n"
+ + " Types: {}\n"
+ + " Threads: {}\n"
+ + " Partitions: {}\n"
+ + " Producers: {}\n"
+ + " Duration: {} ms\n"
+ + " Warmup consumed: {}\n"
+ + " Produced: {}\n"
+ + " Consumed: {}\n"
+ + " Produce rate: {} items/sec\n"
+ + " Consume rate: {} items/sec\n"
+ + " Drop rate: {}%\n",
+ label, typeCount, threads, partitions, PRODUCER_THREADS,
+ durationMs, String.format("%,d", warmupConsumed),
+ String.format("%,d", produced), String.format("%,d", consumed),
+ String.format("%,.0f", producedPerSec),
+ String.format("%,.0f", consumedPerSec),
+ String.format("%.2f", dropRate));
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfigTest.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfigTest.java
new file mode 100644
index 0000000000..0717dae4fe
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueConfigTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class BatchQueueConfigTest {
+
+ @Test
+ public void testDedicatedThreadConfig() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .build();
+ config.validate();
+ assertNotNull(config.getThreads());
+ }
+
+ @Test
+ public void testSharedSchedulerConfig() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .sharedScheduler("IO_POOL", ThreadPolicy.fixed(4))
+ .build();
+ config.validate();
+ assertEquals("IO_POOL", config.getSharedSchedulerName());
+ assertNotNull(config.getSharedSchedulerThreads());
+ }
+
+ @Test
+ public void testRejectsBothThreadsAndShared() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .sharedScheduler("IO_POOL", ThreadPolicy.fixed(4))
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testRejectsNeitherThreadsNorShared() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testRejectsSharedWithoutThreadPolicy() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .sharedSchedulerName("IO_POOL")
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testRejectsZeroBufferSize() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .bufferSize(0)
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testRejectsZeroMinIdleMs() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .minIdleMs(0)
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testRejectsMaxIdleLessThanMinIdle() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .minIdleMs(100)
+ .maxIdleMs(50)
+ .build();
+ assertThrows(IllegalArgumentException.class, config::validate);
+ }
+
+ @Test
+ public void testDefaults() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .build();
+ assertEquals(10_000, config.getBufferSize());
+ assertEquals(BufferStrategy.BLOCKING, config.getStrategy());
+ assertEquals(5, config.getMinIdleMs());
+ assertEquals(200, config.getMaxIdleMs());
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManagerTest.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManagerTest.java
new file mode 100644
index 0000000000..17b6226a37
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueManagerTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BatchQueueManagerTest {
+
+ @AfterEach
+ public void cleanup() {
+ BatchQueueManager.reset();
+ }
+
+ @Test
+ public void testCreateAndGet() {
+ final BatchQueue<String> queue = BatchQueueManager.create("test-queue",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertNotNull(queue);
+ assertEquals("test-queue", queue.getName());
+
+ final BatchQueue<String> retrieved = BatchQueueManager.get("test-queue");
+ assertSame(queue, retrieved);
+ }
+
+ @Test
+ public void testCreateReturnsExistingIfPresent() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build();
+
+ final BatchQueue<String> first = BatchQueueManager.create("absent-test", config);
+ final BatchQueue<String> second = BatchQueueManager.create("absent-test", config);
+ assertSame(first, second);
+ }
+
+ @Test
+ public void testGetNonExistentReturnsNull() {
+ assertNull(BatchQueueManager.get("nonexistent"));
+ }
+
+ @Test
+ public void testShutdownRemovesQueue() {
+ BatchQueueManager.create("shutdown-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertNotNull(BatchQueueManager.get("shutdown-test"));
+ BatchQueueManager.shutdown("shutdown-test");
+ assertNull(BatchQueueManager.get("shutdown-test"));
+ }
+
+ @Test
+ public void testShutdownAllClearsEverything() {
+ BatchQueueManager.create("q1",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ BatchQueueManager.create("q2",
+ BatchQueueConfig.<String>builder()
+ .sharedScheduler("SHARED", ThreadPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ BatchQueueManager.shutdownAll();
+
+ assertNull(BatchQueueManager.get("q1"));
+ assertNull(BatchQueueManager.get("q2"));
+ }
+
+ @Test
+ public void testSharedSchedulerCreatedLazily() {
+ // First queue referencing shared scheduler creates it
+ final BatchQueue<String> q1 = BatchQueueManager.create("lazy1",
+ BatchQueueConfig.<String>builder()
+ .sharedScheduler("LAZY_POOL", ThreadPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ // Second queue uses the same pool
+ final BatchQueue<String> q2 = BatchQueueManager.create("lazy2",
+ BatchQueueConfig.<String>builder()
+ .sharedScheduler("LAZY_POOL", ThreadPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertNotNull(q1);
+ assertNotNull(q2);
+ // Both should be running on shared scheduler (not dedicated)
+ assertNotNull(BatchQueueManager.get("lazy1"));
+ assertNotNull(BatchQueueManager.get("lazy2"));
+ }
+
+ @Test
+ public void testSharedSchedulerRefCounting() {
+ final BatchQueueConfig<String> config = BatchQueueConfig.<String>builder()
+ .sharedScheduler("REF_POOL", ThreadPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build();
+
+ BatchQueueManager.create("ref1", config);
+ BatchQueueManager.create("ref2", config);
+
+ // Capture the scheduler before any shutdown
+ final ScheduledExecutorService scheduler =
+ BatchQueueManager.getOrCreateSharedScheduler("REF_POOL", ThreadPolicy.fixed(2));
+ // Release the extra ref from getOrCreateSharedScheduler call above
+ BatchQueueManager.releaseSharedScheduler("REF_POOL");
+
+ assertFalse(scheduler.isShutdown());
+
+ // Shutting down first queue should NOT shut down the shared scheduler
+ BatchQueueManager.shutdown("ref1");
+ assertFalse(scheduler.isShutdown());
+
+ // Shutting down last queue SHOULD shut down the shared scheduler
+ BatchQueueManager.shutdown("ref2");
+ assertTrue(scheduler.isShutdown());
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueTest.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueTest.java
new file mode 100644
index 0000000000..4f95544e40
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BatchQueueTest.java
@@ -0,0 +1,573 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BatchQueueTest {
+
+ @AfterEach
+ public void cleanup() {
+ BatchQueueManager.reset();
+ }
+
+ // --- Direct consumer mode ---
+
+ @Test
+ public void testDirectConsumerReceivesAllData() {
+ final List<String> received = new CopyOnWriteArrayList<>();
+ final BatchQueue<String> queue = BatchQueueManager.create("direct-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> received.addAll(data))
+ .bufferSize(1000)
+ .build());
+
+ for (int i = 0; i < 100; i++) {
+ queue.produce("item-" + i);
+ }
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> received.size() == 100);
+
+ assertEquals(100, received.size());
+ }
+
+ @Test
+ public void testDirectConsumerWithMultipleThreads() {
+ final List<String> received = Collections.synchronizedList(new ArrayList<>());
+ final BatchQueue<String> queue = BatchQueueManager.create("multi-thread-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(4))
+ .partitions(PartitionPolicy.fixed(4))
+ .consumer(data -> received.addAll(data))
+ .bufferSize(1000)
+ .build());
+
+ for (int i = 0; i < 1000; i++) {
+ queue.produce("item-" + i);
+ }
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> received.size() == 1000);
+
+ assertEquals(1000, received.size());
+ }
+
+ // --- Handler map dispatch ---
+
+ static class MetricA {
+ final String name;
+
+ MetricA(final String name) {
+ this.name = name;
+ }
+ }
+
+ static class MetricB {
+ final String name;
+
+ MetricB(final String name) {
+ this.name = name;
+ }
+ }
+
+ @Test
+ public void testHandlerMapDispatch() {
+ final List<Object> receivedA = new CopyOnWriteArrayList<>();
+ final List<Object> receivedB = new CopyOnWriteArrayList<>();
+
+ final BatchQueue<Object> queue = BatchQueueManager.create("handler-map-test",
+ BatchQueueConfig.<Object>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .bufferSize(1000)
+ .build());
+
+ queue.addHandler(MetricA.class, data -> receivedA.addAll(data));
+ queue.addHandler(MetricB.class, data -> receivedB.addAll(data));
+
+ for (int i = 0; i < 50; i++) {
+ queue.produce(new MetricA("a-" + i));
+ queue.produce(new MetricB("b-" + i));
+ }
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> receivedA.size() == 50 && receivedB.size() == 50);
+
+ assertEquals(50, receivedA.size());
+ assertEquals(50, receivedB.size());
+ }
+
+ @Test
+ public void testHandlerMapWithManyTypes() {
+ // Simulate the 40-type scenario
+ final int typeCount = 40;
+ final int itemsPerType = 100;
+ final AtomicInteger totalReceived = new AtomicInteger(0);
+
+ final BatchQueue<Object> queue = BatchQueueManager.create("many-types-test",
+ BatchQueueConfig.<Object>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .partitions(PartitionPolicy.threadMultiply(2))
+ .bufferSize(10_000)
+ .build());
+
+ // Register one handler per "type" using MetricA as representative
+ // In real use, each type would be a different class.
+ // Here we use a direct consumer to simplify.
+ // We'll test the actual handler-map routing with fewer types above.
+ // This test validates throughput with many items.
+ final List<Object> received = new CopyOnWriteArrayList<>();
+ queue.addHandler(MetricA.class, data -> {
+ totalReceived.addAndGet(data.size());
+ received.addAll(data);
+ });
+
+ for (int i = 0; i < typeCount * itemsPerType; i++) {
+ queue.produce(new MetricA("item-" + i));
+ }
+
+ Awaitility.await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> totalReceived.get() == typeCount * itemsPerType);
+
+ assertEquals(typeCount * itemsPerType, totalReceived.get());
+ }
+
+ // --- Partition assignment ---
+
+ @Test
+ public void testPartitionCountMatchesPolicy() {
+ final BatchQueue<String> queue = BatchQueueManager.create("partition-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .partitions(PartitionPolicy.fixed(8))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertEquals(8, queue.getPartitionCount());
+ assertEquals(2, queue.getTaskCount());
+ }
+
+ @Test
+ public void testThreadMultiplyPartitions() {
+ final BatchQueue<String> queue = BatchQueueManager.create("multiply-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(3))
+ .partitions(PartitionPolicy.threadMultiply(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertEquals(6, queue.getPartitionCount());
+ assertEquals(3, queue.getTaskCount());
+ }
+
+ @Test
+ public void testPartitionsLessThanThreadsReducesThreads() {
+ // 2 partitions but 4 threads → should reduce to 2 tasks
+ final BatchQueue<String> queue = BatchQueueManager.create("reduce-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(4))
+ .partitions(PartitionPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertEquals(2, queue.getPartitionCount());
+ assertEquals(2, queue.getTaskCount());
+ }
+
+ @Test
+ public void testEachPartitionAssignedToExactlyOneTask() {
+ final BatchQueue<String> queue = BatchQueueManager.create("assign-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(3))
+ .partitions(PartitionPolicy.fixed(9))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ final int[][] assignments = queue.getAssignedPartitions();
+ assertEquals(3, assignments.length);
+
+ // Collect all assigned partitions
+ final boolean[] assigned = new boolean[9];
+ for (final int[] taskPartitions : assignments) {
+ for (final int p : taskPartitions) {
+ assertFalse(assigned[p], "Partition " + p + " assigned twice");
+ assigned[p] = true;
+ }
+ }
+ for (int i = 0; i < 9; i++) {
+ assertTrue(assigned[i], "Partition " + i + " not assigned");
+ }
+ }
+
+ // --- Shared scheduler ---
+
+ @Test
+ public void testSharedSchedulerQueuesSharePool() {
+ final List<String> received1 = new CopyOnWriteArrayList<>();
+ final List<String> received2 = new CopyOnWriteArrayList<>();
+
+ final BatchQueue<String> q1 = BatchQueueManager.create("shared-q1",
+ BatchQueueConfig.<String>builder()
+ .sharedScheduler("TEST_POOL", ThreadPolicy.fixed(2))
+ .consumer(data -> received1.addAll(data))
+ .bufferSize(1000)
+ .build());
+
+ final BatchQueue<String> q2 = BatchQueueManager.create("shared-q2",
+ BatchQueueConfig.<String>builder()
+ .sharedScheduler("TEST_POOL", ThreadPolicy.fixed(2))
+ .consumer(data -> received2.addAll(data))
+ .bufferSize(1000)
+ .build());
+
+ assertFalse(q1.isDedicatedScheduler());
+ assertFalse(q2.isDedicatedScheduler());
+
+ for (int i = 0; i < 50; i++) {
+ q1.produce("a-" + i);
+ q2.produce("b-" + i);
+ }
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> received1.size() == 50 && received2.size() == 50);
+ }
+
+ @Test
+ public void testDedicatedSchedulerIsOwned() {
+ final BatchQueue<String> queue = BatchQueueManager.create("dedicated-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertTrue(queue.isDedicatedScheduler());
+ }
+
+ // --- Produce and buffer strategy ---
+
+ @Test
+ public void testIfPossibleDropsWhenFull() {
+ final CountDownLatch blockLatch = new CountDownLatch(1);
+ final BatchQueue<String> queue = BatchQueueManager.create("drop-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .strategy(BufferStrategy.IF_POSSIBLE)
+ .consumer(data -> {
+ try {
+ blockLatch.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ })
+ .bufferSize(5)
+ .build());
+
+ // Produce enough to fill the buffer while consumer is blocked
+ // First batch will be drained and block the consumer.
+ // Wait for it to be picked up, then fill the buffer.
+ Awaitility.await().atMost(2, TimeUnit.SECONDS)
+ .pollInterval(10, TimeUnit.MILLISECONDS)
+ .until(() -> {
+ // Keep producing — once consumer is blocked, buffer fills up
+ queue.produce("x");
+ return !queue.produce("overflow");
+ });
+
+ // At least one produce should have returned false (dropped)
+ blockLatch.countDown();
+ }
+
+ @Test
+ public void testProduceReturnsFalseWhenStopped() {
+ final BatchQueue<String> queue = BatchQueueManager.create("stopped-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> { })
+ .bufferSize(100)
+ .build());
+
+ assertTrue(queue.produce("before-shutdown"));
+ queue.shutdown();
+ assertFalse(queue.produce("after-shutdown"));
+ }
+
+ // --- Shutdown final drain ---
+
+ @Test
+ public void testShutdownDrainsRemainingData() throws Exception {
+ final List<String> received = new CopyOnWriteArrayList<>();
+ final CountDownLatch blockLatch = new CountDownLatch(1);
+ final AtomicInteger consumeCalls = new AtomicInteger(0);
+
+ final BatchQueue<String> queue = BatchQueueManager.create("drain-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> {
+ if (consumeCalls.getAndIncrement() == 0) {
+ // Block on first consume to let items accumulate
+ try {
+ blockLatch.await(2, TimeUnit.SECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ received.addAll(data);
+ })
+ .bufferSize(10_000)
+ .strategy(BufferStrategy.IF_POSSIBLE)
+ .build());
+
+ // Produce items
+ int produced = 0;
+ for (int i = 0; i < 500; i++) {
+ if (queue.produce("item-" + i)) {
+ produced++;
+ }
+ }
+
+ // Unblock consumer and immediately shutdown — should drain remaining
+ blockLatch.countDown();
+ Thread.sleep(50); // Let the drain loop pick up some
+ queue.shutdown();
+
+ // After shutdown, all produced items should be consumed
+ assertEquals(produced, received.size());
+ }
+
+ // --- Idle callback ---
+
+ @Test
+ public void testOnIdleCalledWhenEmpty() {
+ final AtomicInteger idleCalls = new AtomicInteger(0);
+
+ BatchQueueManager.create("idle-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(new HandlerConsumer<String>() {
+ @Override
+ public void consume(final List<String> data) {
+ // no-op
+ }
+
+ @Override
+ public void onIdle() {
+ idleCalls.incrementAndGet();
+ }
+ })
+ .bufferSize(100)
+ .build());
+
+ // Queue starts with no data, so idle should be called quickly
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .until(() -> idleCalls.get() > 0);
+
+ assertTrue(idleCalls.get() > 0);
+ }
+
+ @Test
+ public void testHandlerMapOnIdleCalled() {
+ final AtomicInteger idleA = new AtomicInteger(0);
+ final AtomicInteger idleB = new AtomicInteger(0);
+
+ final BatchQueue<Object> queue = BatchQueueManager.create("handler-idle-test",
+ BatchQueueConfig.<Object>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .bufferSize(100)
+ .build());
+
+ queue.addHandler(MetricA.class, new HandlerConsumer<Object>() {
+ @Override
+ public void consume(final List<Object> data) {
+ }
+
+ @Override
+ public void onIdle() {
+ idleA.incrementAndGet();
+ }
+ });
+ queue.addHandler(MetricB.class, new HandlerConsumer<Object>() {
+ @Override
+ public void consume(final List<Object> data) {
+ }
+
+ @Override
+ public void onIdle() {
+ idleB.incrementAndGet();
+ }
+ });
+
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .until(() -> idleA.get() > 0 && idleB.get() > 0);
+ }
+
+ // --- Error handler ---
+
+ @Test
+ public void testErrorHandlerCalled() {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+
+ final BatchQueue<String> queue = BatchQueueManager.create("error-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .consumer(data -> {
+ throw new RuntimeException("intentional");
+ })
+ .errorHandler((data, t) -> errorCount.incrementAndGet())
+ .bufferSize(100)
+ .build());
+
+ queue.produce("trigger-error");
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> errorCount.get() > 0);
+ }
+
+ // --- Adaptive backoff ---
+
+ @Test
+ public void testAdaptiveBackoffIncreasesDelay() throws Exception {
+ // When queue is idle, consecutive idle cycles increase.
+ // We verify by checking that with no data, the queue
+ // doesn't spin rapidly — idle callbacks should slow down.
+ final List<Long> idleTimestamps = new CopyOnWriteArrayList<>();
+
+ BatchQueueManager.create("backoff-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .minIdleMs(10)
+ .maxIdleMs(500)
+ .consumer(new HandlerConsumer<String>() {
+ @Override
+ public void consume(final List<String> data) {
+ }
+
+ @Override
+ public void onIdle() {
+ idleTimestamps.add(System.currentTimeMillis());
+ }
+ })
+ .bufferSize(100)
+ .build());
+
+ // Wait for several idle cycles
+ Awaitility.await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> idleTimestamps.size() >= 8);
+
+ // Verify that intervals increase (later gaps should be larger than early gaps)
+ final long earlyGap = idleTimestamps.get(1) - idleTimestamps.get(0);
+ final long laterGap = idleTimestamps.get(idleTimestamps.size() - 1)
+ - idleTimestamps.get(idleTimestamps.size() - 2);
+
+ assertTrue(laterGap > earlyGap,
+ "Later gap (" + laterGap + "ms) should be larger than early gap (" + earlyGap + "ms)");
+ }
+
+ @Test
+ public void testBackoffResetsOnData() throws Exception {
+ final AtomicInteger consumeCount = new AtomicInteger(0);
+ final List<Long> idleTimestamps = new CopyOnWriteArrayList<>();
+
+ final BatchQueue<String> queue = BatchQueueManager.create("backoff-reset-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(1))
+ .minIdleMs(10)
+ .maxIdleMs(2000)
+ .consumer(new HandlerConsumer<String>() {
+ @Override
+ public void consume(final List<String> data) {
+ consumeCount.addAndGet(data.size());
+ }
+
+ @Override
+ public void onIdle() {
+ idleTimestamps.add(System.currentTimeMillis());
+ }
+ })
+ .bufferSize(100)
+ .build());
+
+ // Let backoff build up
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> idleTimestamps.size() >= 5);
+
+ // Now send data — this should reset the backoff
+ final int beforeIdles = idleTimestamps.size();
+ for (int i = 0; i < 10; i++) {
+ queue.produce("reset-" + i);
+ }
+
+ // Wait for the data to be consumed
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> consumeCount.get() >= 10);
+
+ // After data is consumed, next idle should come quickly (backoff reset)
+ final long postDataTime = System.currentTimeMillis();
+ Awaitility.await().atMost(3, TimeUnit.SECONDS)
+ .until(() -> {
+ for (int i = beforeIdles; i < idleTimestamps.size(); i++) {
+ if (idleTimestamps.get(i) > postDataTime) {
+ return true;
+ }
+ }
+ return false;
+ });
+ }
+
+ // --- Round robin produce ---
+
+ @Test
+ public void testRoundRobinDistributesAcrossPartitions() {
+ final List<String> received = new CopyOnWriteArrayList<>();
+ final BatchQueue<String> queue = BatchQueueManager.create("roundrobin-test",
+ BatchQueueConfig.<String>builder()
+ .threads(ThreadPolicy.fixed(2))
+ .partitions(PartitionPolicy.fixed(4))
+ .consumer(data -> received.addAll(data))
+ .bufferSize(1000)
+ .build());
+
+ for (int i = 0; i < 400; i++) {
+ queue.produce("item-" + i);
+ }
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() -> received.size() == 400);
+
+ assertEquals(400, received.size());
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BenchmarkMetricTypes.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BenchmarkMetricTypes.java
new file mode 100644
index 0000000000..b055c36f0a
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/BenchmarkMetricTypes.java
@@ -0,0 +1,649 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+/**
+ * 200 distinct metric subclasses for benchmark handler-map dispatch testing.
+ * Each class represents a distinct metric type that gets routed to its own handler.
+ */
+@SuppressWarnings("all")
+class BenchmarkMetricTypes {
+
+ static class TypedMetric {
+ final int typeId;
+ final long value;
+
+ TypedMetric(final int typeId, final long value) {
+ this.typeId = typeId;
+ this.value = value;
+ }
+ }
+
+ @FunctionalInterface
+ interface MetricFactory {
+ TypedMetric create(long value);
+ }
+
+ static final int MAX_TYPES = 200;
+ static final Class<? extends TypedMetric>[] CLASSES = new Class[MAX_TYPES];
+ static final MetricFactory[] FACTORIES = new MetricFactory[MAX_TYPES];
+
+ static {
+ CLASSES[0] = M0.class; FACTORIES[0] = v -> new M0(v);
+ CLASSES[1] = M1.class; FACTORIES[1] = v -> new M1(v);
+ CLASSES[2] = M2.class; FACTORIES[2] = v -> new M2(v);
+ CLASSES[3] = M3.class; FACTORIES[3] = v -> new M3(v);
+ CLASSES[4] = M4.class; FACTORIES[4] = v -> new M4(v);
+ CLASSES[5] = M5.class; FACTORIES[5] = v -> new M5(v);
+ CLASSES[6] = M6.class; FACTORIES[6] = v -> new M6(v);
+ CLASSES[7] = M7.class; FACTORIES[7] = v -> new M7(v);
+ CLASSES[8] = M8.class; FACTORIES[8] = v -> new M8(v);
+ CLASSES[9] = M9.class; FACTORIES[9] = v -> new M9(v);
+ CLASSES[10] = M10.class; FACTORIES[10] = v -> new M10(v);
+ CLASSES[11] = M11.class; FACTORIES[11] = v -> new M11(v);
+ CLASSES[12] = M12.class; FACTORIES[12] = v -> new M12(v);
+ CLASSES[13] = M13.class; FACTORIES[13] = v -> new M13(v);
+ CLASSES[14] = M14.class; FACTORIES[14] = v -> new M14(v);
+ CLASSES[15] = M15.class; FACTORIES[15] = v -> new M15(v);
+ CLASSES[16] = M16.class; FACTORIES[16] = v -> new M16(v);
+ CLASSES[17] = M17.class; FACTORIES[17] = v -> new M17(v);
+ CLASSES[18] = M18.class; FACTORIES[18] = v -> new M18(v);
+ CLASSES[19] = M19.class; FACTORIES[19] = v -> new M19(v);
+ CLASSES[20] = M20.class; FACTORIES[20] = v -> new M20(v);
+ CLASSES[21] = M21.class; FACTORIES[21] = v -> new M21(v);
+ CLASSES[22] = M22.class; FACTORIES[22] = v -> new M22(v);
+ CLASSES[23] = M23.class; FACTORIES[23] = v -> new M23(v);
+ CLASSES[24] = M24.class; FACTORIES[24] = v -> new M24(v);
+ CLASSES[25] = M25.class; FACTORIES[25] = v -> new M25(v);
+ CLASSES[26] = M26.class; FACTORIES[26] = v -> new M26(v);
+ CLASSES[27] = M27.class; FACTORIES[27] = v -> new M27(v);
+ CLASSES[28] = M28.class; FACTORIES[28] = v -> new M28(v);
+ CLASSES[29] = M29.class; FACTORIES[29] = v -> new M29(v);
+ CLASSES[30] = M30.class; FACTORIES[30] = v -> new M30(v);
+ CLASSES[31] = M31.class; FACTORIES[31] = v -> new M31(v);
+ CLASSES[32] = M32.class; FACTORIES[32] = v -> new M32(v);
+ CLASSES[33] = M33.class; FACTORIES[33] = v -> new M33(v);
+ CLASSES[34] = M34.class; FACTORIES[34] = v -> new M34(v);
+ CLASSES[35] = M35.class; FACTORIES[35] = v -> new M35(v);
+ CLASSES[36] = M36.class; FACTORIES[36] = v -> new M36(v);
+ CLASSES[37] = M37.class; FACTORIES[37] = v -> new M37(v);
+ CLASSES[38] = M38.class; FACTORIES[38] = v -> new M38(v);
+ CLASSES[39] = M39.class; FACTORIES[39] = v -> new M39(v);
+ CLASSES[40] = M40.class; FACTORIES[40] = v -> new M40(v);
+ CLASSES[41] = M41.class; FACTORIES[41] = v -> new M41(v);
+ CLASSES[42] = M42.class; FACTORIES[42] = v -> new M42(v);
+ CLASSES[43] = M43.class; FACTORIES[43] = v -> new M43(v);
+ CLASSES[44] = M44.class; FACTORIES[44] = v -> new M44(v);
+ CLASSES[45] = M45.class; FACTORIES[45] = v -> new M45(v);
+ CLASSES[46] = M46.class; FACTORIES[46] = v -> new M46(v);
+ CLASSES[47] = M47.class; FACTORIES[47] = v -> new M47(v);
+ CLASSES[48] = M48.class; FACTORIES[48] = v -> new M48(v);
+ CLASSES[49] = M49.class; FACTORIES[49] = v -> new M49(v);
+ CLASSES[50] = M50.class; FACTORIES[50] = v -> new M50(v);
+ CLASSES[51] = M51.class; FACTORIES[51] = v -> new M51(v);
+ CLASSES[52] = M52.class; FACTORIES[52] = v -> new M52(v);
+ CLASSES[53] = M53.class; FACTORIES[53] = v -> new M53(v);
+ CLASSES[54] = M54.class; FACTORIES[54] = v -> new M54(v);
+ CLASSES[55] = M55.class; FACTORIES[55] = v -> new M55(v);
+ CLASSES[56] = M56.class; FACTORIES[56] = v -> new M56(v);
+ CLASSES[57] = M57.class; FACTORIES[57] = v -> new M57(v);
+ CLASSES[58] = M58.class; FACTORIES[58] = v -> new M58(v);
+ CLASSES[59] = M59.class; FACTORIES[59] = v -> new M59(v);
+ CLASSES[60] = M60.class; FACTORIES[60] = v -> new M60(v);
+ CLASSES[61] = M61.class; FACTORIES[61] = v -> new M61(v);
+ CLASSES[62] = M62.class; FACTORIES[62] = v -> new M62(v);
+ CLASSES[63] = M63.class; FACTORIES[63] = v -> new M63(v);
+ CLASSES[64] = M64.class; FACTORIES[64] = v -> new M64(v);
+ CLASSES[65] = M65.class; FACTORIES[65] = v -> new M65(v);
+ CLASSES[66] = M66.class; FACTORIES[66] = v -> new M66(v);
+ CLASSES[67] = M67.class; FACTORIES[67] = v -> new M67(v);
+ CLASSES[68] = M68.class; FACTORIES[68] = v -> new M68(v);
+ CLASSES[69] = M69.class; FACTORIES[69] = v -> new M69(v);
+ CLASSES[70] = M70.class; FACTORIES[70] = v -> new M70(v);
+ CLASSES[71] = M71.class; FACTORIES[71] = v -> new M71(v);
+ CLASSES[72] = M72.class; FACTORIES[72] = v -> new M72(v);
+ CLASSES[73] = M73.class; FACTORIES[73] = v -> new M73(v);
+ CLASSES[74] = M74.class; FACTORIES[74] = v -> new M74(v);
+ CLASSES[75] = M75.class; FACTORIES[75] = v -> new M75(v);
+ CLASSES[76] = M76.class; FACTORIES[76] = v -> new M76(v);
+ CLASSES[77] = M77.class; FACTORIES[77] = v -> new M77(v);
+ CLASSES[78] = M78.class; FACTORIES[78] = v -> new M78(v);
+ CLASSES[79] = M79.class; FACTORIES[79] = v -> new M79(v);
+ CLASSES[80] = M80.class; FACTORIES[80] = v -> new M80(v);
+ CLASSES[81] = M81.class; FACTORIES[81] = v -> new M81(v);
+ CLASSES[82] = M82.class; FACTORIES[82] = v -> new M82(v);
+ CLASSES[83] = M83.class; FACTORIES[83] = v -> new M83(v);
+ CLASSES[84] = M84.class; FACTORIES[84] = v -> new M84(v);
+ CLASSES[85] = M85.class; FACTORIES[85] = v -> new M85(v);
+ CLASSES[86] = M86.class; FACTORIES[86] = v -> new M86(v);
+ CLASSES[87] = M87.class; FACTORIES[87] = v -> new M87(v);
+ CLASSES[88] = M88.class; FACTORIES[88] = v -> new M88(v);
+ CLASSES[89] = M89.class; FACTORIES[89] = v -> new M89(v);
+ CLASSES[90] = M90.class; FACTORIES[90] = v -> new M90(v);
+ CLASSES[91] = M91.class; FACTORIES[91] = v -> new M91(v);
+ CLASSES[92] = M92.class; FACTORIES[92] = v -> new M92(v);
+ CLASSES[93] = M93.class; FACTORIES[93] = v -> new M93(v);
+ CLASSES[94] = M94.class; FACTORIES[94] = v -> new M94(v);
+ CLASSES[95] = M95.class; FACTORIES[95] = v -> new M95(v);
+ CLASSES[96] = M96.class; FACTORIES[96] = v -> new M96(v);
+ CLASSES[97] = M97.class; FACTORIES[97] = v -> new M97(v);
+ CLASSES[98] = M98.class; FACTORIES[98] = v -> new M98(v);
+ CLASSES[99] = M99.class; FACTORIES[99] = v -> new M99(v);
+ CLASSES[100] = M100.class; FACTORIES[100] = v -> new M100(v);
+ CLASSES[101] = M101.class; FACTORIES[101] = v -> new M101(v);
+ CLASSES[102] = M102.class; FACTORIES[102] = v -> new M102(v);
+ CLASSES[103] = M103.class; FACTORIES[103] = v -> new M103(v);
+ CLASSES[104] = M104.class; FACTORIES[104] = v -> new M104(v);
+ CLASSES[105] = M105.class; FACTORIES[105] = v -> new M105(v);
+ CLASSES[106] = M106.class; FACTORIES[106] = v -> new M106(v);
+ CLASSES[107] = M107.class; FACTORIES[107] = v -> new M107(v);
+ CLASSES[108] = M108.class; FACTORIES[108] = v -> new M108(v);
+ CLASSES[109] = M109.class; FACTORIES[109] = v -> new M109(v);
+ CLASSES[110] = M110.class; FACTORIES[110] = v -> new M110(v);
+ CLASSES[111] = M111.class; FACTORIES[111] = v -> new M111(v);
+ CLASSES[112] = M112.class; FACTORIES[112] = v -> new M112(v);
+ CLASSES[113] = M113.class; FACTORIES[113] = v -> new M113(v);
+ CLASSES[114] = M114.class; FACTORIES[114] = v -> new M114(v);
+ CLASSES[115] = M115.class; FACTORIES[115] = v -> new M115(v);
+ CLASSES[116] = M116.class; FACTORIES[116] = v -> new M116(v);
+ CLASSES[117] = M117.class; FACTORIES[117] = v -> new M117(v);
+ CLASSES[118] = M118.class; FACTORIES[118] = v -> new M118(v);
+ CLASSES[119] = M119.class; FACTORIES[119] = v -> new M119(v);
+ CLASSES[120] = M120.class; FACTORIES[120] = v -> new M120(v);
+ CLASSES[121] = M121.class; FACTORIES[121] = v -> new M121(v);
+ CLASSES[122] = M122.class; FACTORIES[122] = v -> new M122(v);
+ CLASSES[123] = M123.class; FACTORIES[123] = v -> new M123(v);
+ CLASSES[124] = M124.class; FACTORIES[124] = v -> new M124(v);
+ CLASSES[125] = M125.class; FACTORIES[125] = v -> new M125(v);
+ CLASSES[126] = M126.class; FACTORIES[126] = v -> new M126(v);
+ CLASSES[127] = M127.class; FACTORIES[127] = v -> new M127(v);
+ CLASSES[128] = M128.class; FACTORIES[128] = v -> new M128(v);
+ CLASSES[129] = M129.class; FACTORIES[129] = v -> new M129(v);
+ CLASSES[130] = M130.class; FACTORIES[130] = v -> new M130(v);
+ CLASSES[131] = M131.class; FACTORIES[131] = v -> new M131(v);
+ CLASSES[132] = M132.class; FACTORIES[132] = v -> new M132(v);
+ CLASSES[133] = M133.class; FACTORIES[133] = v -> new M133(v);
+ CLASSES[134] = M134.class; FACTORIES[134] = v -> new M134(v);
+ CLASSES[135] = M135.class; FACTORIES[135] = v -> new M135(v);
+ CLASSES[136] = M136.class; FACTORIES[136] = v -> new M136(v);
+ CLASSES[137] = M137.class; FACTORIES[137] = v -> new M137(v);
+ CLASSES[138] = M138.class; FACTORIES[138] = v -> new M138(v);
+ CLASSES[139] = M139.class; FACTORIES[139] = v -> new M139(v);
+ CLASSES[140] = M140.class; FACTORIES[140] = v -> new M140(v);
+ CLASSES[141] = M141.class; FACTORIES[141] = v -> new M141(v);
+ CLASSES[142] = M142.class; FACTORIES[142] = v -> new M142(v);
+ CLASSES[143] = M143.class; FACTORIES[143] = v -> new M143(v);
+ CLASSES[144] = M144.class; FACTORIES[144] = v -> new M144(v);
+ CLASSES[145] = M145.class; FACTORIES[145] = v -> new M145(v);
+ CLASSES[146] = M146.class; FACTORIES[146] = v -> new M146(v);
+ CLASSES[147] = M147.class; FACTORIES[147] = v -> new M147(v);
+ CLASSES[148] = M148.class; FACTORIES[148] = v -> new M148(v);
+ CLASSES[149] = M149.class; FACTORIES[149] = v -> new M149(v);
+ CLASSES[150] = M150.class; FACTORIES[150] = v -> new M150(v);
+ CLASSES[151] = M151.class; FACTORIES[151] = v -> new M151(v);
+ CLASSES[152] = M152.class; FACTORIES[152] = v -> new M152(v);
+ CLASSES[153] = M153.class; FACTORIES[153] = v -> new M153(v);
+ CLASSES[154] = M154.class; FACTORIES[154] = v -> new M154(v);
+ CLASSES[155] = M155.class; FACTORIES[155] = v -> new M155(v);
+ CLASSES[156] = M156.class; FACTORIES[156] = v -> new M156(v);
+ CLASSES[157] = M157.class; FACTORIES[157] = v -> new M157(v);
+ CLASSES[158] = M158.class; FACTORIES[158] = v -> new M158(v);
+ CLASSES[159] = M159.class; FACTORIES[159] = v -> new M159(v);
+ CLASSES[160] = M160.class; FACTORIES[160] = v -> new M160(v);
+ CLASSES[161] = M161.class; FACTORIES[161] = v -> new M161(v);
+ CLASSES[162] = M162.class; FACTORIES[162] = v -> new M162(v);
+ CLASSES[163] = M163.class; FACTORIES[163] = v -> new M163(v);
+ CLASSES[164] = M164.class; FACTORIES[164] = v -> new M164(v);
+ CLASSES[165] = M165.class; FACTORIES[165] = v -> new M165(v);
+ CLASSES[166] = M166.class; FACTORIES[166] = v -> new M166(v);
+ CLASSES[167] = M167.class; FACTORIES[167] = v -> new M167(v);
+ CLASSES[168] = M168.class; FACTORIES[168] = v -> new M168(v);
+ CLASSES[169] = M169.class; FACTORIES[169] = v -> new M169(v);
+ CLASSES[170] = M170.class; FACTORIES[170] = v -> new M170(v);
+ CLASSES[171] = M171.class; FACTORIES[171] = v -> new M171(v);
+ CLASSES[172] = M172.class; FACTORIES[172] = v -> new M172(v);
+ CLASSES[173] = M173.class; FACTORIES[173] = v -> new M173(v);
+ CLASSES[174] = M174.class; FACTORIES[174] = v -> new M174(v);
+ CLASSES[175] = M175.class; FACTORIES[175] = v -> new M175(v);
+ CLASSES[176] = M176.class; FACTORIES[176] = v -> new M176(v);
+ CLASSES[177] = M177.class; FACTORIES[177] = v -> new M177(v);
+ CLASSES[178] = M178.class; FACTORIES[178] = v -> new M178(v);
+ CLASSES[179] = M179.class; FACTORIES[179] = v -> new M179(v);
+ CLASSES[180] = M180.class; FACTORIES[180] = v -> new M180(v);
+ CLASSES[181] = M181.class; FACTORIES[181] = v -> new M181(v);
+ CLASSES[182] = M182.class; FACTORIES[182] = v -> new M182(v);
+ CLASSES[183] = M183.class; FACTORIES[183] = v -> new M183(v);
+ CLASSES[184] = M184.class; FACTORIES[184] = v -> new M184(v);
+ CLASSES[185] = M185.class; FACTORIES[185] = v -> new M185(v);
+ CLASSES[186] = M186.class; FACTORIES[186] = v -> new M186(v);
+ CLASSES[187] = M187.class; FACTORIES[187] = v -> new M187(v);
+ CLASSES[188] = M188.class; FACTORIES[188] = v -> new M188(v);
+ CLASSES[189] = M189.class; FACTORIES[189] = v -> new M189(v);
+ CLASSES[190] = M190.class; FACTORIES[190] = v -> new M190(v);
+ CLASSES[191] = M191.class; FACTORIES[191] = v -> new M191(v);
+ CLASSES[192] = M192.class; FACTORIES[192] = v -> new M192(v);
+ CLASSES[193] = M193.class; FACTORIES[193] = v -> new M193(v);
+ CLASSES[194] = M194.class; FACTORIES[194] = v -> new M194(v);
+ CLASSES[195] = M195.class; FACTORIES[195] = v -> new M195(v);
+ CLASSES[196] = M196.class; FACTORIES[196] = v -> new M196(v);
+ CLASSES[197] = M197.class; FACTORIES[197] = v -> new M197(v);
+ CLASSES[198] = M198.class; FACTORIES[198] = v -> new M198(v);
+ CLASSES[199] = M199.class; FACTORIES[199] = v -> new M199(v);
+ }
+
+ static class M0 extends TypedMetric { M0(final long v) { super(0, v); } }
+
+ static class M1 extends TypedMetric { M1(final long v) { super(1, v); } }
+
+ static class M2 extends TypedMetric { M2(final long v) { super(2, v); } }
+
+ static class M3 extends TypedMetric { M3(final long v) { super(3, v); } }
+
+ static class M4 extends TypedMetric { M4(final long v) { super(4, v); } }
+
+ static class M5 extends TypedMetric { M5(final long v) { super(5, v); } }
+
+ static class M6 extends TypedMetric { M6(final long v) { super(6, v); } }
+
+ static class M7 extends TypedMetric { M7(final long v) { super(7, v); } }
+
+ static class M8 extends TypedMetric { M8(final long v) { super(8, v); } }
+
+ static class M9 extends TypedMetric { M9(final long v) { super(9, v); } }
+
+ static class M10 extends TypedMetric { M10(final long v) { super(10, v); } }
+
+ static class M11 extends TypedMetric { M11(final long v) { super(11, v); } }
+
+ static class M12 extends TypedMetric { M12(final long v) { super(12, v); } }
+
+ static class M13 extends TypedMetric { M13(final long v) { super(13, v); } }
+
+ static class M14 extends TypedMetric { M14(final long v) { super(14, v); } }
+
+ static class M15 extends TypedMetric { M15(final long v) { super(15, v); } }
+
+ static class M16 extends TypedMetric { M16(final long v) { super(16, v); } }
+
+ static class M17 extends TypedMetric { M17(final long v) { super(17, v); } }
+
+ static class M18 extends TypedMetric { M18(final long v) { super(18, v); } }
+
+ static class M19 extends TypedMetric { M19(final long v) { super(19, v); } }
+
+ static class M20 extends TypedMetric { M20(final long v) { super(20, v); } }
+
+ static class M21 extends TypedMetric { M21(final long v) { super(21, v); } }
+
+ static class M22 extends TypedMetric { M22(final long v) { super(22, v); } }
+
+ static class M23 extends TypedMetric { M23(final long v) { super(23, v); } }
+
+ static class M24 extends TypedMetric { M24(final long v) { super(24, v); } }
+
+ static class M25 extends TypedMetric { M25(final long v) { super(25, v); } }
+
+ static class M26 extends TypedMetric { M26(final long v) { super(26, v); } }
+
+ static class M27 extends TypedMetric { M27(final long v) { super(27, v); } }
+
+ static class M28 extends TypedMetric { M28(final long v) { super(28, v); } }
+
+ static class M29 extends TypedMetric { M29(final long v) { super(29, v); } }
+
+ static class M30 extends TypedMetric { M30(final long v) { super(30, v); } }
+
+ static class M31 extends TypedMetric { M31(final long v) { super(31, v); } }
+
+ static class M32 extends TypedMetric { M32(final long v) { super(32, v); } }
+
+ static class M33 extends TypedMetric { M33(final long v) { super(33, v); } }
+
+ static class M34 extends TypedMetric { M34(final long v) { super(34, v); } }
+
+ static class M35 extends TypedMetric { M35(final long v) { super(35, v); } }
+
+ static class M36 extends TypedMetric { M36(final long v) { super(36, v); } }
+
+ static class M37 extends TypedMetric { M37(final long v) { super(37, v); } }
+
+ static class M38 extends TypedMetric { M38(final long v) { super(38, v); } }
+
+ static class M39 extends TypedMetric { M39(final long v) { super(39, v); } }
+
+ static class M40 extends TypedMetric { M40(final long v) { super(40, v); } }
+
+ static class M41 extends TypedMetric { M41(final long v) { super(41, v); } }
+
+ static class M42 extends TypedMetric { M42(final long v) { super(42, v); } }
+
+ static class M43 extends TypedMetric { M43(final long v) { super(43, v); } }
+
+ static class M44 extends TypedMetric { M44(final long v) { super(44, v); } }
+
+ static class M45 extends TypedMetric { M45(final long v) { super(45, v); } }
+
+ static class M46 extends TypedMetric { M46(final long v) { super(46, v); } }
+
+ static class M47 extends TypedMetric { M47(final long v) { super(47, v); } }
+
+ static class M48 extends TypedMetric { M48(final long v) { super(48, v); } }
+
+ static class M49 extends TypedMetric { M49(final long v) { super(49, v); } }
+
+ static class M50 extends TypedMetric { M50(final long v) { super(50, v); } }
+
+ static class M51 extends TypedMetric { M51(final long v) { super(51, v); } }
+
+ static class M52 extends TypedMetric { M52(final long v) { super(52, v); } }
+
+ static class M53 extends TypedMetric { M53(final long v) { super(53, v); } }
+
+ static class M54 extends TypedMetric { M54(final long v) { super(54, v); } }
+
+ static class M55 extends TypedMetric { M55(final long v) { super(55, v); } }
+
+ static class M56 extends TypedMetric { M56(final long v) { super(56, v); } }
+
+ static class M57 extends TypedMetric { M57(final long v) { super(57, v); } }
+
+ static class M58 extends TypedMetric { M58(final long v) { super(58, v); } }
+
+ static class M59 extends TypedMetric { M59(final long v) { super(59, v); } }
+
+ static class M60 extends TypedMetric { M60(final long v) { super(60, v); } }
+
+ static class M61 extends TypedMetric { M61(final long v) { super(61, v); } }
+
+ static class M62 extends TypedMetric { M62(final long v) { super(62, v); } }
+
+ static class M63 extends TypedMetric { M63(final long v) { super(63, v); } }
+
+ static class M64 extends TypedMetric { M64(final long v) { super(64, v); } }
+
+ static class M65 extends TypedMetric { M65(final long v) { super(65, v); } }
+
+ static class M66 extends TypedMetric { M66(final long v) { super(66, v); } }
+
+ static class M67 extends TypedMetric { M67(final long v) { super(67, v); } }
+
+ static class M68 extends TypedMetric { M68(final long v) { super(68, v); } }
+
+ static class M69 extends TypedMetric { M69(final long v) { super(69, v); } }
+
+ static class M70 extends TypedMetric { M70(final long v) { super(70, v); } }
+
+ static class M71 extends TypedMetric { M71(final long v) { super(71, v); } }
+
+ static class M72 extends TypedMetric { M72(final long v) { super(72, v); } }
+
+ static class M73 extends TypedMetric { M73(final long v) { super(73, v); } }
+
+ static class M74 extends TypedMetric { M74(final long v) { super(74, v); } }
+
+ static class M75 extends TypedMetric { M75(final long v) { super(75, v); } }
+
+ static class M76 extends TypedMetric { M76(final long v) { super(76, v); } }
+
+ static class M77 extends TypedMetric { M77(final long v) { super(77, v); } }
+
+ static class M78 extends TypedMetric { M78(final long v) { super(78, v); } }
+
+ static class M79 extends TypedMetric { M79(final long v) { super(79, v); } }
+
+ static class M80 extends TypedMetric { M80(final long v) { super(80, v); } }
+
+ static class M81 extends TypedMetric { M81(final long v) { super(81, v); } }
+
+ static class M82 extends TypedMetric { M82(final long v) { super(82, v); } }
+
+ static class M83 extends TypedMetric { M83(final long v) { super(83, v); } }
+
+ static class M84 extends TypedMetric { M84(final long v) { super(84, v); } }
+
+ static class M85 extends TypedMetric { M85(final long v) { super(85, v); } }
+
+ static class M86 extends TypedMetric { M86(final long v) { super(86, v); } }
+
+ static class M87 extends TypedMetric { M87(final long v) { super(87, v); } }
+
+ static class M88 extends TypedMetric { M88(final long v) { super(88, v); } }
+
+ static class M89 extends TypedMetric { M89(final long v) { super(89, v); } }
+
+ static class M90 extends TypedMetric { M90(final long v) { super(90, v); } }
+
+ static class M91 extends TypedMetric { M91(final long v) { super(91, v); } }
+
+ static class M92 extends TypedMetric { M92(final long v) { super(92, v); } }
+
+ static class M93 extends TypedMetric { M93(final long v) { super(93, v); } }
+
+ static class M94 extends TypedMetric { M94(final long v) { super(94, v); } }
+
+ static class M95 extends TypedMetric { M95(final long v) { super(95, v); } }
+
+ static class M96 extends TypedMetric { M96(final long v) { super(96, v); } }
+
+ static class M97 extends TypedMetric { M97(final long v) { super(97, v); } }
+
+ static class M98 extends TypedMetric { M98(final long v) { super(98, v); } }
+
+ static class M99 extends TypedMetric { M99(final long v) { super(99, v); } }
+
+ static class M100 extends TypedMetric { M100(final long v) { super(100, v); } }
+
+ static class M101 extends TypedMetric { M101(final long v) { super(101, v); } }
+
+ static class M102 extends TypedMetric { M102(final long v) { super(102, v); } }
+
+ static class M103 extends TypedMetric { M103(final long v) { super(103, v); } }
+
+ static class M104 extends TypedMetric { M104(final long v) { super(104, v); } }
+
+ static class M105 extends TypedMetric { M105(final long v) { super(105, v); } }
+
+ static class M106 extends TypedMetric { M106(final long v) { super(106, v); } }
+
+ static class M107 extends TypedMetric { M107(final long v) { super(107, v); } }
+
+ static class M108 extends TypedMetric { M108(final long v) { super(108, v); } }
+
+ static class M109 extends TypedMetric { M109(final long v) { super(109, v); } }
+
+ static class M110 extends TypedMetric { M110(final long v) { super(110, v); } }
+
+ static class M111 extends TypedMetric { M111(final long v) { super(111, v); } }
+
+ static class M112 extends TypedMetric { M112(final long v) { super(112, v); } }
+
+ static class M113 extends TypedMetric { M113(final long v) { super(113, v); } }
+
+ static class M114 extends TypedMetric { M114(final long v) { super(114, v); } }
+
+ static class M115 extends TypedMetric { M115(final long v) { super(115, v); } }
+
+ static class M116 extends TypedMetric { M116(final long v) { super(116, v); } }
+
+ static class M117 extends TypedMetric { M117(final long v) { super(117, v); } }
+
+ static class M118 extends TypedMetric { M118(final long v) { super(118, v); } }
+
+ static class M119 extends TypedMetric { M119(final long v) { super(119, v); } }
+
+ static class M120 extends TypedMetric { M120(final long v) { super(120, v); } }
+
+ static class M121 extends TypedMetric { M121(final long v) { super(121, v); } }
+
+ static class M122 extends TypedMetric { M122(final long v) { super(122, v); } }
+
+ static class M123 extends TypedMetric { M123(final long v) { super(123, v); } }
+
+ static class M124 extends TypedMetric { M124(final long v) { super(124, v); } }
+
+ static class M125 extends TypedMetric { M125(final long v) { super(125, v); } }
+
+ static class M126 extends TypedMetric { M126(final long v) { super(126, v); } }
+
+ static class M127 extends TypedMetric { M127(final long v) { super(127, v); } }
+
+ static class M128 extends TypedMetric { M128(final long v) { super(128, v); } }
+
+ static class M129 extends TypedMetric { M129(final long v) { super(129, v); } }
+
+ static class M130 extends TypedMetric { M130(final long v) { super(130, v); } }
+
+ static class M131 extends TypedMetric { M131(final long v) { super(131, v); } }
+
+ static class M132 extends TypedMetric { M132(final long v) { super(132, v); } }
+
+ static class M133 extends TypedMetric { M133(final long v) { super(133, v); } }
+
+ static class M134 extends TypedMetric { M134(final long v) { super(134, v); } }
+
+ static class M135 extends TypedMetric { M135(final long v) { super(135, v); } }
+
+ static class M136 extends TypedMetric { M136(final long v) { super(136, v); } }
+
+ static class M137 extends TypedMetric { M137(final long v) { super(137, v); } }
+
+ static class M138 extends TypedMetric { M138(final long v) { super(138, v); } }
+
+ static class M139 extends TypedMetric { M139(final long v) { super(139, v); } }
+
+ static class M140 extends TypedMetric { M140(final long v) { super(140, v); } }
+
+ static class M141 extends TypedMetric { M141(final long v) { super(141, v); } }
+
+ static class M142 extends TypedMetric { M142(final long v) { super(142, v); } }
+
+ static class M143 extends TypedMetric { M143(final long v) { super(143, v); } }
+
+ static class M144 extends TypedMetric { M144(final long v) { super(144, v); } }
+
+ static class M145 extends TypedMetric { M145(final long v) { super(145, v); } }
+
+ static class M146 extends TypedMetric { M146(final long v) { super(146, v); } }
+
+ static class M147 extends TypedMetric { M147(final long v) { super(147, v); } }
+
+ static class M148 extends TypedMetric { M148(final long v) { super(148, v); } }
+
+ static class M149 extends TypedMetric { M149(final long v) { super(149, v); } }
+
+ static class M150 extends TypedMetric { M150(final long v) { super(150, v); } }
+
+ static class M151 extends TypedMetric { M151(final long v) { super(151, v); } }
+
+ static class M152 extends TypedMetric { M152(final long v) { super(152, v); } }
+
+ static class M153 extends TypedMetric { M153(final long v) { super(153, v); } }
+
+ static class M154 extends TypedMetric { M154(final long v) { super(154, v); } }
+
+ static class M155 extends TypedMetric { M155(final long v) { super(155, v); } }
+
+ static class M156 extends TypedMetric { M156(final long v) { super(156, v); } }
+
+ static class M157 extends TypedMetric { M157(final long v) { super(157, v); } }
+
+ static class M158 extends TypedMetric { M158(final long v) { super(158, v); } }
+
+ static class M159 extends TypedMetric { M159(final long v) { super(159, v); } }
+
+ static class M160 extends TypedMetric { M160(final long v) { super(160, v); } }
+
+ static class M161 extends TypedMetric { M161(final long v) { super(161, v); } }
+
+ static class M162 extends TypedMetric { M162(final long v) { super(162, v); } }
+
+ static class M163 extends TypedMetric { M163(final long v) { super(163, v); } }
+
+ static class M164 extends TypedMetric { M164(final long v) { super(164, v); } }
+
+ static class M165 extends TypedMetric { M165(final long v) { super(165, v); } }
+
+ static class M166 extends TypedMetric { M166(final long v) { super(166, v); } }
+
+ static class M167 extends TypedMetric { M167(final long v) { super(167, v); } }
+
+ static class M168 extends TypedMetric { M168(final long v) { super(168, v); } }
+
+ static class M169 extends TypedMetric { M169(final long v) { super(169, v); } }
+
+ static class M170 extends TypedMetric { M170(final long v) { super(170, v); } }
+
+ static class M171 extends TypedMetric { M171(final long v) { super(171, v); } }
+
+ static class M172 extends TypedMetric { M172(final long v) { super(172, v); } }
+
+ static class M173 extends TypedMetric { M173(final long v) { super(173, v); } }
+
+ static class M174 extends TypedMetric { M174(final long v) { super(174, v); } }
+
+ static class M175 extends TypedMetric { M175(final long v) { super(175, v); } }
+
+ static class M176 extends TypedMetric { M176(final long v) { super(176, v); } }
+
+ static class M177 extends TypedMetric { M177(final long v) { super(177, v); } }
+
+ static class M178 extends TypedMetric { M178(final long v) { super(178, v); } }
+
+ static class M179 extends TypedMetric { M179(final long v) { super(179, v); } }
+
+ static class M180 extends TypedMetric { M180(final long v) { super(180, v); } }
+
+ static class M181 extends TypedMetric { M181(final long v) { super(181, v); } }
+
+ static class M182 extends TypedMetric { M182(final long v) { super(182, v); } }
+
+ static class M183 extends TypedMetric { M183(final long v) { super(183, v); } }
+
+ static class M184 extends TypedMetric { M184(final long v) { super(184, v); } }
+
+ static class M185 extends TypedMetric { M185(final long v) { super(185, v); } }
+
+ static class M186 extends TypedMetric { M186(final long v) { super(186, v); } }
+
+ static class M187 extends TypedMetric { M187(final long v) { super(187, v); } }
+
+ static class M188 extends TypedMetric { M188(final long v) { super(188, v); } }
+
+ static class M189 extends TypedMetric { M189(final long v) { super(189, v); } }
+
+ static class M190 extends TypedMetric { M190(final long v) { super(190, v); } }
+
+ static class M191 extends TypedMetric { M191(final long v) { super(191, v); } }
+
+ static class M192 extends TypedMetric { M192(final long v) { super(192, v); } }
+
+ static class M193 extends TypedMetric { M193(final long v) { super(193, v); } }
+
+ static class M194 extends TypedMetric { M194(final long v) { super(194, v); } }
+
+ static class M195 extends TypedMetric { M195(final long v) { super(195, v); } }
+
+ static class M196 extends TypedMetric { M196(final long v) { super(196, v); } }
+
+ static class M197 extends TypedMetric { M197(final long v) { super(197, v); } }
+
+ static class M198 extends TypedMetric { M198(final long v) { super(198, v); } }
+
+ static class M199 extends TypedMetric { M199(final long v) { super(199, v); } }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicyTest.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicyTest.java
new file mode 100644
index 0000000000..0f95402c56
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/PartitionPolicyTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class PartitionPolicyTest {
+
+ @Test
+ public void testFixedReturnsExactCount() {
+ assertEquals(1, PartitionPolicy.fixed(1).resolve(4));
+ assertEquals(8, PartitionPolicy.fixed(8).resolve(4));
+ }
+
+ @Test
+ public void testFixedIgnoresThreadCount() {
+ final PartitionPolicy policy = PartitionPolicy.fixed(5);
+ assertEquals(5, policy.resolve(1));
+ assertEquals(5, policy.resolve(100));
+ }
+
+ @Test
+ public void testFixedRejectsZero() {
+ assertThrows(IllegalArgumentException.class, () -> PartitionPolicy.fixed(0));
+ }
+
+ @Test
+ public void testFixedRejectsNegative() {
+ assertThrows(IllegalArgumentException.class, () -> PartitionPolicy.fixed(-1));
+ }
+
+ @Test
+ public void testThreadMultiply() {
+ assertEquals(8, PartitionPolicy.threadMultiply(2).resolve(4));
+ assertEquals(12, PartitionPolicy.threadMultiply(3).resolve(4));
+ }
+
+ @Test
+ public void testThreadMultiplyMinOne() {
+ // Even with 0 thread count, should return at least 1
+ assertEquals(1, PartitionPolicy.threadMultiply(1).resolve(0));
+ }
+
+ @Test
+ public void testThreadMultiplyRejectsZero() {
+ assertThrows(IllegalArgumentException.class, () -> PartitionPolicy.threadMultiply(0));
+ }
+
+ @Test
+ public void testToString() {
+ assertEquals("fixed(4)", PartitionPolicy.fixed(4).toString());
+ assertEquals("threadMultiply(2)", PartitionPolicy.threadMultiply(2).toString());
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicyTest.java b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicyTest.java
new file mode 100644
index 0000000000..4a77f7ad43
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/java/org/apache/skywalking/oap/server/library/batchqueue/ThreadPolicyTest.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.batchqueue;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ThreadPolicyTest {
+
+ @Test
+ public void testFixedReturnsExactCount() {
+ assertEquals(1, ThreadPolicy.fixed(1).resolve());
+ assertEquals(4, ThreadPolicy.fixed(4).resolve());
+ assertEquals(100, ThreadPolicy.fixed(100).resolve());
+ }
+
+ @Test
+ public void testFixedRejectsZero() {
+ assertThrows(IllegalArgumentException.class, () -> ThreadPolicy.fixed(0));
+ }
+
+ @Test
+ public void testFixedRejectsNegative() {
+ assertThrows(IllegalArgumentException.class, () -> ThreadPolicy.fixed(-1));
+ }
+
+ @Test
+ public void testCpuCoresResolvesAtLeastOne() {
+ // Even with a tiny multiplier, resolve should return >= 1
+ assertTrue(ThreadPolicy.cpuCores(0.001).resolve() >= 1);
+ }
+
+ @Test
+ public void testCpuCoresScalesWithProcessors() {
+ final int cores = Runtime.getRuntime().availableProcessors();
+ final int resolved = ThreadPolicy.cpuCores(1.0).resolve();
+ assertEquals(cores, resolved);
+ }
+
+ @Test
+ public void testCpuCoresRejectsZero() {
+ assertThrows(IllegalArgumentException.class, () -> ThreadPolicy.cpuCores(0));
+ }
+
+ @Test
+ public void testCpuCoresRejectsNegative() {
+ assertThrows(IllegalArgumentException.class, () -> ThreadPolicy.cpuCores(-0.5));
+ }
+
+ @Test
+ public void testToStringFixed() {
+ assertEquals("fixed(4)", ThreadPolicy.fixed(4).toString());
+ }
+
+ @Test
+ public void testToStringCpuCores() {
+ assertEquals("cpuCores(0.5)", ThreadPolicy.cpuCores(0.5).toString());
+ }
+}
diff --git a/oap-server/server-library/library-batch-queue/src/test/resources/log4j2-test.xml b/oap-server/server-library/library-batch-queue/src/test/resources/log4j2-test.xml
new file mode 100644
index 0000000000..6c77b4669f
--- /dev/null
+++ b/oap-server/server-library/library-batch-queue/src/test/resources/log4j2-test.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ ~
+ -->
+
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout charset="UTF-8" pattern="%d - %c - [%t] %-5p - %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Logger name="org.apache.skywalking.oap.server.library.batchqueue" level="INFO"/>
+ <Root level="WARN">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/oap-server/server-library/library-datacarrier-queue/src/test/java/org/apache/skywalking/oap/server/library/datacarrier/DataCarrierBenchmark.java b/oap-server/server-library/library-datacarrier-queue/src/test/java/org/apache/skywalking/oap/server/library/datacarrier/DataCarrierBenchmark.java
new file mode 100644
index 0000000000..f6efad4171
--- /dev/null
+++ b/oap-server/server-library/library-datacarrier-queue/src/test/java/org/apache/skywalking/oap/server/library/datacarrier/DataCarrierBenchmark.java
@@ -0,0 +1,417 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.datacarrier;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.skywalking.oap.server.library.datacarrier.buffer.BufferStrategy;
+import org.apache.skywalking.oap.server.library.datacarrier.consumer.BulkConsumePool;
+import org.apache.skywalking.oap.server.library.datacarrier.consumer.IConsumer;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Throughput benchmark for DataCarrier, comparable to BatchQueueBenchmark.
+ *
+ * <p>Simulates the real production pattern: N DataCarrier instances (one per
+ * metric type), each with 1 channel, all sharing a {@link BulkConsumePool}
+ * with 4 {@link org.apache.skywalking.oap.server.library.datacarrier.consumer.MultipleChannelsConsumer} threads.
+ * Producers round-robin across the N carriers by type.
+ *
+ * <p>Run with: mvn test -pl oap-server/server-library/library-datacarrier-queue
+ * -Dtest=DataCarrierBenchmark -DfailIfNoTests=false
+ *
+ * <h3>Reference results (Apple M3 Max, 128 GB RAM, macOS 26.2, JDK 17)</h3>
+ * <pre>
+ * Benchmark Config IF_POSSIBLE BLOCKING
+ * ------------------------------- ---------------------- ------------- -------------
+ * direct-1ch-1consumer 1ch, 1 consumer ~35,300,000 ~35,300,000
+ * direct-8ch-4consumer 8ch, 4 consumers ~31,100,000 ~21,900,000
+ * 40-types-pool4t (prod-like) 40 carriers, pool(4) ~42,000,000 ~40,600,000
+ * 200-types-pool4t (prod-like) 200 carriers, pool(4) ~25,100,000 ~23,900,000
+ * 200reg/50hot, 100 producers 200 carriers, pool(4) ~17,600,000 ~23,800,000
+ *
+ * All runs: 4 producer threads (except hot-types: 100), bufferSize=50,000,
+ * consumeCycle=1ms, 0% drop rate.
+ * </pre>
+ *
+ * <h3>Comparison with BatchQueue — production case (4 consumer threads)</h3>
+ * <pre>
+ * Scenario DataCarrier BatchQueue
+ * (N carriers (1 queue +
+ * + BulkPool) handler map)
+ * ------------------------------- ----------- -----------
+ * 40 types, 4 producers ~42,000,000 ~31,200,000
+ * 200 types, 4 producers ~25,100,000 ~43,900,000
+ * 200 reg / 50 hot, 100 producers ~17,600,000 ~26,900,000 (IF_POSSIBLE)
+ * 200 reg / 50 hot, 100 producers ~23,800,000 ~26,100,000 (BLOCKING)
+ *
+ * BatchQueue uses configurable PartitionSelector (default: typeHash).
+ * Same-type items always land on the same partition, so dispatch grouping
+ * is effectively a no-op. In the 200-type case, BatchQueue reaches ~44M/s
+ * (1.75x DataCarrier's 25M) because 200 types spread across 32 partitions
+ * with minimal collision, while DataCarrier's consumer threads must iterate
+ * all 200 groups per cycle. In the production-realistic hot-types scenario
+ * (200reg/50hot/100 producers), BatchQueue leads ~27M vs ~18M (IF_POSSIBLE).
+ * DataCarrier wins the 40-type case (~42M vs ~31M) where fewer types mean
+ * less iteration overhead for its per-carrier polling model.
+ * </pre>
+ */
+@SuppressWarnings("all")
+public class DataCarrierBenchmark {
+
+ private static final int WARMUP_SECONDS = 2;
+ private static final int MEASURE_SECONDS = 5;
+ private static final int PRODUCER_THREADS = 4;
+ private static final int BUFFER_SIZE = 50_000;
+
+ // ---- Production-like: N types sharing BulkConsumePool ----
+
+ @Test
+ public void benchmark40TypesSharedPool4Threads() throws Exception {
+ runSharedPoolBenchmark("40-types-pool4t-ifpossible",
+ 40, 4, 1, BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmark200TypesSharedPool4Threads() throws Exception {
+ runSharedPoolBenchmark("200-types-pool4t-ifpossible",
+ 200, 4, 1, BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmark40TypesSharedPool4ThreadsBlocking() throws Exception {
+ runSharedPoolBenchmark("40-types-pool4t-blocking",
+ 40, 4, 1, BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmark200TypesSharedPool4ThreadsBlocking() throws Exception {
+ runSharedPoolBenchmark("200-types-pool4t-blocking",
+ 200, 4, 1, BufferStrategy.BLOCKING);
+ }
+
+ // ---- Production-realistic: 200 types, 50 hot, 100 gRPC producers ----
+
+ @Test
+ public void benchmarkHotTypes200x50x100producers() throws Exception {
+ runHotTypesPoolBenchmark("200reg-50hot-100p-ifpossible",
+ 200, 50, 100, 4, 1, BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkHotTypes200x50x100producersBlocking() throws Exception {
+ runHotTypesPoolBenchmark("200reg-50hot-100p-blocking",
+ 200, 50, 100, 4, 1, BufferStrategy.BLOCKING);
+ }
+
+ // ---- Single carrier baselines ----
+
+ @Test
+ public void benchmarkDirectConsumer1Thread() throws Exception {
+ runSingleBenchmark("direct-1ch-1consumer-ifpossible",
+ 1, 1, 1, BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkDirectConsumer4Threads() throws Exception {
+ runSingleBenchmark("direct-8ch-4consumer-ifpossible",
+ 8, 4, 1, BufferStrategy.IF_POSSIBLE);
+ }
+
+ @Test
+ public void benchmarkDirectConsumer1ThreadBlocking() throws Exception {
+ runSingleBenchmark("direct-1ch-1consumer-blocking",
+ 1, 1, 1, BufferStrategy.BLOCKING);
+ }
+
+ @Test
+ public void benchmarkDirectConsumer4ThreadsBlocking() throws Exception {
+ runSingleBenchmark("direct-8ch-4consumer-blocking",
+ 8, 4, 1, BufferStrategy.BLOCKING);
+ }
+
+ /**
+ * Production-like benchmark: N DataCarrier instances sharing one BulkConsumePool.
+ * Each carrier has 1 channel. Producers round-robin across carriers by type.
+ */
+ private void runSharedPoolBenchmark(final String label, final int typeCount,
+ final int poolThreads, final long consumeCycleMs,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final BulkConsumePool pool = new BulkConsumePool(
+ "bench-pool", poolThreads, consumeCycleMs, false);
+
+ final DataCarrier<Long>[] carriers = new DataCarrier[typeCount];
+ for (int i = 0; i < typeCount; i++) {
+ carriers[i] = new DataCarrier<>(
+ "type-" + i, "bench", 1, BUFFER_SIZE, strategy);
+ carriers[i].consume(pool, new IConsumer<Long>() {
+ @Override
+ public void consume(final List<Long> data) {
+ consumed.addAndGet(data.size());
+ }
+
+ @Override
+ public void onError(final List<Long> data, final Throwable t) {
+ t.printStackTrace();
+ }
+ });
+ }
+
+ // Warmup
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducers(carriers, warmupEnd);
+ Thread.sleep(200);
+ final long warmupConsumed = consumed.get();
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducers(carriers, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ pool.close(null);
+
+ System.out.printf("%n=== DataCarrier Benchmark: %s ===%n"
+ + " Types: %d (1 DataCarrier per type, 1 channel each)%n"
+ + " Pool threads:%d%n"
+ + " ConsumeCycle:%d ms%n"
+ + " Strategy: %s%n"
+ + " Producers: %d%n"
+ + " Duration: %d ms%n"
+ + " Warmup consumed: %,d%n"
+ + " Produced: %,d%n"
+ + " Consumed: %,d%n"
+ + " Produce rate: %,.0f items/sec%n"
+ + " Consume rate: %,.0f items/sec%n"
+ + " Drop rate: %.2f%%%n",
+ label, typeCount, poolThreads, consumeCycleMs, strategy,
+ PRODUCER_THREADS, measureDuration, warmupConsumed,
+ produced, totalConsumed,
+ produced * 1000.0 / measureDuration,
+ totalConsumed * 1000.0 / measureDuration,
+ produced > 0 ? (produced - totalConsumed) * 100.0 / produced : 0);
+ }
+
+ /**
+ * Production-realistic: 200 DataCarrier instances created, but only 50 hot types
+ * get traffic from 100 producer threads (simulating gRPC connections).
+ * Consumer threads must iterate all 200 groups each cycle, including 150 idle ones.
+ */
+ private void runHotTypesPoolBenchmark(final String label, final int totalTypes,
+ final int hotTypes, final int producerCount,
+ final int poolThreads, final long consumeCycleMs,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final BulkConsumePool pool = new BulkConsumePool(
+ "bench-hot-pool", poolThreads, consumeCycleMs, false);
+
+ // Create ALL carriers (200), each registered with the shared pool
+ final DataCarrier<Long>[] carriers = new DataCarrier[totalTypes];
+ for (int i = 0; i < totalTypes; i++) {
+ carriers[i] = new DataCarrier<>(
+ "type-" + i, "bench", 1, BUFFER_SIZE, strategy);
+ carriers[i].consume(pool, new IConsumer<Long>() {
+ @Override
+ public void consume(final List<Long> data) {
+ consumed.addAndGet(data.size());
+ }
+
+ @Override
+ public void onError(final List<Long> data, final Throwable t) {
+ t.printStackTrace();
+ }
+ });
+ }
+
+ // Only hot carriers get traffic
+ final DataCarrier<Long>[] hotCarriers = new DataCarrier[hotTypes];
+ System.arraycopy(carriers, 0, hotCarriers, 0, hotTypes);
+
+ // Warmup
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducersN(hotCarriers, producerCount, warmupEnd);
+ Thread.sleep(200);
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducersN(hotCarriers, producerCount, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ pool.close(null);
+
+ System.out.printf("%n=== DataCarrier Benchmark: %s ===%n"
+ + " Total types: %d (all registered with pool)%n"
+ + " Hot types: %d (receiving traffic)%n"
+ + " Pool threads:%d%n"
+ + " ConsumeCycle:%d ms%n"
+ + " Strategy: %s%n"
+ + " Producers: %d (simulating gRPC connections)%n"
+ + " Duration: %d ms%n"
+ + " Produced: %,d%n"
+ + " Consumed: %,d%n"
+ + " Produce rate: %,.0f items/sec%n"
+ + " Consume rate: %,.0f items/sec%n"
+ + " Drop rate: %.2f%%%n",
+ label, totalTypes, hotTypes, poolThreads, consumeCycleMs, strategy,
+ producerCount, measureDuration,
+ produced, totalConsumed,
+ produced * 1000.0 / measureDuration,
+ totalConsumed * 1000.0 / measureDuration,
+ produced > 0 ? (produced - totalConsumed) * 100.0 / produced : 0);
+ }
+
+ private long runProducersN(final DataCarrier<Long>[] carriers,
+ final int producerCount,
+ final long endTimeMs) throws InterruptedException {
+ final int carrierCount = carriers.length;
+ final AtomicLong totalProduced = new AtomicLong(0);
+ final CountDownLatch done = new CountDownLatch(producerCount);
+
+ for (int p = 0; p < producerCount; p++) {
+ final int producerIndex = p;
+ final Thread thread = new Thread(() -> {
+ long count = 0;
+ int typeIndex = producerIndex;
+ while (System.currentTimeMillis() < endTimeMs) {
+ for (int batch = 0; batch < 100; batch++) {
+ final int type = typeIndex % carrierCount;
+ if (carriers[type].produce(count)) {
+ count++;
+ }
+ typeIndex++;
+ }
+ }
+ totalProduced.addAndGet(count);
+ done.countDown();
+ });
+ thread.setName("gRPC-" + producerIndex);
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ done.await(MEASURE_SECONDS + 10, TimeUnit.SECONDS);
+ return totalProduced.get();
+ }
+
+ private void runSingleBenchmark(final String label, final int channelSize,
+ final int consumerThreads, final long consumeCycleMs,
+ final BufferStrategy strategy) throws Exception {
+ final AtomicLong consumed = new AtomicLong(0);
+
+ final DataCarrier<Long> carrier = new DataCarrier<>(
+ "bench-" + label, "bench", channelSize, BUFFER_SIZE, strategy);
+
+ carrier.consume(new IConsumer<Long>() {
+ @Override
+ public void consume(final List<Long> data) {
+ consumed.addAndGet(data.size());
+ }
+
+ @Override
+ public void onError(final List<Long> data, final Throwable t) {
+ t.printStackTrace();
+ }
+ }, consumerThreads, consumeCycleMs);
+
+ final DataCarrier<Long>[] carriers = new DataCarrier[] {carrier};
+
+ // Warmup
+ final long warmupEnd = System.currentTimeMillis() + WARMUP_SECONDS * 1000L;
+ runProducers(carriers, warmupEnd);
+ Thread.sleep(200);
+ final long warmupConsumed = consumed.get();
+ consumed.set(0);
+
+ // Measure
+ final long measureStart = System.currentTimeMillis();
+ final long measureEnd = measureStart + MEASURE_SECONDS * 1000L;
+ final long produced = runProducers(carriers, measureEnd);
+ final long measureDuration = System.currentTimeMillis() - measureStart;
+
+ Thread.sleep(500);
+ final long totalConsumed = consumed.get();
+
+ carrier.shutdownConsumers();
+
+ System.out.printf("%n=== DataCarrier Benchmark: %s ===%n"
+ + " Channels: %d%n"
+ + " Consumers: %d%n"
+ + " ConsumeCycle:%d ms%n"
+ + " Strategy: %s%n"
+ + " Producers: %d%n"
+ + " Duration: %d ms%n"
+ + " Warmup consumed: %,d%n"
+ + " Produced: %,d%n"
+ + " Consumed: %,d%n"
+ + " Produce rate: %,.0f items/sec%n"
+ + " Consume rate: %,.0f items/sec%n"
+ + " Drop rate: %.2f%%%n",
+ label, channelSize, consumerThreads, consumeCycleMs, strategy,
+ PRODUCER_THREADS, measureDuration, warmupConsumed,
+ produced, totalConsumed,
+ produced * 1000.0 / measureDuration,
+ totalConsumed * 1000.0 / measureDuration,
+ produced > 0 ? (produced - totalConsumed) * 100.0 / produced : 0);
+ }
+
+ private long runProducers(final DataCarrier<Long>[] carriers,
+ final long endTimeMs) throws InterruptedException {
+ final int carrierCount = carriers.length;
+ final AtomicLong totalProduced = new AtomicLong(0);
+ final CountDownLatch done = new CountDownLatch(PRODUCER_THREADS);
+
+ for (int p = 0; p < PRODUCER_THREADS; p++) {
+ final int producerIndex = p;
+ final Thread thread = new Thread(() -> {
+ long count = 0;
+ int typeIndex = producerIndex;
+ while (System.currentTimeMillis() < endTimeMs) {
+ for (int batch = 0; batch < 100; batch++) {
+ final int type = typeIndex % carrierCount;
+ if (carriers[type].produce(count)) {
+ count++;
+ }
+ typeIndex++;
+ }
+ }
+ totalProduced.addAndGet(count);
+ done.countDown();
+ });
+ thread.setName("Producer-" + producerIndex);
+ thread.start();
+ }
+
+ done.await(MEASURE_SECONDS + 10, TimeUnit.SECONDS);
+ return totalProduced.get();
+ }
+}
diff --git a/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/benchmark/StringFormatGroupBenchmark.java b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/benchmark/StringFormatGroupBenchmark.java
new file mode 100644
index 0000000000..ad850288de
--- /dev/null
+++ b/oap-server/server-library/library-util/src/test/java/org/apache/skywalking/oap/server/library/util/benchmark/StringFormatGroupBenchmark.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.library.util.benchmark;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.library.util.StringFormatGroup;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+public class StringFormatGroupBenchmark {
+
+ private static final int WARMUP_ITERATIONS = 5000;
+ private static final int MEASURE_ITERATIONS = 100_000;
+
+ @Test
+ public void testMatch() {
+ for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+ doMatch();
+ }
+ final long start = System.nanoTime();
+ for (int i = 0; i < MEASURE_ITERATIONS; i++) {
+ doMatch();
+ }
+ final long elapsed = System.nanoTime() - start;
+ log.info("testMatch: {} ops, {} ns/op, {} ops/s",
+ MEASURE_ITERATIONS, elapsed / MEASURE_ITERATIONS,
+ MEASURE_ITERATIONS * 1_000_000_000L / elapsed);
+ }
+
+ @Test
+ public void test100Rule() {
+ for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+ do100Rule();
+ }
+ final long start = System.nanoTime();
+ for (int i = 0; i < MEASURE_ITERATIONS; i++) {
+ do100Rule();
+ }
+ final long elapsed = System.nanoTime() - start;
+ log.info("test100Rule: {} ops, {} ns/op, {} ops/s",
+ MEASURE_ITERATIONS, elapsed / MEASURE_ITERATIONS,
+ MEASURE_ITERATIONS * 1_000_000_000L / elapsed);
+ }
+
+ private void doMatch() {
+ StringFormatGroup group = new StringFormatGroup();
+ group.addRule("/name/*/add", "/name/.+/add");
+ group.format("/name/test/add");
+
+ group = new StringFormatGroup();
+ group.addRule("/name/*/add/{orderId}", "/name/.+/add/.*");
+ group.format("/name/test/add/12323");
+ }
+
+ private void do100Rule() {
+ final StringFormatGroup group = new StringFormatGroup();
+ group.addRule("/name/*/add/{orderId}", "/name/.+/add/.*");
+ for (int i = 0; i < 100; i++) {
+ group.addRule("/name/*/add/{orderId}" + "/" + 1, "/name/.+/add/.*" + "/abc");
+ }
+ group.format("/name/test/add/12323");
+ }
+}
diff --git a/oap-server/server-library/pom.xml b/oap-server/server-library/pom.xml
index 84eeae3260..3505d22893 100644
--- a/oap-server/server-library/pom.xml
+++ b/oap-server/server-library/pom.xml
@@ -39,5 +39,6 @@
<module>library-pprof-parser</module>
<module>library-integration-test</module>
<module>library-banyandb-client</module>
+ <module>library-batch-queue</module>
</modules>
</project>