You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/19 21:18:33 UTC
(pulsar) 14/16: [improve][broker] Add scalable-topic controller and layout metrics (#26983)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 974d6669c04a8815e0d1240ff01e561929b80299
Author: Penghui Li <pe...@apache.org>
AuthorDate: Mon Oct 19 12:11:45 2026 -0700
[improve][broker] Add scalable-topic controller and layout metrics (#26983)
(cherry picked from commit 2f5bd98e960c72cdce6e588f8db616fe8918d03b)
---
.../broker/service/scalable/AutoScaleDecision.java | 24 +-
.../service/scalable/AutoScalePolicyEvaluator.java | 86 ++-
.../service/scalable/ScalableTopicController.java | 175 +++++-
.../scalable/ScalableTopicControllerMetrics.java | 417 ++++++++++++++
.../service/scalable/ScalableTopicService.java | 39 +-
.../scalable/AutoScalePolicyEvaluatorTest.java | 258 ++++++++-
.../ScalableTopicControllerAutoScaleTest.java | 22 +
.../ScalableTopicControllerFailoverTest.java | 219 ++++++++
...calableTopicControllerMetricsIsolationTest.java | 435 +++++++++++++++
.../ScalableTopicControllerMetricsTest.java | 618 +++++++++++++++++++++
.../scalable/ScalableTopicControllerTest.java | 192 ++++++-
.../service/scalable/ScalableTopicServiceTest.java | 75 +++
12 files changed, 2506 insertions(+), 54 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java
index 9dd74afb16ff..966ce1db1417 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java
@@ -18,14 +18,16 @@
*/
package org.apache.pulsar.broker.service.scalable;
+import java.util.List;
+
/**
- * The outcome of one {@link AutoScalePolicyEvaluator} evaluation (PIP-483): split one
- * segment, merge two adjacent segments, or do nothing. Each non-{@link NoAction} variant
- * carries a short {@code reason} string used for logging and metrics.
+ * The outcome of one {@link AutoScalePolicyEvaluator} evaluation (PIP-483): split, merge,
+ * rebucket, or do nothing. Actions and blocked evaluations carry a short {@code reason}
+ * string used for logging and metrics.
*/
public sealed interface AutoScaleDecision
permits AutoScaleDecision.Split, AutoScaleDecision.Merge, AutoScaleDecision.Rebucket,
- AutoScaleDecision.NoAction {
+ AutoScaleDecision.Blocked, AutoScaleDecision.NoAction {
/** Split {@code segmentId} at its midpoint. */
record Split(long segmentId, String reason) implements AutoScaleDecision {
@@ -41,11 +43,21 @@ public sealed interface AutoScaleDecision
* instead of a split. One decision carries the whole batch so a multi-segment topic
* converges to a uniform bucketing in a single evaluation, not one segment per cooldown.
*/
- record Rebucket(java.util.List<Long> segmentIds, int newBucketCount, String reason)
+ record Rebucket(List<Long> segmentIds, int newBucketCount, String reason)
implements AutoScaleDecision {
}
- /** No action this evaluation. */
+ /**
+ * Eligible demand was blocked by a hard limit. The evaluator emits
+ * {@code max_segments}, {@code max_entry_buckets}, {@code max_dag_depth}, or
+ * {@code unsplittable}; policy resolution may report {@code invalid_policy} separately.
+ * An unsplittable split candidate ends evaluation without selecting an alternative action.
+ * Cooldowns alone never produce a blocked decision.
+ */
+ record Blocked(String reason) implements AutoScaleDecision {
+ }
+
+ /** No action this evaluation, without eligible demand blocked by a hard limit. */
record NoAction() implements AutoScaleDecision {
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java
index 80c4c3332f88..26b8f1f9c1a3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java
@@ -44,6 +44,10 @@ import org.apache.pulsar.common.scalable.SegmentLoadStats;
* consumer that owns a bucket today, since the rebucket lane cannot grow past the
* ceiling to win the capacity back.</li>
* </ol>
+ * A selected single-slot split candidate is reported as {@link AutoScaleDecision.Blocked}
+ * without selecting an alternative action. Other hard limits are reported only for
+ * otherwise-eligible demand, after the remaining lanes have had a chance to act;
+ * cooldowns do not themselves count as blocked evaluations.
*/
public final class AutoScalePolicyEvaluator {
@@ -90,17 +94,26 @@ public final class AutoScalePolicyEvaluator {
AutoScaleDecision consumerScale = tryConsumerScale(active, loadBySegment,
streamConsumerCount, bucketSharingSubscriptions, config, nowMs, lastSplitAtMs,
lastRebucketAtMs);
- if (!(consumerScale instanceof AutoScaleDecision.NoAction)) {
+ if (endsEvaluation(consumerScale)) {
return consumerScale;
}
AutoScaleDecision split = trySplit(active, loadBySegment, config, nowMs, lastSplitAtMs);
- if (!(split instanceof AutoScaleDecision.NoAction)) {
+ if (endsEvaluation(split)) {
return split;
}
- return tryMerge(active, layout, loadBySegment, streamConsumerCount, config, nowMs,
- lastMergeAtMs);
+ AutoScaleDecision merge = tryMerge(active, layout, loadBySegment, streamConsumerCount,
+ config, nowMs, lastMergeAtMs);
+ if (endsEvaluation(merge)) {
+ return merge;
+ }
+ // Cap diagnostics must not suppress a valid action in another lane. If none can
+ // act, report the first blocked lane in the same priority order as the actions.
+ if (consumerScale instanceof AutoScaleDecision.Blocked) {
+ return consumerScale;
+ }
+ return split instanceof AutoScaleDecision.Blocked ? split : merge;
}
// --- Consumer-driven scale-up: segments vs entry-buckets (PIP-486) ---
@@ -113,9 +126,9 @@ public final class AutoScalePolicyEvaluator {
* inbound rate is at or above {@code splitVsRebucketMinMsgRateIn} and the topic is
* under {@code maxSegments} — today's "segments first" behavior.</li>
* <li><b>Rebucket-up</b> otherwise (a low-throughput topic, or the topic is at the
- * segment cap): if the existing entry-bucket capacity cannot absorb the surplus,
- * roll the smallest-bucketed segment over to the smallest power of two that lets
- * every consumer own a bucket, capped at {@code maxEntryBucketsPerSegment}.
+ * segment cap): if the existing entry-bucket capacity cannot
+ * absorb the surplus, roll under-target segments over to a common power-of-two
+ * bucket count sized to the consumers, capped at {@code maxEntryBucketsPerSegment}.
* Raising is fast (one rollover sized to the surplus); lowering is deliberately
* not automated here — spiky consumer counts must not flap the bucketing. Only
* bucket-sharing subscriptions count here: a checkpoint group's members each read
@@ -146,13 +159,25 @@ public final class AutoScalePolicyEvaluator {
boolean atSegmentCap = segments >= config.maxSegments();
boolean belowSplitFloor = statsOf(busiest.segmentId(), loadBySegment).msgRateIn()
< config.splitVsRebucketMinMsgRateIn();
+ boolean splitCoolingDown = withinCooldown(nowMs, lastSplitAtMs, config.splitCooldown().toMillis());
+ AutoScaleDecision blocked = AutoScaleDecision.NONE;
if (!atSegmentCap && !belowSplitFloor) {
- // Traffic justifies a physical segment.
- if (withinCooldown(nowMs, lastSplitAtMs, config.splitCooldown().toMillis())) {
+ // Traffic justifies a physical segment, but a single-slot hot spot cannot split.
+ if (splitCoolingDown) {
return AutoScaleDecision.NONE;
}
- return new AutoScaleDecision.Split(busiest.segmentId(), "consumer-count");
+ return busiest.hashRange().size() < 2
+ ? new AutoScaleDecision.Blocked("unsplittable")
+ : new AutoScaleDecision.Split(busiest.segmentId(), "consumer-count");
+ } else if (atSegmentCap && !belowSplitFloor && !splitCoolingDown) {
+ // Whole-segment readers cannot use buckets as an alternative to another segment.
+ int wholeSegmentConsumers = streamConsumerCount.entrySet().stream()
+ .filter(e -> !bucketSharingSubscriptions.contains(e.getKey()))
+ .mapToInt(Map.Entry::getValue).max().orElse(0);
+ if (wholeSegmentConsumers > segments) {
+ blocked = new AutoScaleDecision.Blocked("max_segments");
+ }
}
// Bucket lane: absorb the bucket-sharing subscriptions' surplus with entry-buckets.
@@ -165,10 +190,10 @@ public final class AutoScalePolicyEvaluator {
}
if (bucketConsumers <= capacity) {
// The existing buckets already absorb the surplus (broker-side fan-out).
- return AutoScaleDecision.NONE;
+ return blocked;
}
if (withinCooldown(nowMs, lastRebucketAtMs, config.rebucketCooldown().toMillis())) {
- return AutoScaleDecision.NONE;
+ return blocked;
}
// One shot: bring every segment below the common per-segment target up to it in a
// single decision, so the topic converges to a uniform bucketing in one evaluation —
@@ -183,7 +208,7 @@ public final class AutoScalePolicyEvaluator {
}
if (below.isEmpty()) {
// Bucket capacity is maxed out; the remaining surplus stays idle.
- return AutoScaleDecision.NONE;
+ return new AutoScaleDecision.Blocked("max_entry_buckets");
}
below.sort(Long::compareTo);
return new AutoScaleDecision.Rebucket(below, target,
@@ -199,9 +224,6 @@ public final class AutoScalePolicyEvaluator {
long nowMs,
long lastSplitAtMs) {
- if (active.size() >= config.maxSegments()) {
- return AutoScaleDecision.NONE;
- }
if (withinCooldown(nowMs, lastSplitAtMs, config.splitCooldown().toMillis())) {
return AutoScaleDecision.NONE;
}
@@ -235,7 +257,13 @@ public final class AutoScalePolicyEvaluator {
}
}
if (hottest != null) {
- return new AutoScaleDecision.Split(hottest.segmentId(), hottestReason);
+ // At the cap this scan is diagnostic only: baseline skipped splitting and could merge.
+ if (active.size() >= config.maxSegments()) {
+ return new AutoScaleDecision.Blocked("max_segments");
+ }
+ return hottest.hashRange().size() < 2
+ ? new AutoScaleDecision.Blocked("unsplittable")
+ : new AutoScaleDecision.Split(hottest.segmentId(), hottestReason);
}
return AutoScaleDecision.NONE;
@@ -278,6 +306,7 @@ public final class AutoScalePolicyEvaluator {
long usedParallelism = Math.min(consumers, totalCapacity);
AutoScaleDecision.Merge coldest = null;
+ AutoScaleDecision blocked = AutoScaleDecision.NONE;
double coldestCombined = Double.MAX_VALUE;
for (int i = 0; i < active.size(); i++) {
for (int j = i + 1; j < active.size(); j++) {
@@ -286,17 +315,21 @@ public final class AutoScalePolicyEvaluator {
if (!a.hashRange().isAdjacentTo(b.hashRange())) {
continue;
}
+ if (!coldEnough(a.segmentId(), loadBySegment, config, nowMs, mergeWindowMs)
+ || !coldEnough(b.segmentId(), loadBySegment, config, nowMs, mergeWindowMs)) {
+ continue;
+ }
if (layout.mergeDepth(a.segmentId()) >= config.maxDagDepth()
|| layout.mergeDepth(b.segmentId()) >= config.maxDagDepth()) {
+ blocked = new AutoScaleDecision.Blocked("max_dag_depth");
continue;
}
long pairCapacity = (long) a.bucketCount() + b.bucketCount();
long mergedCapacity = Math.min(pairCapacity, config.maxEntryBucketsPerSegment());
if (totalCapacity - pairCapacity + mergedCapacity < usedParallelism) {
- continue;
- }
- if (!coldEnough(a.segmentId(), loadBySegment, config, nowMs, mergeWindowMs)
- || !coldEnough(b.segmentId(), loadBySegment, config, nowMs, mergeWindowMs)) {
+ if (!(blocked instanceof AutoScaleDecision.Blocked)) {
+ blocked = new AutoScaleDecision.Blocked("max_entry_buckets");
+ }
continue;
}
double combined = combinedRate(a.segmentId(), loadBySegment)
@@ -307,7 +340,7 @@ public final class AutoScalePolicyEvaluator {
}
}
}
- return coldest != null ? coldest : AutoScaleDecision.NONE;
+ return coldest != null ? coldest : blocked;
}
/**
@@ -338,6 +371,15 @@ public final class AutoScalePolicyEvaluator {
// --- Helpers ---
+ private static boolean endsEvaluation(AutoScaleDecision decision) {
+ // Preserve the original split candidate's priority even when it would fail later.
+ return decision instanceof AutoScaleDecision.Split
+ || decision instanceof AutoScaleDecision.Rebucket
+ || decision instanceof AutoScaleDecision.Merge
+ || decision instanceof AutoScaleDecision.Blocked blocked
+ && "unsplittable".equals(blocked.reason());
+ }
+
private static boolean withinCooldown(long nowMs, long lastAtMs, long cooldownMs) {
return lastAtMs != Long.MIN_VALUE && nowMs - lastAtMs < cooldownMs;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
index 20c3d8017a0b..8a6e79716905 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
@@ -36,7 +36,9 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import lombok.Getter;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -96,6 +98,26 @@ public class ScalableTopicController {
private final Duration gcInterval;
private volatile SegmentLayout currentLayout;
+ private ScalableTopicControllerMetrics metrics = ScalableTopicControllerMetrics.NOOP;
+ /** Monotonic start of the election term, including takeover retry backoff. Guarded by this. */
+ private long takeoverStartedNanos;
+
+ void setMetrics(ScalableTopicControllerMetrics metrics) {
+ this.metrics = metrics;
+ }
+
+ /**
+ * Read volatile state and an immutable layout without acquiring the controller monitor: controller work
+ * must not block the shared Prometheus scrape or OpenTelemetry collection of unrelated metrics.
+ */
+ ScalableTopicControllerMetrics.Counts metricCounts() {
+ SegmentLayout layout = currentLayout;
+ if (closed || leaderState != LeaderElectionState.Leading || layout == null) {
+ return new ScalableTopicControllerMetrics.Counts(0, 0, 0);
+ }
+ long active = layout.getActiveSegments().size();
+ return new ScalableTopicControllerMetrics.Counts(1, active, layout.getAllSegments().size() - active);
+ }
private Consumer<SubscriptionCoordinator.RebalanceReason> rebalanceListener = __ -> { };
@@ -394,6 +416,7 @@ public class ScalableTopicController {
}
electionState = LeaderElectionState.Leading;
term = ++this.term;
+ takeoverStartedNanos = System.nanoTime();
leadership = new CompletableFuture<>();
takeOverBackoff.reset();
}
@@ -401,12 +424,23 @@ public class ScalableTopicController {
}
private void startTakeOver(long term) {
- if (!isCurrentTerm(term)) {
- return;
+ long startedNanos;
+ synchronized (this) {
+ if (!isCurrentTerm(term)) {
+ return;
+ }
+ startedNanos = takeoverStartedNanos;
}
// Not under the monitor: a takeover that finds the topic deleted closes this controller,
// and with it the leader election, whose monitor comes first.
- CompletableFuture<Void> takeover = takeOver(term);
+ CompletableFuture<Void> takeover = FutureUtil.supplySafely(() -> takeOver(term))
+ .whenComplete((__, ex) -> {
+ // Superseded or closed attempts are abandoned, not failed takeovers.
+ // Successful takeover is recorded at the actual transition to serving.
+ if (ex != null && isCurrentTerm(term)) {
+ metrics.recordTakeover(elapsedSeconds(startedNanos), false);
+ }
+ });
long retryDelayMs;
synchronized (this) {
if (term != this.term) {
@@ -472,6 +506,7 @@ public class ScalableTopicController {
private void serveAsLeader(long term) {
CompletableFuture<Void> served;
+ long startedNanos;
synchronized (this) {
if (closed || term != this.term) {
return;
@@ -480,7 +515,10 @@ public class ScalableTopicController {
scheduleGcTask();
scheduleAutoScaleTask();
served = leadership;
+ startedNanos = takeoverStartedNanos;
}
+ // Record the completed takeover before waiters can close or supersede the serving controller.
+ metrics.recordTakeover(elapsedSeconds(startedNanos), true);
// Not under the monitor: the requests that waited for the takeover go on from here.
served.complete(null);
}
@@ -763,6 +801,9 @@ public class ScalableTopicController {
try {
return AutoScaleConfig.resolve(brokerConfig, ns, topic);
} catch (IllegalArgumentException e) {
+ if (isLeader()) {
+ metrics.recordBlocked("invalid_policy");
+ }
log.warn().attr("reason", e.getMessage())
.log("Resolved auto split/merge policy is invalid; treating auto "
+ "split/merge as disabled for this topic until the namespace "
@@ -778,10 +819,16 @@ public class ScalableTopicController {
private CompletableFuture<Void> dispatch(AutoScaleDecision decision, AutoScaleConfig config,
String trigger) {
+ if (!isLeader() || closed) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (decision instanceof AutoScaleDecision.Blocked blocked) {
+ metrics.recordBlocked(blocked.reason());
+ }
if (decision instanceof AutoScaleDecision.Split split) {
log.info().attr("segmentId", split.segmentId()).attr("reason", split.reason())
.attr("trigger", trigger).log("Auto split");
- return splitSegment(split.segmentId())
+ return splitSegment(split.segmentId(), "auto")
.thenApply(__ -> {
scheduleFollowUpEvaluation(config);
return null;
@@ -798,7 +845,7 @@ public class ScalableTopicController {
CompletableFuture<Void> chain = CompletableFuture.completedFuture(null);
for (long segmentId : rebucket.segmentIds()) {
chain = chain.thenCompose(__ ->
- rebucketSegment(segmentId, rebucket.newBucketCount()))
+ rebucketSegment(segmentId, rebucket.newBucketCount(), "auto"))
.thenApply(__ -> null);
}
return chain.whenComplete((__, ___) ->
@@ -812,7 +859,7 @@ public class ScalableTopicController {
if (decision instanceof AutoScaleDecision.Merge merge) {
log.info().attr("segmentId1", merge.segmentId1()).attr("segmentId2", merge.segmentId2())
.attr("reason", merge.reason()).attr("trigger", trigger).log("Auto merge");
- return mergeSegments(merge.segmentId1(), merge.segmentId2()).thenApply(__ -> null);
+ return mergeSegments(merge.segmentId1(), merge.segmentId2(), "auto").thenApply(__ -> null);
}
return CompletableFuture.completedFuture(null);
}
@@ -1150,6 +1197,15 @@ public class ScalableTopicController {
* messages published before a consumer subscribes would be missed.
*/
public CompletableFuture<SegmentLayout> splitSegment(long segmentId) {
+ return splitSegment(segmentId, "manual");
+ }
+
+ private CompletableFuture<SegmentLayout> splitSegment(long segmentId, String trigger) {
+ LayoutOperation operation = new LayoutOperation("split", trigger);
+ return operation.track(() -> splitSegment(segmentId, operation));
+ }
+
+ private CompletableFuture<SegmentLayout> splitSegment(long segmentId, LayoutOperation operation) {
checkLeader();
// Single timestamp shared by the local preview and the CAS-retried metadata update,
@@ -1170,13 +1226,13 @@ public class ScalableTopicController {
// already provisioned (the create call routes to each segment's owning broker).
return resources.listSubscriptionsAsync(topicName)
.thenCompose(parentSubs -> {
- var subList = new java.util.ArrayList<>(parentSubs);
+ var subList = new ArrayList<>(parentSubs);
return createSegmentTopic(child1, subList)
.thenCompose(__ -> createSegmentTopic(child2, subList));
})
// Step 3: Terminate the parent segment topic so producers get TopicTerminated
- .thenCompose(__ -> terminateSegmentTopic(parentTopicName))
+ .thenCompose(__ -> terminateSegmentTopic(parentTopicName).thenRun(operation::sealed))
// Step 4: Atomic metadata update (only after topics + cursors are ready + parent terminated)
.thenCompose(__ -> resources.updateScalableTopicAsync(topicName, md -> {
@@ -1184,6 +1240,7 @@ public class ScalableTopicController {
SegmentLayout updated = latest.splitSegment(segmentId, nowMs);
return updated.toMetadata(md);
}))
+ .thenRun(operation::committed)
.thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true))
.thenCompose(optMd -> {
currentLayout = SegmentLayout.fromMetadata(optMd.orElseThrow());
@@ -1207,6 +1264,16 @@ public class ScalableTopicController {
* created before the parent is terminated, and the metadata CAS lands last.
*/
public CompletableFuture<SegmentLayout> rebucketSegment(long segmentId, int newBucketCount) {
+ return rebucketSegment(segmentId, newBucketCount, "manual");
+ }
+
+ private CompletableFuture<SegmentLayout> rebucketSegment(long segmentId, int newBucketCount, String trigger) {
+ LayoutOperation operation = new LayoutOperation("rebucket", trigger);
+ return operation.track(() -> rebucketSegment(segmentId, newBucketCount, operation));
+ }
+
+ private CompletableFuture<SegmentLayout> rebucketSegment(long segmentId, int newBucketCount,
+ LayoutOperation operation) {
checkLeader();
int maxBuckets = maxEntryBucketsPerSegment();
if (newBucketCount < 1 || newBucketCount > maxBuckets) {
@@ -1231,13 +1298,14 @@ public class ScalableTopicController {
return resources.listSubscriptionsAsync(topicName)
.thenCompose(parentSubs ->
- createSegmentTopic(successor, new java.util.ArrayList<>(parentSubs)))
- .thenCompose(__ -> terminateSegmentTopic(parentTopicName))
+ createSegmentTopic(successor, new ArrayList<>(parentSubs)))
+ .thenCompose(__ -> terminateSegmentTopic(parentTopicName).thenRun(operation::sealed))
.thenCompose(__ -> resources.updateScalableTopicAsync(topicName, md -> {
SegmentLayout latest = SegmentLayout.fromMetadata(md);
SegmentLayout updated = latest.rebucketSegment(segmentId, newSplits, nowMs);
return updated.toMetadata(md);
}))
+ .thenRun(operation::committed)
.thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true))
.thenCompose(optMd -> {
currentLayout = SegmentLayout.fromMetadata(optMd.orElseThrow());
@@ -1261,6 +1329,16 @@ public class ScalableTopicController {
* are created before the metadata update.
*/
public CompletableFuture<SegmentLayout> mergeSegments(long segmentId1, long segmentId2) {
+ return mergeSegments(segmentId1, segmentId2, "manual");
+ }
+
+ private CompletableFuture<SegmentLayout> mergeSegments(long segmentId1, long segmentId2, String trigger) {
+ LayoutOperation operation = new LayoutOperation("merge", trigger);
+ return operation.track(() -> mergeSegments(segmentId1, segmentId2, operation));
+ }
+
+ private CompletableFuture<SegmentLayout> mergeSegments(long segmentId1, long segmentId2,
+ LayoutOperation operation) {
checkLeader();
// Single timestamp shared by the local preview and the CAS-retried metadata
@@ -1280,11 +1358,11 @@ public class ScalableTopicController {
// truth, see splitSegment), then create the merged segment topic with those
// subscriptions provisioned.
return resources.listSubscriptionsAsync(topicName)
- .thenCompose(parentSubs -> createSegmentTopic(merged, new java.util.ArrayList<>(parentSubs)))
+ .thenCompose(parentSubs -> createSegmentTopic(merged, new ArrayList<>(parentSubs)))
// Step 2: Terminate both parent segment topics
- .thenCompose(__ -> terminateSegmentTopic(parent1Topic))
- .thenCompose(__ -> terminateSegmentTopic(parent2Topic))
+ .thenCompose(__ -> terminateSegmentTopic(parent1Topic).thenRun(operation::sealed))
+ .thenCompose(__ -> terminateSegmentTopic(parent2Topic).thenRun(operation::sealed))
// Step 3: Atomic metadata update (only after topic + cursors are ready + parents terminated)
.thenCompose(__ -> resources.updateScalableTopicAsync(topicName, md -> {
@@ -1293,6 +1371,7 @@ public class ScalableTopicController {
maxEntryBucketsPerSegment());
return updated.toMetadata(md);
}))
+ .thenRun(operation::committed)
.thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true))
.thenCompose(optMd -> {
currentLayout = SegmentLayout.fromMetadata(optMd.orElseThrow());
@@ -1303,6 +1382,59 @@ public class ScalableTopicController {
}).thenApply(__ -> currentLayout);
}
+ private static double elapsedSeconds(long startNanos) {
+ return (System.nanoTime() - startNanos) / 1_000_000_000.0;
+ }
+
+ /** Tracks the irreversible boundaries separately from notification/readback failures. */
+ private final class LayoutOperation {
+ private final String operation;
+ private final String trigger;
+ private final long startedNanos = System.nanoTime();
+ private final boolean startedAsLeader;
+ private long sealedNanos;
+ private boolean committed;
+ private final AtomicInteger pendingStorageDeletes = new AtomicInteger();
+ private boolean layoutChanged;
+
+ private LayoutOperation(String operation, String trigger) {
+ this.operation = operation;
+ this.trigger = trigger;
+ startedAsLeader = isLeader();
+ }
+
+ private void sealed() {
+ if (sealedNanos == 0) {
+ sealedNanos = System.nanoTime();
+ }
+ }
+
+ private void committed() {
+ committed = true;
+ if (startedAsLeader && sealedNanos != 0) {
+ metrics.recordSeal(operation, elapsedSeconds(sealedNanos));
+ }
+ }
+
+ private <T> CompletableFuture<T> track(Supplier<CompletableFuture<T>> action) {
+ return FutureUtil.supplySafely(action).whenComplete((__, ex) -> {
+ // Outcomes belong to the broker that started the operation, even after leadership changes.
+ // Requests rejected before starting as leader still contribute nothing.
+ if (!startedAsLeader) {
+ return;
+ }
+ if ("prune".equals(operation) && committed && !layoutChanged) {
+ return;
+ }
+ boolean partial = "prune".equals(operation)
+ ? committed && pendingStorageDeletes.get() > 0
+ : sealedNanos != 0 && !committed;
+ String status = partial ? "partial" : ex == null ? "success" : "failure";
+ metrics.recordOperation(operation, trigger, status, elapsedSeconds(startedNanos));
+ });
+ }
+ }
+
// --- Consumer management ---
/**
@@ -1837,6 +1969,12 @@ public class ScalableTopicController {
if (!isLeader() || closed) {
return CompletableFuture.completedFuture(null);
}
+ LayoutOperation operation = new LayoutOperation("prune", "auto");
+ operation.pendingStorageDeletes.set(drained.size());
+ return operation.track(() -> pruneAllAsync(drained, operation));
+ }
+
+ private CompletableFuture<Void> pruneAllAsync(List<SegmentInfo> drained, LayoutOperation operation) {
for (SegmentInfo s : drained) {
log.info().attr("segmentId", s.segmentId())
.attr("sealedAtMs", s.sealedAtMs())
@@ -1852,15 +1990,17 @@ public class ScalableTopicController {
updated = updated.pruneSegment(s.segmentId());
}
}
+ operation.layoutChanged = updated != latest;
return updated == latest ? md : updated.toMetadata(md);
- }).thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true))
+ }).thenRun(operation::committed)
+ .thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true))
.thenCompose(optMd -> {
currentLayout = SegmentLayout.fromMetadata(optMd.orElseThrow());
return notifySubscriptions(currentLayout);
})
.thenCompose(__ -> {
CompletableFuture<?>[] deletes = drained.stream()
- .map(s -> deleteSegmentBackingTopic(s)
+ .map(s -> deleteSegmentBackingTopic(s, operation)
// The segment is gone from the layout — drop its load record
// too, or the .../segments/{id}/load entry leaks forever.
.thenCompose(___ ->
@@ -1915,20 +2055,21 @@ public class ScalableTopicController {
* has already pruned the segment from the layout (the point of no return),
* so a failed delete is just leaked storage that the next tick will retry.
*/
- private CompletableFuture<Void> deleteSegmentBackingTopic(SegmentInfo seg) {
+ private CompletableFuture<Void> deleteSegmentBackingTopic(SegmentInfo seg, LayoutOperation operation) {
String name = SegmentTopicName.backingTopicName(topicName, seg);
try {
PulsarAdmin admin = brokerService.getPulsar().getAdminClient();
CompletableFuture<Void> delete = seg.isLegacy()
? admin.topics().deleteAsync(name, /* force */ true)
: admin.scalableTopics().deleteSegmentAsync(name, /* force */ true);
- return delete
+ return delete.thenRun(() -> operation.pendingStorageDeletes.decrementAndGet())
.exceptionally(ex -> {
Throwable cause =
org.apache.pulsar.common.util.FutureUtil.unwrapCompletionException(ex);
if (cause instanceof org.apache.pulsar.client.admin.PulsarAdminException
.NotFoundException) {
// Already gone — fine.
+ operation.pendingStorageDeletes.decrementAndGet();
return null;
}
log.warn().attr("segment", name).exceptionMessage(cause)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetrics.java
new file mode 100644
index 000000000000..5090ff77a9ef
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetrics.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.pulsar.broker.service.scalable;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.BatchCallback;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.api.metrics.ObservableLongMeasurement;
+import io.prometheus.client.Collector;
+import io.prometheus.client.Collector.MetricFamilySamples;
+import io.prometheus.client.Collector.MetricFamilySamples.Sample;
+import io.prometheus.client.Collector.Type;
+import io.prometheus.client.CollectorRegistry;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.DoubleAdder;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.Supplier;
+import lombok.CustomLog;
+
+/**
+ * Broker-level controller metrics, with shared state for Prometheus and OpenTelemetry.
+ *
+ * <p>The ordinary deployment scope is one broker SDK {@link Meter}, one Prometheus registry and one cluster.
+ * Real SDK histogram history is Meter-wide and survives provider recreation. Sharing that Meter across distinct
+ * registries or clusters does not provide independent histogram isolation: each registry with a live provider
+ * exposes the Meter's shared history, including its other clusters. OpenTelemetry cluster scope comes from the
+ * broker SDK resource, not an instrument attribute. Use separate SDK Meters for independent scopes.
+ *
+ * <p>Embedded providers sharing a registry and cluster aggregate counters from live providers only. Membership
+ * changes, including partial close or recreation, may lower or reset cumulative counter series, even though the
+ * OpenTelemetry observers report monotonic sums. This is not general monotonic multi-broker aggregation.
+ * With the API no-op Meter, Prometheus histogram history is provider-local instead of Meter-wide.
+ */
+@CustomLog
+public final class ScalableTopicControllerMetrics implements AutoCloseable {
+ private static final String OTEL_PREFIX = "pulsar.broker.scalable_topic.";
+ private static final String ATTRIBUTE_PREFIX = "pulsar.scalable_topic.";
+ private static final Meter NOOP_METER = MeterProvider.noop().get(ScalableTopicControllerMetrics.class.getName());
+
+ /** Shared explicit histogram boundaries in seconds, from 1 ms through 5 minutes. */
+ static final List<Double> DURATION_BUCKETS =
+ List.of(0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 300.0);
+
+ private static final Set<String> OPERATIONS = Set.of("split", "merge", "rebucket", "prune");
+ private static final Set<String> TRIGGERS = Set.of("manual", "auto");
+ private static final Set<String> STATUSES = Set.of("success", "failure", "partial");
+ private static final Set<String> BLOCKED_REASONS = Set.of("max_segments", "max_entry_buckets", "max_dag_depth",
+ "unsplittable", "invalid_policy");
+
+ // A registry can serve several in-JVM brokers. One collector merges identical series, without broker labels.
+ // Likewise, one callback per Meter avoids duplicate observations when services share an SDK Meter.
+ private static final Map<CollectorRegistry, RegistryCollector> COLLECTORS = new IdentityHashMap<>();
+ private static final Map<Meter, MeterCallbacks> CALLBACKS = new IdentityHashMap<>();
+ // SDK histograms outlive service recreation. Values must not reference their weak Meter keys.
+ private static final Map<Meter, Map<HistogramKey, Histogram>> HISTOGRAMS = new WeakHashMap<>();
+
+ public static final ScalableTopicControllerMetrics NOOP = new ScalableTopicControllerMetrics();
+
+ public record Counts(long controllers, long active, long sealed) { }
+
+ private enum Metric {
+ CONTROLLERS("controller.count", Type.GAUGE),
+ TAKEOVER("controller.takeover.duration", Type.HISTOGRAM, "operation.status"),
+ SEGMENTS("segment.count", Type.GAUGE, "segment.state"),
+ OPERATIONS("layout.operation.count", Type.COUNTER, "operation", "operation.trigger", "operation.status"),
+ OPERATION_DURATION("layout.operation.duration", Type.HISTOGRAM, "operation", "operation.status"),
+ SEAL_DURATION("layout.seal.duration", Type.HISTOGRAM, "operation"),
+ BLOCKED("auto_scale.blocked.count", Type.COUNTER, "auto_scale.reason");
+
+ private final String name;
+ private final String prometheusName;
+ private final Type type;
+ private final List<String> attributes;
+
+ Metric(String name, Type type, String... attributes) {
+ this.name = name;
+ this.type = type;
+ this.attributes = List.of(attributes);
+ String suffix = type == Type.COUNTER ? name.substring(0, name.length() - ".count".length()) + ".total"
+ : name;
+ this.prometheusName = "pulsar_scalable_topic_" + suffix.replace('.', '_')
+ + (type == Type.HISTOGRAM ? "_seconds" : "");
+ }
+
+ private MetricFamilySamples family(List<Sample> samples) {
+ return new MetricFamilySamples(prometheusName, type, "Scalable topic " + name, samples);
+ }
+ }
+
+ private record Point(Metric metric, Attributes attributes) { }
+
+ private record SampleKey(String name, List<String> labelNames, List<String> labelValues) { }
+
+ private record HistogramKey(String cluster, Point point) { }
+
+ private final String cluster;
+ private final Supplier<Counts> counts;
+ private final CollectorRegistry registry;
+ private final Meter meter;
+ private final Map<Point, LongAdder> counters = new ConcurrentHashMap<>();
+ private final Map<HistogramKey, Histogram> histograms;
+ private final Map<Metric, DoubleHistogram> otelHistograms = new HashMap<>();
+ private final AtomicBoolean closed = new AtomicBoolean();
+
+ private ScalableTopicControllerMetrics() {
+ cluster = null;
+ counts = null;
+ registry = null;
+ meter = null;
+ histograms = null;
+ closed.set(true);
+ }
+
+ /**
+ * Creates metrics for the ordinary one-broker SDK Meter, default registry and cluster scope.
+ * Real SDK histogram history is Meter-wide, not independently isolated by distinct registries or clusters;
+ * use separate SDK Meters for independent scopes. Embedded providers sharing the registry and cluster aggregate
+ * live-provider counters only: membership changes or partial close may lower or reset cumulative series,
+ * rather than provide general monotonic multi-broker aggregation.
+ */
+ public ScalableTopicControllerMetrics(Meter meter, String cluster, Supplier<Counts> counts) {
+ this(meter, cluster, counts, CollectorRegistry.defaultRegistry);
+ }
+
+ /**
+ * Creates metrics with an explicit registry for embedded deployments and tests; the ordinary deployment
+ * scope remains one broker SDK Meter, registry and cluster. A distinct registry or cluster does not isolate
+ * real SDK Meter-wide histogram history; use separate SDK Meters for independent scopes. Providers sharing a
+ * registry and cluster contribute live-provider counters, so membership changes or partial close may lower or
+ * reset cumulative series. This is not general monotonic multi-broker aggregation.
+ */
+ ScalableTopicControllerMetrics(Meter meter, String cluster, Supplier<Counts> counts, CollectorRegistry registry) {
+ this.meter = Objects.requireNonNull(meter);
+ this.cluster = Objects.requireNonNull(cluster);
+ this.counts = Objects.requireNonNull(counts);
+ this.registry = Objects.requireNonNull(registry);
+ if (meter == NOOP_METER) {
+ // The API no-op Meter is shared across brokers and retains no SDK observations.
+ // Its Prometheus histograms must belong to this provider, not the singleton Meter's lifetime.
+ histograms = new ConcurrentHashMap<>();
+ } else {
+ synchronized (HISTOGRAMS) {
+ histograms = HISTOGRAMS.computeIfAbsent(meter, ignored -> new ConcurrentHashMap<>());
+ }
+ }
+ for (Metric metric : List.of(Metric.TAKEOVER, Metric.OPERATION_DURATION, Metric.SEAL_DURATION)) {
+ otelHistograms.put(metric, meter.histogramBuilder(OTEL_PREFIX + metric.name)
+ .setUnit("s").setDescription("Scalable topic " + metric.name)
+ .setExplicitBucketBoundariesAdvice(DURATION_BUCKETS).build());
+ }
+ synchronized (COLLECTORS) {
+ RegistryCollector existing = COLLECTORS.get(registry);
+ if (existing == null) {
+ existing = new RegistryCollector();
+ registry.register(existing);
+ COLLECTORS.put(registry, existing);
+ }
+ existing.providers.add(this);
+ }
+ try {
+ synchronized (CALLBACKS) {
+ CALLBACKS.computeIfAbsent(meter, MeterCallbacks::new).providers.add(this);
+ }
+ } catch (RuntimeException e) {
+ // Callback setup failed after publishing this provider: remove only its registry ownership.
+ synchronized (COLLECTORS) {
+ RegistryCollector collector = COLLECTORS.get(registry);
+ collector.providers.remove(this);
+ if (collector.providers.isEmpty()) {
+ registry.unregister(collector);
+ COLLECTORS.remove(registry);
+ }
+ }
+ throw e;
+ }
+ }
+
+ public void recordTakeover(double seconds, boolean success) {
+ if (!closed.get()) {
+ recordHistogram(Metric.TAKEOVER, seconds, attributes("operation.status", success ? "success" : "failure"));
+ }
+ }
+
+ public void recordOperation(String operation, String trigger, String status, double seconds) {
+ if (!closed.get()) {
+ validate(operation, OPERATIONS);
+ validate(trigger, TRIGGERS);
+ validate(status, STATUSES);
+ validateDuration(seconds);
+ counters.computeIfAbsent(new Point(Metric.OPERATIONS, attributes("operation", operation,
+ "operation.trigger", trigger, "operation.status", status)), ignored -> new LongAdder()).increment();
+ recordHistogram(Metric.OPERATION_DURATION, seconds,
+ attributes("operation", operation, "operation.status", status));
+ }
+ }
+
+ public void recordSeal(String operation, double seconds) {
+ if (!closed.get()) {
+ validate(operation, Set.of("split", "merge", "rebucket"));
+ recordHistogram(Metric.SEAL_DURATION, seconds, attributes("operation", operation));
+ }
+ }
+
+ public void recordBlocked(String reason) {
+ if (!closed.get()) {
+ validate(reason, BLOCKED_REASONS);
+ counters.computeIfAbsent(new Point(Metric.BLOCKED, attributes("auto_scale.reason", reason)),
+ ignored -> new LongAdder()).increment();
+ }
+ }
+
+ private static void validate(String value, Set<String> allowed) {
+ if (value == null || !allowed.contains(value)) {
+ throw new IllegalArgumentException("Invalid scalable topic metric label: " + value);
+ }
+ }
+
+ private static void validateDuration(double seconds) {
+ if (!Double.isFinite(seconds) || seconds < 0) {
+ throw new IllegalArgumentException("Duration must be finite and nonnegative");
+ }
+ }
+
+ private static Attributes attributes(String... pairs) {
+ var builder = Attributes.builder();
+ for (int i = 0; i < pairs.length; i += 2) {
+ builder.put(ATTRIBUTE_PREFIX + pairs[i], pairs[i + 1]);
+ }
+ return builder.build();
+ }
+
+ private void recordHistogram(Metric metric, double seconds, Attributes attributes) {
+ validateDuration(seconds);
+ histograms.computeIfAbsent(new HistogramKey(cluster, new Point(metric, attributes)),
+ ignored -> new Histogram()).record(seconds);
+ otelHistograms.get(metric).record(seconds, attributes);
+ }
+
+ private void collectState(Map<Point, Long> values) {
+ try {
+ Counts current = Objects.requireNonNull(counts.get());
+ values.merge(new Point(Metric.CONTROLLERS, Attributes.empty()), current.controllers(), Long::sum);
+ values.merge(new Point(Metric.SEGMENTS, attributes("segment.state", "active")),
+ current.active(), Long::sum);
+ values.merge(new Point(Metric.SEGMENTS, attributes("segment.state", "sealed")),
+ current.sealed(), Long::sum);
+ } catch (RuntimeException e) {
+ // A failed new state read must not fail the shared scrape or suppress other providers' metrics.
+ log.debug().exceptionMessage(e).log("Failed to collect scalable topic controller counts");
+ }
+ counters.forEach((point, value) -> values.merge(point, value.sum(), Long::sum));
+ }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ synchronized (COLLECTORS) {
+ RegistryCollector collector = COLLECTORS.get(registry);
+ collector.providers.remove(this);
+ if (collector.providers.isEmpty()) {
+ registry.unregister(collector);
+ COLLECTORS.remove(registry);
+ }
+ }
+ synchronized (CALLBACKS) {
+ MeterCallbacks callbacks = CALLBACKS.get(meter);
+ callbacks.providers.remove(this);
+ if (callbacks.providers.isEmpty()) {
+ callbacks.callback.close();
+ CALLBACKS.remove(meter);
+ }
+ }
+ }
+
+ private static final class Histogram {
+ private final LongAdder[] buckets = new LongAdder[DURATION_BUCKETS.size() + 1];
+ private final DoubleAdder sum = new DoubleAdder();
+
+ private Histogram() {
+ for (int i = 0; i < buckets.length; i++) {
+ buckets[i] = new LongAdder();
+ }
+ }
+
+ private void record(double seconds) {
+ int bucket = 0;
+ while (bucket < DURATION_BUCKETS.size() && seconds > DURATION_BUCKETS.get(bucket)) {
+ bucket++;
+ }
+ buckets[bucket].increment();
+ sum.add(seconds);
+ }
+ }
+
+ private static final class MeterCallbacks {
+ private final Set<ScalableTopicControllerMetrics> providers = ConcurrentHashMap.newKeySet();
+ private final BatchCallback callback;
+
+ private MeterCallbacks(Meter meter) {
+ Map<Metric, ObservableLongMeasurement> observers = new HashMap<>();
+ observers.put(Metric.CONTROLLERS, meter.upDownCounterBuilder(OTEL_PREFIX + Metric.CONTROLLERS.name)
+ .setUnit("{controller}").buildObserver());
+ observers.put(Metric.SEGMENTS, meter.upDownCounterBuilder(OTEL_PREFIX + Metric.SEGMENTS.name)
+ .setUnit("{segment}").buildObserver());
+ observers.put(Metric.OPERATIONS, meter.counterBuilder(OTEL_PREFIX + Metric.OPERATIONS.name)
+ .setUnit("{operation}").buildObserver());
+ observers.put(Metric.BLOCKED, meter.counterBuilder(OTEL_PREFIX + Metric.BLOCKED.name)
+ .setUnit("{evaluation}").buildObserver());
+ callback = meter.batchCallback(() -> {
+ Map<Point, Long> values = new HashMap<>();
+ providers.forEach(provider -> provider.collectState(values));
+ values.forEach((point, value) -> observers.get(point.metric).record(value, point.attributes));
+ }, observers.get(Metric.CONTROLLERS), observers.get(Metric.SEGMENTS),
+ observers.get(Metric.OPERATIONS), observers.get(Metric.BLOCKED));
+ }
+ }
+
+ /**
+ * Merges live-provider state and counters by cluster in one registry, without general monotonic multi-broker
+ * guarantees: provider membership changes or partial close can lower or reset counters. The ordinary scope is
+ * one broker SDK Meter, registry and cluster. Each real SDK Meter's histogram history is included once while
+ * any provider using it remains in this registry, including history recorded through other registries or clusters.
+ * Distinct registries or clusters sharing that Meter are not independent histogram scopes; use separate SDK
+ * Meters for isolation. API no-op Meter histograms instead follow each provider's lifetime.
+ */
+ private static final class RegistryCollector extends Collector implements Collector.Describable {
+ private final Set<ScalableTopicControllerMetrics> providers = ConcurrentHashMap.newKeySet();
+
+ @Override
+ public List<MetricFamilySamples> describe() {
+ List<MetricFamilySamples> result = new ArrayList<>();
+ for (Metric metric : Metric.values()) {
+ result.add(metric.family(List.of()));
+ }
+ return result;
+ }
+
+ @Override
+ public List<MetricFamilySamples> collect() {
+ Map<Metric, Map<SampleKey, Double>> families = new LinkedHashMap<>();
+ Set<Map<HistogramKey, Histogram>> histograms = Collections.newSetFromMap(new IdentityHashMap<>());
+ for (ScalableTopicControllerMetrics provider : providers) {
+ Map<Point, Long> values = new HashMap<>();
+ provider.collectState(values);
+ values.forEach((point, value) -> add(families, point, provider.cluster, "", null, value));
+ histograms.add(provider.histograms);
+ }
+ histograms.forEach(state -> state.forEach((key, histogram) -> {
+ long count = 0;
+ for (int i = 0; i < histogram.buckets.length; i++) {
+ count += histogram.buckets[i].sum();
+ String bound = i < DURATION_BUCKETS.size()
+ ? doubleToGoString(DURATION_BUCKETS.get(i)) : "+Inf";
+ add(families, key.point, key.cluster, "_bucket", bound, count);
+ }
+ add(families, key.point, key.cluster, "_count", null, count);
+ add(families, key.point, key.cluster, "_sum", null, histogram.sum.sum());
+ }));
+ List<MetricFamilySamples> result = new ArrayList<>();
+ families.forEach((metric, values) -> {
+ List<Sample> samples = new ArrayList<>();
+ values.forEach((key, value) ->
+ samples.add(new Sample(key.name, key.labelNames, key.labelValues, value)));
+ result.add(metric.family(samples));
+ });
+ return result;
+ }
+
+ private static void add(Map<Metric, Map<SampleKey, Double>> families, Point point, String cluster,
+ String suffix, String bound, double value) {
+ List<String> names = new ArrayList<>(List.of("cluster"));
+ List<String> values = new ArrayList<>(List.of(cluster));
+ for (String attribute : point.metric.attributes) {
+ names.add(attribute.replace('.', '_'));
+ values.add(point.attributes.get(AttributeKey.stringKey(ATTRIBUTE_PREFIX + attribute)));
+ }
+ if (bound != null) {
+ names.add("le");
+ values.add(bound);
+ }
+ SampleKey key = new SampleKey(point.metric.prometheusName + suffix, names, values);
+ families.computeIfAbsent(point.metric, ignored -> new LinkedHashMap<>()).merge(key, value, Double::sum);
+ }
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
index 39ac9767ee50..e930db40aa3b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
@@ -18,10 +18,12 @@
*/
package org.apache.pulsar.broker.service.scalable;
+import io.opentelemetry.api.OpenTelemetry;
import io.prometheus.client.CollectorRegistry;
import java.time.Duration;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import lombok.CustomLog;
@@ -68,6 +70,7 @@ public class ScalableTopicService {
private final BrokerService brokerService;
private final ScalableTopicResources resources;
private final CoordinationService coordinationService;
+ private final ScalableTopicControllerMetrics metrics;
/**
* Active controllers for topics this broker coordinates. The value is a future so
@@ -83,6 +86,38 @@ public class ScalableTopicService {
this.brokerService = brokerService;
this.resources = resources;
this.coordinationService = coordinationService;
+ this.metrics = createControllerMetrics();
+ }
+
+ private ScalableTopicControllerMetrics createControllerMetrics() {
+ try {
+ var pulsar = brokerService.getPulsar();
+ var telemetry = pulsar.getOpenTelemetry();
+ var config = pulsar.getConfiguration();
+ return new ScalableTopicControllerMetrics(telemetry != null ? telemetry.getMeter()
+ : OpenTelemetry.noop().getMeter(ScalableTopicService.class.getName()),
+ config != null ? Objects.toString(config.getClusterName(), "") : "", this::metricCounts);
+ } catch (RuntimeException e) {
+ log.warn().exceptionMessage(e)
+ .log("Failed to initialize scalable-topic controller metrics; continuing without them");
+ return ScalableTopicControllerMetrics.NOOP;
+ }
+ }
+
+ private ScalableTopicControllerMetrics.Counts metricCounts() {
+ long leaders = 0;
+ long active = 0;
+ long sealed = 0;
+ for (CompletableFuture<ScalableTopicController> future : controllers.values()) {
+ if (!future.isDone() || future.isCompletedExceptionally()) {
+ continue;
+ }
+ ScalableTopicControllerMetrics.Counts counts = future.join().metricCounts();
+ leaders += counts.controllers();
+ active += counts.active();
+ sealed += counts.sealed();
+ }
+ return new ScalableTopicControllerMetrics.Counts(leaders, active, sealed);
}
private volatile ScalableTopicConsumerMetrics consumerMetrics;
@@ -104,11 +139,12 @@ public class ScalableTopicService {
log.info("ScalableTopicService started");
}
- /** Release only this service's new instruments, without changing controller teardown. */
+ /** Release only this service's instruments, without changing controller or metadata-store lifecycles. */
public void closeMetrics() {
if (consumerMetrics != null) {
consumerMetrics.close();
}
+ metrics.close();
}
public void close() {
@@ -165,6 +201,7 @@ public class ScalableTopicService {
private CompletableFuture<ScalableTopicController> newController(String key, TopicName topic) {
ScalableTopicController controller = new ScalableTopicController(
topic, resources, brokerService, coordinationService);
+ controller.setMetrics(metrics);
if (consumerMetrics != null) {
controller.setRebalanceListener(consumerMetrics::recordRebalance);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java
index 4ca53f290afd..99a6a5b8c6fa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service.scalable;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import java.time.Duration;
@@ -25,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentLoadStats;
import org.testng.annotations.Test;
@@ -228,7 +230,7 @@ public class AutoScalePolicyEvaluatorTest {
ScalableTopicController.createInitialMetadata(1, 8, Map.of()));
AutoScaleDecision none = decide(maxed, load, Map.of("sub", 5_000),
baseConfig().maxEntryBucketsPerSegment(8).build());
- assertTrue(none instanceof AutoScaleDecision.NoAction, none.toString());
+ assertThat(none).isEqualTo(new AutoScaleDecision.Blocked("max_entry_buckets"));
}
@Test
@@ -385,7 +387,7 @@ public class AutoScalePolicyEvaluatorTest {
Map<Long, SegmentLoadSample> load = Map.of(0L, cold(old()), 1L, cold(old()));
AutoScaleDecision d = decide(layout, load, Map.of("sub", 12),
baseConfig().maxEntryBucketsPerSegment(8).build());
- assertTrue(d instanceof AutoScaleDecision.NoAction, d.toString());
+ assertThat(d).isEqualTo(new AutoScaleDecision.Blocked("max_entry_buckets"));
}
@Test
@@ -427,7 +429,7 @@ public class AutoScalePolicyEvaluatorTest {
Map<Long, SegmentLoadSample> load = Map.of(0L, cold(old()), 1L, cold(old()));
AutoScaleDecision d = decide(layout, load, Map.of("sub", 12), Set.of(),
baseConfig().maxEntryBucketsPerSegment(8).build());
- assertTrue(d instanceof AutoScaleDecision.NoAction, d.toString());
+ assertThat(d).isEqualTo(new AutoScaleDecision.Blocked("max_entry_buckets"));
}
@Test
@@ -510,7 +512,7 @@ public class AutoScalePolicyEvaluatorTest {
// maxDagDepth=1: both at the cap → no merge.
AutoScaleDecision blocked = decide(layout, load, Map.of(),
baseConfig().maxDagDepth(1).build());
- assertTrue(blocked instanceof AutoScaleDecision.NoAction, "merge blocked at maxDagDepth");
+ assertThat(blocked).isEqualTo(new AutoScaleDecision.Blocked("max_dag_depth"));
// maxDagDepth=2: under the cap → merge allowed.
AutoScaleDecision allowed = decide(layout, load, Map.of(),
@@ -536,6 +538,254 @@ public class AutoScalePolicyEvaluatorTest {
"should pick the coldest adjacent pair {2,3}, got " + m);
}
+ // --- blocked evaluations and baseline action priority ---
+
+ private static SegmentLayout singleSlotLayout() {
+ return new SegmentLayout(0, 3, Map.of(
+ 0L, SegmentInfo.active(0, HashRange.of(0, 0), 0, NOW),
+ 1L, SegmentInfo.active(1, HashRange.of(1, 32767), 0, NOW),
+ 2L, SegmentInfo.active(2, HashRange.of(32768, HashRange.MAX_HASH), 0, NOW)));
+ }
+
+ @Test
+ public void testHotTopicAtSegmentCapIsBlocked() {
+ AutoScaleDecision decision = decide(initialLayout(1),
+ Map.of(0L, sample(SPLIT_MSG_IN + 1, 0, 0, 0, old())), Map.of(),
+ baseConfig().maxSegments(1).build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("max_segments"));
+ }
+
+ @Test
+ public void testCappedIdleTopicIsNotBlocked() {
+ // All three ceilings are reached, but no split, rebucket, or mature merge is demanded.
+ AutoScaleDecision decision = decide(initialLayout(2),
+ Map.of(0L, cold(0), 1L, cold(0)), Map.of(),
+ baseConfig().maxSegments(2).maxEntryBucketsPerSegment(2).maxDagDepth(0).build());
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testSplitCooldownIsNotBlockedEvenAtSegmentCap() {
+ AutoScaleDecision decision = AutoScalePolicyEvaluator.decide(initialLayout(1),
+ Map.of(0L, sample(SPLIT_MSG_IN + 1, 0, 0, 0, old())), Map.of(), Set.of(),
+ baseConfig().maxSegments(1).build(), NOW, NOW - 1, NO_PRIOR, NO_PRIOR);
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testRebucketCooldownIsNotBlockedEvenAtBucketCap() {
+ AutoScaleDecision decision = AutoScalePolicyEvaluator.decide(initialLayout(1),
+ Map.of(0L, cold(0)), Map.of("sub", 5), Set.of("sub"),
+ baseConfig().maxEntryBucketsPerSegment(4).build(), NOW, NO_PRIOR, NO_PRIOR, NOW - 1);
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testMergeCooldownIsNotBlockedEvenAtDepthCap() {
+ AutoScaleDecision decision = AutoScalePolicyEvaluator.decide(initialLayout(2),
+ Map.of(0L, cold(old()), 1L, cold(old())), Map.of(), Set.of(),
+ baseConfig().maxDagDepth(0).build(), NOW, NO_PRIOR, NOW - 1, NO_PRIOR);
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testDepthCapRequiresOtherwiseEligibleMerge() {
+ SegmentLayout layout = initialLayout(2);
+ AutoScaleConfig config = baseConfig().maxDagDepth(0).build();
+ assertThat(decide(layout, Map.of(), Map.of(), config)).isEqualTo(AutoScaleDecision.NONE);
+ assertThat(decide(layout, Map.of(0L, cold(0), 1L, cold(0)), Map.of(), config))
+ .isEqualTo(AutoScaleDecision.NONE);
+ assertThat(decide(layout, Map.of(0L, sample(MERGE_MSG_IN + 1, 0, 0, 0, old()),
+ 1L, cold(old())), Map.of(), config)).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testCheckpointDemandAtSegmentCapRequiresSplitFloor() {
+ AutoScaleConfig config = baseConfig().maxSegments(1).build();
+ assertThat(decide(initialLayout(1), Map.of(0L, cold(0)), Map.of("group", 2), Set.of(), config))
+ .isEqualTo(AutoScaleDecision.NONE);
+ assertThat(decide(initialLayout(1), Map.of(0L, sample(2_000, 0, 0, 0, 0)),
+ Map.of("group", 2), Set.of(), config))
+ .isEqualTo(new AutoScaleDecision.Blocked("max_segments"));
+ }
+
+ @Test
+ public void testLoadSplitKeepsHottestSingleSlotCandidate() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(SPLIT_MSG_IN * 10, 0, 0, 0, 0),
+ 1L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0)), Map.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testLoadSplitKeepsValidHottestCandidate() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0),
+ 1L, sample(SPLIT_MSG_IN * 10, 0, 0, 0, 0)), Map.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Split(1, "msgRateIn"));
+ }
+
+ @Test
+ public void testConsumerSplitKeepsBusiestSingleSlotCandidate() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(4_000, 0, 0, 0, 0),
+ 1L, sample(2_000, 0, 0, 0, 0)), Map.of("group", 4), Set.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testConsumerSplitKeepsValidBusiestCandidate() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(2_000, 0, 0, 0, 0),
+ 1L, sample(4_000, 0, 0, 0, 0)), Map.of("group", 4), Set.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Split(1, "consumer-count"));
+ }
+
+ @Test
+ public void testOnlyHotSingleSlotSegmentIsBlocked() {
+ AutoScaleDecision decision = decide(singleSlotLayout(),
+ Map.of(0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0)), Map.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testUnsplittableConsumerDemandIsBlocked() {
+ AutoScaleDecision decision = decide(singleSlotLayout(),
+ Map.of(0L, sample(2_000, 0, 0, 0, 0)), Map.of("group", 4), Set.of(),
+ baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testUnsplittableConsumerDemandDoesNotRebucketInstead() {
+ AutoScaleDecision decision = decide(singleSlotLayout(),
+ Map.of(0L, sample(2_000, 0, 0, 0, 0)), Map.of("sub", 4), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testUnsplittableConsumerDemandDoesNotSelectTrafficSplit() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(2_000, 0, 0, 0, 0),
+ 1L, sample(0, 0, 0, SPLIT_BYTES_OUT * 2, 0)),
+ Map.of("group", 4), Set.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testBlockedSplitDoesNotSuppressColdMerge() {
+ SegmentLayout layout = initialLayout(3);
+ AutoScaleDecision decision = decide(layout, Map.of(
+ 0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, old()),
+ 1L, cold(old()), 2L, cold(old())), Map.of(), baseConfig().maxSegments(3).build());
+ assertThat(decision).isInstanceOf(AutoScaleDecision.Merge.class);
+ AutoScaleDecision.Merge merge = (AutoScaleDecision.Merge) decision;
+ assertThat(Set.of(merge.segmentId1(), merge.segmentId2())).isEqualTo(Set.of(1L, 2L));
+ }
+
+ @Test
+ public void testUnsplittableTrafficSplitDoesNotSelectColdMerge() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, old()),
+ 1L, cold(old()), 2L, cold(old())), Map.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testUnsplittableConsumerSplitDoesNotSelectColdMerge() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(2_000, 0, 0, 0, old()),
+ 1L, cold(old()), 2L, cold(old())), Map.of("group", 4), Set.of(), baseConfig().build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Blocked("unsplittable"));
+ }
+
+ @Test
+ public void testSingleSlotAtSegmentCapReportsCap() {
+ SegmentLayout layout = singleSlotLayout();
+ AutoScaleConfig config = baseConfig().maxSegments(3).build();
+ assertThat(decide(layout, Map.of(0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0)), Map.of(), config))
+ .isEqualTo(new AutoScaleDecision.Blocked("max_segments"));
+ assertThat(decide(layout, Map.of(0L, sample(2_000, 0, 0, 0, 0)),
+ Map.of("group", 4), Set.of(), config))
+ .isEqualTo(new AutoScaleDecision.Blocked("max_segments"));
+ }
+
+ @Test
+ public void testSingleSlotAtSegmentCapDoesNotSuppressTrafficLaneMerge() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, old()),
+ 1L, cold(old()), 2L, cold(old())), Map.of(), baseConfig().maxSegments(3).build());
+ assertThat(decision).isInstanceOf(AutoScaleDecision.Merge.class);
+ AutoScaleDecision.Merge merge = (AutoScaleDecision.Merge) decision;
+ assertThat(Set.of(merge.segmentId1(), merge.segmentId2())).isEqualTo(Set.of(1L, 2L));
+ }
+
+ @Test
+ public void testSingleSlotAtSegmentCapDoesNotSuppressConsumerLaneMerge() {
+ AutoScaleDecision decision = decide(singleSlotLayout(), Map.of(
+ 0L, sample(2_000, 0, 0, 0, old()),
+ 1L, cold(old()), 2L, cold(old())), Map.of("group", 4), Set.of(),
+ baseConfig().maxSegments(3).build());
+ assertThat(decision).isInstanceOf(AutoScaleDecision.Merge.class);
+ AutoScaleDecision.Merge merge = (AutoScaleDecision.Merge) decision;
+ assertThat(Set.of(merge.segmentId1(), merge.segmentId2())).isEqualTo(Set.of(1L, 2L));
+ }
+
+ @Test
+ public void testSingleSlotAtSegmentCapCanStillRebucket() {
+ AutoScaleDecision decision = decide(singleSlotLayout(),
+ Map.of(0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0)), Map.of("sub", 4),
+ baseConfig().maxSegments(3).build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Rebucket(List.of(0L, 1L, 2L), 2,
+ "at-max-segments"));
+ }
+
+ @Test
+ public void testBucketCapDoesNotSuppressTrafficSplit() {
+ AutoScaleDecision decision = decide(initialLayout(1),
+ Map.of(0L, sample(0, 0, 0, SPLIT_BYTES_OUT + 1, 0)), Map.of("sub", 5),
+ baseConfig().maxEntryBucketsPerSegment(4).build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Split(0, "byteRateOut"));
+ }
+
+ @Test
+ public void testBlockedCheckpointSplitDoesNotSuppressStreamRebucket() {
+ AutoScaleDecision decision = decide(initialLayout(1),
+ Map.of(0L, sample(2_000, 0, 0, 0, 0)), Map.of("group", 10, "sub", 5), Set.of("sub"),
+ baseConfig().maxSegments(1).build());
+ assertThat(decision).isEqualTo(new AutoScaleDecision.Rebucket(List.of(0L), 8, "at-max-segments"));
+ }
+
+ @Test
+ public void testBucketCapDoesNotBlockImmatureMerge() {
+ AutoScaleConfig config = baseConfig().maxEntryBucketsPerSegment(2).build();
+ AutoScaleDecision decision = decide(initialLayout(2), Map.of(0L, cold(0), 1L, cold(0)),
+ Map.of("sub", 4), config);
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testSingleSlotSplitCooldownIsNotBlocked() {
+ AutoScaleDecision decision = AutoScalePolicyEvaluator.decide(singleSlotLayout(),
+ Map.of(0L, sample(SPLIT_MSG_IN * 2, 0, 0, 0, 0)), Map.of("group", 4), Set.of(),
+ baseConfig().build(), NOW, NOW - 1, NO_PRIOR, NO_PRIOR);
+ assertThat(decision).isEqualTo(AutoScaleDecision.NONE);
+ }
+
+ @Test
+ public void testDepthBlockedPairDoesNotSuppressOtherMerge() {
+ SegmentLayout layout = initialLayout(3).mergeSegments(0, 1, NOW).splitSegment(3, NOW);
+ // Segments 4 and 5 inherited merge depth 1; split segment 2 gives an eligible
+ // depth-zero pair, even though the other adjacent pairs are at the cap.
+ layout = layout.splitSegment(2, NOW);
+ AutoScaleDecision decision = decide(layout, Map.of(
+ 4L, cold(old()), 5L, cold(old()), 6L, cold(old()), 7L, cold(old())),
+ Map.of(), baseConfig().maxDagDepth(1).build());
+ assertThat(decision).isInstanceOf(AutoScaleDecision.Merge.class);
+ AutoScaleDecision.Merge merge = (AutoScaleDecision.Merge) decision;
+ assertThat(Set.of(merge.segmentId1(), merge.segmentId2())).isEqualTo(Set.of(6L, 7L));
+ }
+
@Test
public void testMergedPairIsAlwaysAdjacent() {
SegmentLayout layout = initialLayout(4);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java
index c04f6a156e2e..46d7bef9d372 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java
@@ -19,8 +19,11 @@
package org.apache.pulsar.broker.service.scalable;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyDouble;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertThrows;
@@ -82,6 +85,7 @@ public class ScalableTopicControllerAutoScaleTest {
private Policies namespacePolicies;
private ScalableTopicController controller;
private TopicName topicName;
+ private ScalableTopicControllerMetrics metrics;
@BeforeMethod
public void setUp() throws Exception {
@@ -93,6 +97,7 @@ public class ScalableTopicControllerAutoScaleTest {
// Auto-scale tuned for deterministic single-shot evaluation: no cooldowns/windows so a
// single evaluateAutoScaleForTest() call acts immediately, low-ish thresholds.
+ metrics = mock(ScalableTopicControllerMetrics.class);
config = new ServiceConfiguration();
config.setScalableTopicAutoScaleEnabled(true);
config.setScalableTopicMaxSegments(64);
@@ -157,6 +162,7 @@ public class ScalableTopicControllerAutoScaleTest {
ScalableTopicController.createInitialMetadata(initialSegments, 4, Map.of())).get();
controller = new ScalableTopicController(topicName, resources, brokerService,
coordinationService);
+ controller.setMetrics(metrics);
controller.initialize().get();
}
@@ -164,6 +170,17 @@ public class ScalableTopicControllerAutoScaleTest {
return controller.getLayout().get().getActiveSegments().size();
}
+ @Test
+ public void testBlockedAutoScaleMetric() throws Exception {
+ config.setScalableTopicMaxSegments(2);
+ startController(2);
+ resources.reportSegmentLoadAsync(topicName, 0,
+ new SegmentLoadStats(20_000, 0, 0, 0)).get();
+ controller.evaluateAutoScaleForTest().get();
+ verify(metrics).recordBlocked("max_segments");
+ assertEquals(activeSegmentCount(), 2);
+ }
+
@Test
public void testLoadDrivenSplit() throws Exception {
startController(2);
@@ -175,6 +192,7 @@ public class ScalableTopicControllerAutoScaleTest {
controller.evaluateAutoScaleForTest().get();
assertEquals(activeSegmentCount(), 3, "hot segment should have been split");
+ verify(metrics).recordOperation(eq("split"), eq("auto"), eq("success"), anyDouble());
}
@Test
@@ -330,6 +348,7 @@ public class ScalableTopicControllerAutoScaleTest {
controller.evaluateAutoScaleForTest().get();
assertEquals(activeSegmentCount(), 2,
"invalid override combination must disable auto split/merge, not split");
+ verify(metrics).recordBlocked("invalid_policy");
}
/**
@@ -489,6 +508,7 @@ public class ScalableTopicControllerAutoScaleTest {
controller.close().join();
controller = new ScalableTopicController(topicName, resources, brokerService,
coordinationService);
+ controller.setMetrics(metrics);
controller.initialize().get();
// …so another consumer-driven rollover is blocked by the seeded rebucket cooldown…
@@ -557,6 +577,7 @@ public class ScalableTopicControllerAutoScaleTest {
}
controller = new ScalableTopicController(topicName, resources, brokerService,
coordinationService);
+ controller.setMetrics(metrics);
controller.initialize().get();
controller.evaluateAutoScaleForTest().get();
@@ -585,6 +606,7 @@ public class ScalableTopicControllerAutoScaleTest {
controller.close().join();
controller = new ScalableTopicController(topicName, resources, brokerService,
coordinationService);
+ controller.setMetrics(metrics);
controller.initialize().get();
resources.reportSegmentLoadAsync(topicName, 1,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerFailoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerFailoverTest.java
index 9dc64a37133a..52dc5987bc12 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerFailoverTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerFailoverTest.java
@@ -25,11 +25,19 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import io.prometheus.client.CollectorRegistry;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -40,7 +48,11 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.resources.ScalableTopicMetadata;
import org.apache.pulsar.broker.resources.ScalableTopicResources;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.TransportCnx;
@@ -53,6 +65,7 @@ import org.apache.pulsar.metadata.TestZKServer;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.NotificationType;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
+import org.apache.pulsar.metadata.api.coordination.LeaderElectionState;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.api.extended.SessionEvent;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
@@ -66,6 +79,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -175,6 +189,211 @@ public class ScalableTopicControllerFailoverTest {
.hasMessageContaining("not the leader");
}
+ @DataProvider
+ public Object[][] takeoverInterruptions() {
+ return new Object[][]{
+ {"close", false}, {"close", true},
+ {"leadership_loss", false}, {"leadership_loss", true},
+ {"superseded", false}, {"superseded", true}
+ };
+ }
+
+ @Test(dataProvider = "takeoverInterruptions", timeOut = 30_000)
+ public void testInterruptedTakeoverIsNotFailure(String interruption, boolean restoreFails) throws Exception {
+ ScalableTopicResources heldResources = spy(broker1.resources);
+ CompletableFuture<List<String>> restore = new CompletableFuture<>();
+ CompletableFuture<Void> entered = new CompletableFuture<>();
+ AtomicInteger restores = new AtomicInteger();
+ doAnswer(invocation -> {
+ if (restores.getAndIncrement() == 0) {
+ entered.complete(null);
+ return restore;
+ }
+ return CompletableFuture.completedFuture(List.of());
+ }).when(heldResources).listSubscriptionsAsync(topic);
+ ScalableTopicController first = new ScalableTopicController(topic, heldResources,
+ broker1.brokerService, broker1.coordinationService);
+ broker1.controllers.add(first);
+ CollectorRegistry registry = new CollectorRegistry();
+ var reader = InMemoryMetricReader.create();
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build();
+ var metrics = new ScalableTopicControllerMetrics(sdk.get("interrupted-takeover"), "test-cluster",
+ first::metricCounts, registry)) {
+ first.setMetrics(metrics);
+ CompletableFuture<Void> initialization = first.initialize();
+ entered.get(5, TimeUnit.SECONDS);
+ assertThat(first.isLeader()).as("the old attempt has not started serving").isFalse();
+ if ("close".equals(interruption)) {
+ first.close().get(5, TimeUnit.SECONDS);
+ } else {
+ ScalableTopicController second = broker2.newController();
+ second.initialize().get(5, TimeUnit.SECONDS);
+ broker2.takeOverControllerLock();
+ Awaitility.await().atMost(Duration.ofSeconds(5)).until(() -> second.isLeader()
+ && first.getLeaderState() == LeaderElectionState.Following);
+ if ("superseded".equals(interruption)) {
+ broker1.takeOverControllerLock();
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .until(() -> first.isLeader() && !second.isLeader());
+ }
+ }
+ if (restoreFails) {
+ restore.completeExceptionally(new IllegalStateException("obsolete restoration failed"));
+ } else {
+ restore.complete(List.of());
+ }
+ // An obsolete initialization may finish normally or exceptionally; neither is a takeover failure.
+ initialization.handle((__, ex) -> null).get(5, TimeUnit.SECONDS);
+ assertThat(registry.getSampleValue(
+ "pulsar_scalable_topic_controller_takeover_duration_seconds_count",
+ new String[]{"cluster", "operation_status"}, new String[]{"test-cluster", "failure"})).isNull();
+ Collection<MetricData> collected = reader.collectAllMetrics();
+ if ("superseded".equals(interruption)) {
+ assertThat(scalableMetric(collected, "controller.takeover.duration").getHistogramData().getPoints())
+ .singleElement().satisfies(point -> {
+ assertThat(point.getCount()).isEqualTo(1);
+ assertThat(point.getAttributes()).isEqualTo(Attributes.builder()
+ .put("pulsar.scalable_topic.operation.status", "success").build());
+ });
+ assertThat(registry.getSampleValue(
+ "pulsar_scalable_topic_controller_takeover_duration_seconds_count",
+ new String[]{"cluster", "operation_status"}, new String[]{"test-cluster", "success"}))
+ .isEqualTo(1);
+ } else {
+ assertThat(collected).noneSatisfy(metric -> assertThat(metric.getName())
+ .isEqualTo("pulsar.broker.scalable_topic.controller.takeover.duration"));
+ assertThat(registry.getSampleValue(
+ "pulsar_scalable_topic_controller_takeover_duration_seconds_count",
+ new String[]{"cluster", "operation_status"}, new String[]{"test-cluster", "success"}))
+ .isNull();
+ }
+ } finally {
+ restore.completeExceptionally(new IllegalStateException("release held restoration during cleanup"));
+ }
+ }
+
+ @DataProvider
+ public Object[][] layoutOperationOutcomes() {
+ return new Object[][]{
+ {"split", false}, {"split", true},
+ {"merge", false}, {"merge", true},
+ {"rebucket", false}, {"rebucket", true}
+ };
+ }
+
+ @Test(dataProvider = "layoutOperationOutcomes", timeOut = 30_000)
+ public void testLayoutOutcomeRetainedAfterLeadershipLoss(String operation, boolean succeeds) throws Exception {
+ ScalableTopicResources heldResources = spy(broker1.resources);
+ CompletableFuture<Void> updateRequested = new CompletableFuture<>();
+ CompletableFuture<Void> resumeCommit = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ Function<ScalableTopicMetadata, ScalableTopicMetadata> update = invocation.getArgument(1);
+ // The operation reaches the metadata update only after all source seals completed.
+ updateRequested.complete(null);
+ return resumeCommit.thenCompose(__ -> broker1.resources.updateScalableTopicAsync(topic, update));
+ }).when(heldResources).updateScalableTopicAsync(eq(topic), any());
+
+ ScalableTopicController first = new ScalableTopicController(topic, heldResources,
+ broker1.brokerService, broker1.coordinationService);
+ broker1.controllers.add(first);
+ ScalableTopicController second = broker2.newController();
+ CollectorRegistry registry = new CollectorRegistry();
+ var reader = InMemoryMetricReader.create();
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build();
+ var metrics = new ScalableTopicControllerMetrics(sdk.get("deposed-controller"), "test-cluster",
+ first::metricCounts, registry)) {
+ first.setMetrics(metrics);
+ first.initialize().get(5, TimeUnit.SECONDS);
+ second.initialize().get(5, TimeUnit.SECONDS);
+ CompletableFuture<SegmentLayout> pending = startLayoutOperation(first, operation);
+ updateRequested.get(5, TimeUnit.SECONDS);
+ assertThat(pending).as("the layout commit is still pending after sealing").isNotDone();
+
+ broker2.takeOverControllerLock();
+ Awaitility.await().atMost(Duration.ofSeconds(5)).until(() -> second.isLeader() && !first.isLeader());
+ ScalableTopicControllerMetrics.Counts newLeaderCounts = second.metricCounts();
+ assertThat(newLeaderCounts.controllers()).isEqualTo(1);
+ assertThat(first.metricCounts()).isEqualTo(new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+
+ if (succeeds) {
+ resumeCommit.complete(null);
+ pending.get(5, TimeUnit.SECONDS);
+ } else {
+ IllegalStateException failure = new IllegalStateException("layout commit failed after leadership loss");
+ resumeCommit.completeExceptionally(failure);
+ assertThatThrownBy(() -> pending.get(5, TimeUnit.SECONDS)).hasCause(failure);
+ }
+ // Requests rejected before starting as leader must not add another outcome.
+ assertThatThrownBy(() -> startLayoutOperation(first, operation).get(5, TimeUnit.SECONDS))
+ .hasRootCauseInstanceOf(IllegalStateException.class);
+
+ String status = succeeds ? "success" : "partial";
+ assertThat(registry.getSampleValue("pulsar_scalable_topic_layout_operation_total",
+ new String[]{"cluster", "operation", "operation_trigger", "operation_status"},
+ new String[]{"test-cluster", operation, "manual", status})).isEqualTo(1);
+ Collection<MetricData> collected = reader.collectAllMetrics();
+ assertThat(scalableMetric(collected, "layout.operation.count").getLongSumData().getPoints())
+ .singleElement().satisfies(point -> {
+ assertThat(point.getValue()).isEqualTo(1);
+ assertThat(point.getAttributes()).isEqualTo(Attributes.builder()
+ .put("pulsar.scalable_topic.operation", operation)
+ .put("pulsar.scalable_topic.operation.trigger", "manual")
+ .put("pulsar.scalable_topic.operation.status", status).build());
+ });
+ assertThat(scalableMetric(collected, "layout.operation.duration").getHistogramData().getPoints())
+ .singleElement().satisfies(point -> {
+ assertThat(point.getCount()).isEqualTo(1);
+ assertThat(point.getAttributes()).isEqualTo(Attributes.builder()
+ .put("pulsar.scalable_topic.operation", operation)
+ .put("pulsar.scalable_topic.operation.status", status).build());
+ assertThat(registry.getSampleValue(
+ "pulsar_scalable_topic_layout_operation_duration_seconds_count",
+ new String[]{"cluster", "operation", "operation_status"},
+ new String[]{"test-cluster", operation, status})).isEqualTo(point.getCount());
+ });
+ if (succeeds) {
+ assertThat(scalableMetric(collected, "layout.seal.duration").getHistogramData().getPoints())
+ .singleElement().satisfies(point -> {
+ assertThat(point.getCount()).isEqualTo(1);
+ assertThat(point.getAttributes()).isEqualTo(Attributes.builder()
+ .put("pulsar.scalable_topic.operation", operation).build());
+ assertThat(registry.getSampleValue(
+ "pulsar_scalable_topic_layout_seal_duration_seconds_count",
+ new String[]{"cluster", "operation"}, new String[]{"test-cluster", operation}))
+ .isEqualTo(1);
+ });
+ } else {
+ assertThat(collected).noneSatisfy(metric -> assertThat(metric.getName())
+ .isEqualTo("pulsar.broker.scalable_topic.layout.seal.duration"));
+ assertThat(registry.getSampleValue("pulsar_scalable_topic_layout_seal_duration_seconds_count",
+ new String[]{"cluster", "operation"}, new String[]{"test-cluster", operation})).isNull();
+ }
+ assertThat(registry.getSampleValue("pulsar_scalable_topic_controller_count",
+ new String[]{"cluster"}, new String[]{"test-cluster"})).isZero();
+ assertThat(scalableMetric(collected, "controller.count").getLongSumData().getPoints())
+ .singleElement().satisfies(point -> assertThat(point.getValue()).isZero());
+ assertThat(first.metricCounts()).isEqualTo(new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+ assertThat(second.metricCounts()).isEqualTo(newLeaderCounts);
+ } finally {
+ resumeCommit.completeExceptionally(new IllegalStateException("release held update during cleanup"));
+ }
+ }
+
+ private static CompletableFuture<SegmentLayout> startLayoutOperation(ScalableTopicController controller,
+ String operation) {
+ return switch (operation) {
+ case "split" -> controller.splitSegment(0);
+ case "merge" -> controller.mergeSegments(0, 1);
+ case "rebucket" -> controller.rebucketSegment(0, 4);
+ default -> throw new IllegalArgumentException(operation);
+ };
+ }
+
+ private static MetricData scalableMetric(Collection<MetricData> metrics, String name) {
+ return metrics.stream().filter(metric -> metric.getName().equals("pulsar.broker.scalable_topic." + name))
+ .findFirst().orElseThrow();
+ }
+
/**
* A leader deposed by its election closes the election, as a follower does once elected: the
* election would otherwise take the lock back by itself once the new leader releases it, and for
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsIsolationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsIsolationTest.java
new file mode 100644
index 000000000000..2a38b07617ce
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsIsolationTest.java
@@ -0,0 +1,435 @@
+/*
+ * 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.pulsar.broker.service.scalable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import io.prometheus.client.Collector;
+import io.prometheus.client.Collector.MetricFamilySamples;
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.Counter;
+import io.prometheus.client.Gauge;
+import io.prometheus.client.Histogram;
+import io.prometheus.client.Summary;
+import java.io.ByteArrayOutputStream;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicMetrics.RequestType;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicMetrics.WatchType;
+import org.apache.pulsar.broker.stats.BrokerOperabilityMetrics;
+import org.apache.pulsar.broker.stats.OpenTelemetryTopicStats;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGeneratorUtils;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-api")
+public class ScalableTopicControllerMetricsIsolationTest {
+ private static final String PREFIX = "pulsar_metrics_isolation_";
+ private static final String OTEL_PREFIX = "pulsar.broker.scalable_topic.";
+ private static final Set<String> CONTROLLER_METRICS = Set.of(
+ OTEL_PREFIX + "controller.count", OTEL_PREFIX + "controller.takeover.duration",
+ OTEL_PREFIX + "segment.count", OTEL_PREFIX + "layout.operation.count",
+ OTEL_PREFIX + "layout.operation.duration", OTEL_PREFIX + "layout.seal.duration",
+ OTEL_PREFIX + "auto_scale.blocked.count");
+ private static final String REQUEST_DURATION = "pulsar_scalable_topic_request_duration_seconds";
+ private static final String WATCH_SESSIONS = "pulsar_scalable_topic_watch_session_count";
+ private static final String WATCH_UPDATES = "pulsar_scalable_topic_watch_update";
+ private static final Set<String> SESSION_METRICS = Set.of(REQUEST_DURATION, WATCH_SESSIONS, WATCH_UPDATES);
+ private static final String CLUSTER = "metrics-isolation";
+ private static final ScalableTopicControllerMetrics.Counts COUNTS =
+ new ScalableTopicControllerMetrics.Counts(1, 2, 3);
+
+ @DataProvider
+ public Object[][] registries() {
+ return new Object[][]{{new CollectorRegistry()}, {CollectorRegistry.defaultRegistry}};
+ }
+
+ @Test(dataProvider = "registries")
+ public void testExistingPrometheusCollectorsRemainUnchanged(CollectorRegistry registry) {
+ Counter counter = Counter.build(PREFIX + "counter_total", "Existing counter")
+ .labelNames("cluster", "topic").register(registry);
+ Gauge gauge = Gauge.build(PREFIX + "gauge", "Existing gauge")
+ .labelNames("cluster", "topic").register(registry);
+ Histogram histogram = Histogram.build(PREFIX + "histogram_seconds", "Existing histogram")
+ .labelNames("cluster", "topic").buckets(0.01, 0.1, 1).register(registry);
+ Summary summary = Summary.build(PREFIX + "summary_seconds", "Existing summary")
+ .labelNames("cluster", "topic").quantile(0.5, 0.01).register(registry);
+ List<Collector> existing = List.of(counter, gauge, histogram, summary);
+ try (var sdk = SdkMeterProvider.builder().build()) {
+ Meter meter = sdk.get("shared-broker-meter");
+ counter.labels(CLUSTER, "persistent://tenant/ns/topic").inc(7);
+ gauge.labels(CLUSTER, "persistent://tenant/ns/topic").set(3);
+ histogram.labels(CLUSTER, "persistent://tenant/ns/topic").observe(0.05);
+ summary.labels(CLUSTER, "persistent://tenant/ns/topic").observe(0.02);
+ List<MetricFamilySamples> before = existingPrometheus(registry);
+ assertThat(before).hasSize(4);
+ try (var added = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> COUNTS, registry)) {
+ recordAllMetrics(added);
+ assertThat(existingPrometheus(registry)).containsExactlyInAnyOrderElementsOf(before);
+ assertThat(existingPrometheus(registry)).containsExactlyInAnyOrderElementsOf(before);
+ counter.labels(CLUSTER, "persistent://tenant/ns/topic").inc(2);
+ gauge.labels(CLUSTER, "persistent://tenant/ns/topic").dec();
+ histogram.labels(CLUSTER, "persistent://tenant/ns/topic").observe(0.5);
+ summary.labels(CLUSTER, "persistent://tenant/ns/topic").observe(0.2);
+ }
+ List<MetricFamilySamples> after = existingPrometheus(registry);
+ assertThat(after).isNotEqualTo(before);
+ assertThat(counter.labels(CLUSTER, "persistent://tenant/ns/topic").get()).isEqualTo(9);
+ assertThat(gauge.labels(CLUSTER, "persistent://tenant/ns/topic").get()).isEqualTo(2);
+ try (var replacement = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> COUNTS, registry)) {
+ recordAllMetrics(replacement);
+ assertThat(existingPrometheus(registry)).containsExactlyInAnyOrderElementsOf(after);
+ }
+ assertThat(existingPrometheus(registry)).containsExactlyInAnyOrderElementsOf(after);
+ } finally {
+ existing.forEach(registry::unregister);
+ }
+ }
+
+ @Test
+ public void testExistingOpenTelemetryInstrumentsRemainUnchanged() {
+ var reader = InMemoryMetricReader.create();
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build()) {
+ Meter meter = sdk.get("shared-broker-meter");
+ Attributes attributes = Attributes.builder().put("pulsar.topic", "persistent://tenant/ns/topic").build();
+ AtomicLong connections = new AtomicLong(7);
+ AtomicLong producers = new AtomicLong(3);
+ // Use the existing names and kinds on the very same SDK Meter as the added instruments.
+ try (var counter = meter.counterBuilder(BrokerOperabilityMetrics.CONNECTION_COUNTER_METRIC_NAME)
+ .setDescription("Existing connection counter").setUnit("{connection}")
+ .buildWithCallback(m -> m.record(connections.get(), attributes));
+ var upDownCounter = meter.upDownCounterBuilder(OpenTelemetryTopicStats.PRODUCER_COUNTER)
+ .setDescription("Existing producer count").setUnit("{producer}")
+ .buildWithCallback(m -> m.record(producers.get(), attributes))) {
+ var histogram = meter.histogramBuilder(BrokerOperabilityMetrics.TOPIC_PUBLISH_LATENCY_METRIC_NAME)
+ .setDescription("Existing publish latency").setUnit("s")
+ .setExplicitBucketBoundariesAdvice(List.of(0.01, 0.1, 1.0)).build();
+ histogram.record(0.05, attributes);
+ Map<String, Object> before = existingOpenTelemetry(reader.collectAllMetrics());
+ assertThat(before).hasSize(3);
+ Map<String, Object> updated;
+ try (var added = new ScalableTopicControllerMetrics(
+ meter, CLUSTER, () -> COUNTS, new CollectorRegistry())) {
+ recordAllMetrics(added);
+ assertThat(existingOpenTelemetry(reader.collectAllMetrics())).isEqualTo(before);
+ assertThat(existingOpenTelemetry(reader.collectAllMetrics())).isEqualTo(before);
+ connections.addAndGet(2);
+ producers.decrementAndGet();
+ histogram.record(0.5, attributes);
+ updated = existingOpenTelemetry(reader.collectAllMetrics());
+ assertThat(updated).isNotEqualTo(before);
+ }
+ assertThat(existingOpenTelemetry(reader.collectAllMetrics())).isEqualTo(updated);
+ try (var replacement = new ScalableTopicControllerMetrics(
+ meter, CLUSTER, () -> COUNTS, new CollectorRegistry())) {
+ recordAllMetrics(replacement);
+ assertThat(existingOpenTelemetry(reader.collectAllMetrics())).isEqualTo(updated);
+ }
+ assertThat(existingOpenTelemetry(reader.collectAllMetrics())).isEqualTo(updated);
+ }
+ }
+ }
+
+ @DataProvider
+ public Object[][] sessionMetricsModes() {
+ return new Object[][]{
+ {true, new CollectorRegistry()}, {true, CollectorRegistry.defaultRegistry},
+ {false, new CollectorRegistry()}, {false, CollectorRegistry.defaultRegistry}
+ };
+ }
+
+ @DataProvider
+ public Object[][] residentSessionMetricsModes() {
+ return new Object[][]{{true}, {false}};
+ }
+
+ @Test(dataProvider = "residentSessionMetricsModes")
+ public void testCoexistenceWithResidentSessionProvider(boolean telemetryEnabled) {
+ CollectorRegistry registry = CollectorRegistry.defaultRegistry;
+ String residentCluster = "resident-session-" + UUID.randomUUID();
+ var reader = InMemoryMetricReader.create();
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build()) {
+ Meter meter = telemetryEnabled ? sdk.get("resident-broker") : MeterProvider.noop().get("resident-broker");
+ try (var resident = new ScalableTopicMetrics(meter, residentCluster, registry)) {
+ resident.sessionOpened(WatchType.TOPIC);
+ resident.recordWatchUpdate(true);
+ AtomicLong successfulUpdates = new AtomicLong(1);
+ Runnable residentActivity = () -> {
+ resident.startRequest(RequestType.LOOKUP).success();
+ resident.recordWatchUpdate(true);
+ successfulUpdates.incrementAndGet();
+ };
+ for (int attempt = 0; attempt < 2; attempt++) {
+ assertSessionMetricsSurviveControllerLifecycle(telemetryEnabled, registry, residentActivity);
+ assertThat(registry.getSampleValue(WATCH_SESSIONS, new String[]{"cluster", "watch_type"},
+ new String[]{residentCluster, "topic"})).isEqualTo(1);
+ assertThat(registry.getSampleValue(WATCH_UPDATES + "_total",
+ new String[]{"cluster", "operation_status"}, new String[]{residentCluster, "success"}))
+ .isEqualTo((double) successfulUpdates.get());
+ assertThat(Collections.list(registry.metricFamilySamples())).extracting(family -> family.name)
+ .containsAll(SESSION_METRICS);
+ if (telemetryEnabled) {
+ assertThat(reader.collectAllMetrics()).anySatisfy(metric -> {
+ assertThat(metric.getName()).isEqualTo(ScalableTopicMetrics.WATCH_UPDATE_COUNT_METRIC_NAME);
+ assertThat(metric.getLongSumData().getPoints()).filteredOn(point -> point.getAttributes()
+ .equals(Attributes.builder()
+ .put("pulsar.scalable_topic.operation.status", "success").build()))
+ .singleElement().satisfies(point ->
+ assertThat(point.getValue()).isEqualTo(successfulUpdates.get()));
+ });
+ assertThat(reader.collectAllMetrics()).anySatisfy(metric -> {
+ assertThat(metric.getName()).isEqualTo(ScalableTopicMetrics.REQUEST_DURATION_METRIC_NAME);
+ assertThat(metric.getHistogramData().getPoints()).singleElement().satisfies(point ->
+ assertThat(point.getCount()).isEqualTo(successfulUpdates.get() - 1));
+ });
+ } else {
+ assertThat(reader.collectAllMetrics()).isEmpty();
+ }
+ }
+ }
+ }
+ }
+
+ @Test(dataProvider = "sessionMetricsModes")
+ public void testUpstreamSessionMetricsSurviveControllerLifecycle(boolean telemetryEnabled,
+ CollectorRegistry registry) {
+ assertSessionMetricsSurviveControllerLifecycle(telemetryEnabled, registry, () -> { });
+ }
+
+ private void assertSessionMetricsSurviveControllerLifecycle(boolean telemetryEnabled, CollectorRegistry registry,
+ Runnable residentActivity) {
+ String cluster = "session-isolation-" + UUID.randomUUID();
+ var reader = InMemoryMetricReader.create();
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build()) {
+ Meter meter = telemetryEnabled ? sdk.get("shared-broker-meter")
+ : MeterProvider.noop().get("shared-broker-meter");
+ try (var sessions = new ScalableTopicMetrics(meter, cluster, registry)) {
+ for (RequestType type : RequestType.values()) {
+ sessions.startRequest(type).success();
+ sessions.startRequest(type).failure();
+ }
+ sessions.sessionOpened(WatchType.TOPIC);
+ sessions.sessionOpened(WatchType.TOPIC);
+ sessions.sessionClosed(WatchType.TOPIC);
+ sessions.sessionOpened(WatchType.NAMESPACE);
+ sessions.recordWatchUpdate(true);
+ sessions.recordWatchUpdate(true);
+ sessions.recordWatchUpdate(false);
+ SessionMetricsSnapshot before = sessionMetricsSnapshot(registry, reader, cluster);
+ assertThat(before.prometheus()).extracting(family -> family.name)
+ .containsExactlyInAnyOrderElementsOf(SESSION_METRICS);
+ assertThat(before.prometheus()).filteredOn(family -> family.name.equals(REQUEST_DURATION))
+ .singleElement().satisfies(family -> assertThat(family.samples)
+ .filteredOn(sample -> sample.name.equals(REQUEST_DURATION + "_count"))
+ .hasSize(8).allSatisfy(sample -> {
+ assertThat(sample.value).isEqualTo(1);
+ assertThat(sample.labelNames)
+ .containsExactly("cluster", "request_type", "operation_status");
+ }));
+ assertThat(registry.getSampleValue(WATCH_SESSIONS,
+ new String[]{"cluster", "watch_type"}, new String[]{cluster, "topic"})).isEqualTo(1);
+ assertThat(registry.getSampleValue(WATCH_SESSIONS,
+ new String[]{"cluster", "watch_type"}, new String[]{cluster, "namespace"})).isEqualTo(1);
+ assertThat(registry.getSampleValue(WATCH_UPDATES + "_total",
+ new String[]{"cluster", "operation_status"}, new String[]{cluster, "success"})).isEqualTo(2);
+ assertThat(registry.getSampleValue(WATCH_UPDATES + "_total",
+ new String[]{"cluster", "operation_status"}, new String[]{cluster, "failure"})).isEqualTo(1);
+ if (telemetryEnabled) {
+ assertThat(before.openTelemetry()).containsOnlyKeys(
+ ScalableTopicMetrics.REQUEST_DURATION_METRIC_NAME,
+ ScalableTopicMetrics.WATCH_SESSION_COUNT_METRIC_NAME,
+ ScalableTopicMetrics.WATCH_UPDATE_COUNT_METRIC_NAME);
+ assertThat(reader.collectAllMetrics()).anySatisfy(metric -> {
+ assertThat(metric.getName()).isEqualTo(ScalableTopicMetrics.REQUEST_DURATION_METRIC_NAME);
+ assertThat(metric.getHistogramData().getPoints()).hasSize(8).allSatisfy(point -> {
+ assertThat(point.getCount()).isEqualTo(1);
+ assertThat(point.getBoundaries()).containsExactly(
+ 0.001, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 5.0, 30.0);
+ });
+ });
+ } else {
+ assertThat(before.openTelemetry()).isEmpty();
+ }
+ SessionMetricsSnapshot updated;
+ residentActivity.run();
+ try (var added = new ScalableTopicControllerMetrics(meter, cluster, () -> COUNTS, registry)) {
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(before);
+ recordAllMetrics(added);
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(before);
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(before);
+ // Actual upstream instruments must still record while controller instruments are present.
+ sessions.startRequest(RequestType.LOOKUP).success();
+ sessions.startRequest(RequestType.WATCH_TOPICS).failure();
+ sessions.sessionOpened(WatchType.TOPIC);
+ sessions.sessionClosed(WatchType.NAMESPACE);
+ sessions.recordWatchUpdate(true);
+ sessions.recordWatchUpdate(false);
+ updated = sessionMetricsSnapshot(registry, reader, cluster);
+ assertThat(updated.prometheus()).isNotEqualTo(before.prometheus());
+ if (telemetryEnabled) {
+ assertThat(updated.openTelemetry()).isNotEqualTo(before.openTelemetry());
+ }
+ residentActivity.run();
+ recordAllMetrics(added);
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(updated);
+ }
+ residentActivity.run();
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(updated);
+ try (var replacement = new ScalableTopicControllerMetrics(meter, cluster, () -> COUNTS, registry)) {
+ residentActivity.run();
+ recordAllMetrics(replacement);
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(updated);
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(updated);
+ }
+ assertThat(sessionMetricsSnapshot(registry, reader, cluster)).isEqualTo(updated);
+ sessions.sessionClosed(WatchType.TOPIC);
+ sessions.startRequest(RequestType.UNSUBSCRIBE).success();
+ sessions.recordWatchUpdate(false);
+ SessionMetricsSnapshot after = sessionMetricsSnapshot(registry, reader, cluster);
+ assertThat(after.prometheus()).isNotEqualTo(updated.prometheus());
+ if (telemetryEnabled) {
+ assertThat(after.openTelemetry()).isNotEqualTo(updated.openTelemetry());
+ }
+ }
+ // Other brokers may keep the shared collector and these families registered.
+ assertThat(sessionPrometheusSnapshot(registry, cluster)).isEmpty();
+ }
+ }
+
+ private record SessionMetricsSnapshot(List<MetricFamilySamples> prometheus, Map<String, Object> openTelemetry) { }
+
+ private static SessionMetricsSnapshot sessionMetricsSnapshot(CollectorRegistry registry,
+ InMemoryMetricReader reader, String cluster) {
+ // The SDK reader belongs to this fixture; only the Prometheus registry is shared with other brokers.
+ return new SessionMetricsSnapshot(sessionPrometheusSnapshot(registry, cluster),
+ existingOpenTelemetry(reader.collectAllMetrics()));
+ }
+
+ private static List<MetricFamilySamples> sessionPrometheusSnapshot(CollectorRegistry registry, String cluster) {
+ return Collections.list(registry.metricFamilySamples()).stream()
+ .filter(family -> SESSION_METRICS.contains(family.name))
+ .map(family -> new MetricFamilySamples(family.name, family.unit, family.type, family.help,
+ family.samples.stream().filter(sample -> {
+ int index = sample.labelNames.indexOf("cluster");
+ return index >= 0 && cluster.equals(sample.labelValues.get(index));
+ }).toList()))
+ .filter(family -> !family.samples.isEmpty())
+ .toList();
+ }
+
+ @DataProvider
+ public Object[][] brokenStates() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "brokenStates")
+ public void testBrokenNewStateDoesNotBreakExistingMetrics(boolean nullState) throws Exception {
+ var reader = InMemoryMetricReader.create();
+ CollectorRegistry registry = CollectorRegistry.defaultRegistry;
+ Counter existing = Counter.build(PREFIX + "scrape_total", "Existing scrape sentinel").register(registry);
+ try (var sdk = SdkMeterProvider.builder().registerMetricReader(reader).build()) {
+ Meter meter = sdk.get("shared-broker-meter");
+ var counter = meter.counterBuilder(BrokerOperabilityMetrics.CONNECTION_COUNTER_METRIC_NAME).build();
+ counter.add(11);
+ existing.inc(7);
+ Map<String, Object> before = existingOpenTelemetry(reader.collectAllMetrics());
+ try (var broken = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> {
+ if (nullState) {
+ return null;
+ }
+ throw new IllegalStateException("Cannot read new controller state");
+ });
+ var healthy = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> COUNTS)) {
+ broken.recordBlocked("invalid_policy");
+ recordAllMetrics(healthy);
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ PrometheusMetricsGeneratorUtils.generate(CLUSTER, output, List.of());
+ assertThat(output.toString()).contains(PREFIX + "scrape_total{cluster=\"" + CLUSTER + "\"} 7.0");
+ assertThat(registry.getSampleValue("pulsar_scalable_topic_controller_count",
+ new String[]{"cluster"}, new String[]{CLUSTER})).isEqualTo(1);
+ Collection<MetricData> collected = reader.collectAllMetrics();
+ assertThat(existingOpenTelemetry(collected)).isEqualTo(before);
+ assertThat(collected).anySatisfy(metric -> {
+ assertThat(metric.getName()).isEqualTo(OTEL_PREFIX + "controller.count");
+ assertThat(metric.getLongSumData().getPoints()).singleElement()
+ .satisfies(point -> assertThat(point.getValue()).isEqualTo(1));
+ });
+ }
+ assertThat(existing.get()).isEqualTo(7);
+ counter.add(2);
+ assertThat(reader.collectAllMetrics().stream()
+ .filter(metric -> !metric.getName().startsWith(OTEL_PREFIX)).toList())
+ .singleElement().satisfies(metric ->
+ assertThat(metric.getLongSumData().getPoints()).singleElement()
+ .satisfies(point -> assertThat(point.getValue()).isEqualTo(13)));
+ } finally {
+ registry.unregister(existing);
+ }
+ }
+
+ private static void recordAllMetrics(ScalableTopicControllerMetrics metrics) {
+ metrics.recordTakeover(0.005, true);
+ metrics.recordOperation("split", "auto", "success", 0.1);
+ metrics.recordSeal("split", 0.01);
+ metrics.recordBlocked("max_segments");
+ }
+
+ private static List<MetricFamilySamples> existingPrometheus(CollectorRegistry registry) {
+ return Collections.list(registry.metricFamilySamples()).stream()
+ .filter(metric -> metric.name.startsWith(PREFIX)).toList();
+ }
+
+ /**
+ * Exclude only controller/layout instruments, not upstream request/watch instruments sharing their prefix.
+ * Ignore collection timestamps, but preserve descriptors, start times, labels, values and histogram buckets.
+ */
+ private static Map<String, Object> existingOpenTelemetry(Collection<MetricData> metrics) {
+ Map<String, Object> result = new HashMap<>();
+ for (MetricData metric : metrics) {
+ if (CONTROLLER_METRICS.contains(metric.getName())) {
+ continue;
+ }
+ Object data = switch (metric.getType()) {
+ case LONG_SUM -> List.of(metric.getLongSumData().isMonotonic(),
+ metric.getLongSumData().getAggregationTemporality(),
+ metric.getLongSumData().getPoints().stream().map(point ->
+ List.of(point.getAttributes(), point.getValue(), point.getStartEpochNanos())).toList());
+ case HISTOGRAM -> List.of(metric.getHistogramData().getAggregationTemporality(),
+ metric.getHistogramData().getPoints().stream().map(point -> List.of(point.getAttributes(),
+ point.getStartEpochNanos(), point.getCount(), point.getSum(), point.getMin(),
+ point.getMax(), point.getBoundaries(), point.getCounts())).toList());
+ default -> throw new AssertionError("Unexpected existing metric type: " + metric.getType());
+ };
+ result.put(metric.getName(), List.of(metric.getDescription(), metric.getUnit(), metric.getType(),
+ metric.getResource(), metric.getInstrumentationScopeInfo(), data));
+ }
+ return result;
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsTest.java
new file mode 100644
index 000000000000..5359a4ef7162
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerMetricsTest.java
@@ -0,0 +1,618 @@
+/*
+ * 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.pulsar.broker.service.scalable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.data.Offset.offset;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.ObservableMeasurement;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import io.prometheus.client.Collector;
+import io.prometheus.client.Collector.MetricFamilySamples;
+import io.prometheus.client.Collector.MetricFamilySamples.Sample;
+import io.prometheus.client.CollectorRegistry;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-api")
+public class ScalableTopicControllerMetricsTest {
+ private static final String OTEL_PREFIX = "pulsar.broker.scalable_topic.";
+ private static final String PROM_PREFIX = "pulsar_scalable_topic_";
+ private static final String ATTRIBUTE_PREFIX = "pulsar.scalable_topic.";
+ private static final String CLUSTER = "test-cluster";
+ private static final ScalableTopicControllerMetrics.Counts EMPTY =
+ new ScalableTopicControllerMetrics.Counts(0, 0, 0);
+
+ private InMemoryMetricReader reader;
+ private SdkMeterProvider sdk;
+ private Meter meter;
+ private CollectorRegistry registry;
+
+ @BeforeMethod(alwaysRun = true)
+ public void setup() {
+ reader = InMemoryMetricReader.create();
+ sdk = SdkMeterProvider.builder().registerMetricReader(reader).build();
+ meter = sdk.get("scalable-topic-metrics-test");
+ registry = new CollectorRegistry();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ sdk.close();
+ }
+
+ @Test
+ public void testCounterAndHistogramParity() {
+ try (var metrics = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry)) {
+ metrics.recordTakeover(0.005, true);
+ metrics.recordTakeover(0.2, false);
+ metrics.recordOperation("split", "auto", "success", 0.001);
+ metrics.recordOperation("split", "auto", "success", 0.05);
+ metrics.recordOperation("merge", "manual", "failure", 0.1);
+ metrics.recordOperation("rebucket", "manual", "partial", 350);
+ metrics.recordOperation("prune", "auto", "success", 0);
+ metrics.recordSeal("split", 0.002);
+ metrics.recordSeal("merge", 2);
+ metrics.recordSeal("rebucket", 31);
+ for (String reason : List.of("max_segments", "max_entry_buckets", "max_dag_depth", "unsplittable",
+ "invalid_policy")) {
+ metrics.recordBlocked(reason);
+ metrics.recordBlocked(reason);
+ }
+
+ Collection<MetricData> otel = reader.collectAllMetrics();
+ List<MetricFamilySamples> prometheus = Collections.list(registry.metricFamilySamples());
+ assertThat(otel).extracting(MetricData::getName).containsExactlyInAnyOrder(
+ OTEL_PREFIX + "controller.count", OTEL_PREFIX + "segment.count",
+ OTEL_PREFIX + "controller.takeover.duration", OTEL_PREFIX + "layout.operation.count",
+ OTEL_PREFIX + "layout.operation.duration", OTEL_PREFIX + "layout.seal.duration",
+ OTEL_PREFIX + "auto_scale.blocked.count");
+ assertThat(prometheus).hasSize(7);
+ for (MetricFamilySamples family : prometheus) {
+ assertThat(family.type).isIn(Collector.Type.COUNTER, Collector.Type.GAUGE, Collector.Type.HISTOGRAM);
+ for (Sample sample : family.samples) {
+ assertThat(labels(sample)).containsEntry("cluster", CLUSTER);
+ assertThat(sample.labelNames).doesNotContain("topic", "broker", "namespace", "segment");
+ }
+ }
+ assertCounterParity(otel, prometheus, "layout.operation.count", "layout_operation_total", 5);
+ assertCounterParity(otel, prometheus, "auto_scale.blocked.count", "auto_scale_blocked_total", 10);
+ assertHistogramParity(otel, prometheus, "controller.takeover.duration",
+ "controller_takeover_duration_seconds", 2, 0.205);
+ assertHistogramParity(otel, prometheus, "layout.operation.duration",
+ "layout_operation_duration_seconds", 5, 350.151);
+ assertHistogramParity(otel, prometheus, "layout.seal.duration", "layout_seal_duration_seconds", 3, 33.002);
+
+ assertThat(sample(prometheus, "layout_operation_total", Map.of("cluster", CLUSTER,
+ "operation", "split", "operation_trigger", "auto", "operation_status", "success")))
+ .isEqualTo(2);
+ assertThat(prometheus.stream().flatMap(family -> family.samples.stream())
+ .filter(value -> value.name.endsWith("_bucket")))
+ .allSatisfy(value -> assertThat(value.labelNames).contains("le"));
+ }
+ }
+
+ @Test
+ public void testGaugesReadCurrentCountsAtCollection() {
+ AtomicReference<ScalableTopicControllerMetrics.Counts> counts = new AtomicReference<>(EMPTY);
+ AtomicInteger reads = new AtomicInteger();
+ try (var metrics = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> {
+ reads.incrementAndGet();
+ return counts.get();
+ }, registry)) {
+ assertThat(reads).hasValue(0);
+ counts.set(new ScalableTopicControllerMetrics.Counts(2, 5, 7));
+ assertGauges(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()), 2, 5, 7);
+ assertThat(reads).hasValue(2);
+ counts.set(new ScalableTopicControllerMetrics.Counts(1, 3, 11));
+ assertGauges(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()), 1, 3, 11);
+ assertThat(reads).hasValue(4);
+ }
+ }
+
+ @Test
+ public void testMultipleInstancesAggregateAndCleanUp() {
+ AtomicInteger firstReads = new AtomicInteger();
+ var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> {
+ firstReads.incrementAndGet();
+ return new ScalableTopicControllerMetrics.Counts(1, 2, 3);
+ }, registry);
+ var second = new ScalableTopicControllerMetrics(meter, CLUSTER,
+ () -> new ScalableTopicControllerMetrics.Counts(4, 5, 6), registry);
+ try {
+ first.recordOperation("split", "manual", "success", 0.005);
+ second.recordOperation("split", "manual", "success", 0.01);
+ first.recordBlocked("max_segments");
+ first.recordBlocked("max_segments");
+ second.recordBlocked("max_segments");
+ var otel = reader.collectAllMetrics();
+ var prometheus = Collections.list(registry.metricFamilySamples());
+ assertGauges(otel, prometheus, 5, 7, 9);
+ assertCounterParity(otel, prometheus, "layout.operation.count", "layout_operation_total", 2);
+ assertCounterParity(otel, prometheus, "auto_scale.blocked.count", "auto_scale_blocked_total", 3);
+ assertHistogramParity(otel, prometheus, "layout.operation.duration",
+ "layout_operation_duration_seconds", 2, 0.015);
+ assertThat(prometheus.stream().flatMap(family -> family.samples.stream())
+ .filter(value -> value.name.equals(PROM_PREFIX + "layout_operation_total"))).hasSize(1);
+
+ first.close();
+ first.close();
+ int readsBefore = firstReads.get();
+ var afterCloseOtel = reader.collectAllMetrics();
+ var afterClosePrometheus = Collections.list(registry.metricFamilySamples());
+ assertGauges(afterCloseOtel, afterClosePrometheus, 4, 5, 6);
+ assertThat(firstReads).hasValue(readsBefore);
+ // Embedded counters include only live providers, even for OTel's monotonic sum observers.
+ // Partial close intentionally lowers both cumulative series; histogram history remains Meter-wide.
+ assertCounterParity(afterCloseOtel, afterClosePrometheus,
+ "layout.operation.count", "layout_operation_total", 1);
+ assertCounterParity(afterCloseOtel, afterClosePrometheus,
+ "auto_scale.blocked.count", "auto_scale_blocked_total", 1);
+ assertHistogramParity(afterCloseOtel, afterClosePrometheus,
+ "layout.operation.duration", "layout_operation_duration_seconds", 2, 0.015);
+ second.close();
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ assertThat(reader.collectAllMetrics()).extracting(MetricData::getName)
+ .doesNotContain(OTEL_PREFIX + "controller.count", OTEL_PREFIX + "segment.count",
+ OTEL_PREFIX + "layout.operation.count", OTEL_PREFIX + "auto_scale.blocked.count");
+ // SDK synchronous histograms retain their observations; callbacks and Prom collectors do not.
+ try (var replacement = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry)) {
+ assertGauges(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()), 0, 0, 0);
+ }
+ } finally {
+ first.close();
+ second.close();
+ }
+ }
+
+ @DataProvider
+ public Object[][] callbackFailureRegistryLayouts() {
+ return new Object[][]{{true}, {false}};
+ }
+
+ @Test(dataProvider = "callbackFailureRegistryLayouts")
+ public void testFailedCallbackSetupRollsBackOnlyFailedProvider(boolean healthyProviderPresent) {
+ var failingReader = InMemoryMetricReader.create();
+ AtomicInteger failedReads = new AtomicInteger();
+ String failedCluster = "failed-cluster";
+ AtomicReference<ScalableTopicControllerMetrics.Counts> healthyCounts =
+ new AtomicReference<>(new ScalableTopicControllerMetrics.Counts(1, 2, 3));
+ try (var failingSdk = SdkMeterProvider.builder().registerMetricReader(failingReader).build();
+ var healthy = healthyProviderPresent
+ ? new ScalableTopicControllerMetrics(meter, CLUSTER, healthyCounts::get, registry) : null) {
+ Meter failingMeter = spy(failingSdk.get("failed-callback-setup"));
+ IllegalArgumentException failure = new IllegalArgumentException("Injected callback setup failure");
+ doThrow(failure).doCallRealMethod().when(failingMeter).batchCallback(
+ any(Runnable.class), any(ObservableMeasurement.class), any(ObservableMeasurement[].class));
+ if (healthy != null) {
+ healthy.recordBlocked("max_segments");
+ healthy.recordTakeover(0.005, true);
+ }
+ assertThatThrownBy(() -> new ScalableTopicControllerMetrics(failingMeter, failedCluster, () -> {
+ failedReads.incrementAndGet();
+ return new ScalableTopicControllerMetrics.Counts(9, 9, 9);
+ }, registry)).isSameAs(failure);
+ assertThat(registry.getSampleValue(PROM_PREFIX + "controller_count",
+ new String[]{"cluster"}, new String[]{failedCluster})).isNull();
+ if (healthy != null) {
+ assertGauges(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()), 1, 2, 3);
+ assertCounterParity(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "auto_scale.blocked.count", "auto_scale_blocked_total", 1);
+ assertHistogramParity(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "controller.takeover.duration", "controller_takeover_duration_seconds", 1, 0.005);
+ } else {
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ assertThat(reader.collectAllMetrics()).extracting(MetricData::getName)
+ .doesNotContain(OTEL_PREFIX + "controller.count");
+ }
+ assertThat(failingReader.collectAllMetrics()).extracting(MetricData::getName)
+ .doesNotContain(OTEL_PREFIX + "controller.count", OTEL_PREFIX + "segment.count",
+ OTEL_PREFIX + "layout.operation.count", OTEL_PREFIX + "auto_scale.blocked.count");
+ assertThat(failedReads).hasValue(0);
+
+ // Retry on the same Meter and registry must create a usable callback, without the failed owner.
+ try (var retry = new ScalableTopicControllerMetrics(failingMeter, failedCluster,
+ () -> new ScalableTopicControllerMetrics.Counts(4, 5, 6), registry)) {
+ assertThat(registry.getSampleValue(PROM_PREFIX + "controller_count",
+ new String[]{"cluster"}, new String[]{failedCluster})).isEqualTo(4.0);
+ assertThat(metric(failingReader.collectAllMetrics(), "controller.count")
+ .getLongSumData().getPoints()).singleElement()
+ .satisfies(point -> assertThat(point.getValue()).isEqualTo(4));
+ }
+ assertThat(registry.getSampleValue(PROM_PREFIX + "controller_count",
+ new String[]{"cluster"}, new String[]{failedCluster})).isNull();
+ assertThat(failingReader.collectAllMetrics()).extracting(MetricData::getName)
+ .doesNotContain(OTEL_PREFIX + "controller.count", OTEL_PREFIX + "segment.count",
+ OTEL_PREFIX + "layout.operation.count", OTEL_PREFIX + "auto_scale.blocked.count");
+ assertThat(failedReads).hasValue(0);
+ if (healthy != null) {
+ healthyCounts.set(new ScalableTopicControllerMetrics.Counts(7, 8, 9));
+ healthy.recordBlocked("max_segments");
+ assertGauges(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()), 7, 8, 9);
+ assertCounterParity(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "auto_scale.blocked.count", "auto_scale_blocked_total", 2);
+ }
+ }
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ assertThat(reader.collectAllMetrics()).extracting(MetricData::getName)
+ .doesNotContain(OTEL_PREFIX + "controller.count", OTEL_PREFIX + "segment.count",
+ OTEL_PREFIX + "layout.operation.count", OTEL_PREFIX + "auto_scale.blocked.count");
+ }
+
+ @Test
+ public void testRecreatedServiceKeepsHistogramParity() {
+ try (var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry)) {
+ first.recordTakeover(0.005, true);
+ assertHistogramParity(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "controller.takeover.duration", "controller_takeover_duration_seconds", 1, 0.005);
+ }
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ try (var replacement = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry)) {
+ replacement.recordTakeover(0.01, true);
+ assertHistogramParity(reader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "controller.takeover.duration", "controller_takeover_duration_seconds", 2, 0.015);
+ }
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ }
+
+ @Test
+ public void testDistinctMetersRemoveClosedBrokerHistograms() {
+ var otherReader = InMemoryMetricReader.create();
+ try (var otherSdk = SdkMeterProvider.builder().registerMetricReader(otherReader).build();
+ var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry);
+ var second = new ScalableTopicControllerMetrics(otherSdk.get("other-broker"), CLUSTER,
+ () -> EMPTY, registry)) {
+ first.recordTakeover(0.005, true);
+ second.recordTakeover(0.01, true);
+ var before = Collections.list(registry.metricFamilySamples());
+ assertThat(sample(before, "controller_takeover_duration_seconds_count",
+ Map.of("cluster", CLUSTER, "operation_status", "success"))).isEqualTo(2);
+ first.close();
+ assertHistogramParity(otherReader.collectAllMetrics(), Collections.list(registry.metricFamilySamples()),
+ "controller.takeover.duration", "controller_takeover_duration_seconds", 1, 0.01);
+ }
+ }
+
+ @DataProvider
+ public Object[][] sdkHistogramClusters() {
+ return new Object[][]{{CLUSTER}, {"other-cluster"}};
+ }
+
+ @Test(dataProvider = "sdkHistogramClusters")
+ public void testSharedSdkMeterHistogramHistorySpansRegistriesAndClusters(String otherCluster) {
+ var otherRegistry = new CollectorRegistry();
+ try (var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry);
+ var second = new ScalableTopicControllerMetrics(meter, otherCluster, () -> EMPTY, otherRegistry)) {
+ recordHistograms(first, 0.005);
+ // A registry with a live provider sees the Meter's history, even before that provider records anything.
+ assertHistogramState(registry, CLUSTER, 1, 0.005);
+ assertHistogramState(otherRegistry, CLUSTER, 1, 0.005);
+ assertSdkHistogramState(reader.collectAllMetrics(), 1, 0.005);
+
+ recordHistograms(second, 0.01);
+ boolean sameCluster = CLUSTER.equals(otherCluster);
+ for (var currentRegistry : List.of(registry, otherRegistry)) {
+ assertHistogramState(currentRegistry, CLUSTER, sameCluster ? 2 : 1, sameCluster ? 0.015 : 0.005);
+ assertHistogramState(currentRegistry, otherCluster, sameCluster ? 2 : 1, sameCluster ? 0.015 : 0.01);
+ }
+ // Cluster is not an instrument attribute: the SDK combines both providers' observations.
+ assertSdkHistogramState(reader.collectAllMetrics(), 2, 0.015);
+
+ first.close();
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ assertHistogramState(otherRegistry, CLUSTER, sameCluster ? 2 : 1, sameCluster ? 0.015 : 0.005);
+ assertHistogramState(otherRegistry, otherCluster, sameCluster ? 2 : 1, sameCluster ? 0.015 : 0.01);
+ assertSdkHistogramState(reader.collectAllMetrics(), 2, 0.015);
+ }
+ assertThat(Collections.list(otherRegistry.metricFamilySamples())).isEmpty();
+ }
+
+ @Test(dataProvider = "sdkHistogramClusters")
+ public void testDistinctSdkMetersIsolateHistogramScopes(String otherCluster) {
+ var otherRegistry = new CollectorRegistry();
+ var otherReader = InMemoryMetricReader.create();
+ try (var otherSdk = SdkMeterProvider.builder().registerMetricReader(otherReader).build();
+ var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry);
+ var second = new ScalableTopicControllerMetrics(otherSdk.get("other-broker"), otherCluster,
+ () -> EMPTY, otherRegistry)) {
+ recordHistograms(first, 0.005);
+ assertHistogramState(otherRegistry, CLUSTER, 0, 0);
+ recordHistograms(second, 0.01);
+ assertHistogramState(registry, CLUSTER, 1, 0.005);
+ assertHistogramState(otherRegistry, otherCluster, 1, 0.01);
+ assertSdkHistogramState(reader.collectAllMetrics(), 1, 0.005);
+ assertSdkHistogramState(otherReader.collectAllMetrics(), 1, 0.01);
+ if (!CLUSTER.equals(otherCluster)) {
+ assertHistogramState(registry, otherCluster, 0, 0);
+ assertHistogramState(otherRegistry, CLUSTER, 0, 0);
+ }
+
+ first.close();
+ assertThat(Collections.list(registry.metricFamilySamples())).isEmpty();
+ assertHistogramState(otherRegistry, otherCluster, 1, 0.01);
+ assertSdkHistogramState(otherReader.collectAllMetrics(), 1, 0.01);
+ }
+ assertThat(Collections.list(otherRegistry.metricFamilySamples())).isEmpty();
+ }
+
+ @Test
+ public void testDefaultRegistryAndClusterIsolationWithDisabledOpenTelemetry() {
+ assertDisabledDefaultRegistryCycle("first", "second");
+ }
+
+ @Test
+ public void testDisabledDefaultRegistryIsolationWithSurvivorAndRepeatedCycles() {
+ Meter disabled = OpenTelemetry.noop().getMeter("resident");
+ try (var resident = new ScalableTopicControllerMetrics(disabled, "resident-default", () -> EMPTY)) {
+ recordHistograms(resident, 0.005);
+ for (int attempt = 0; attempt < 2; attempt++) {
+ assertDisabledDefaultRegistryCycle("first-with-survivor", "second-with-survivor");
+ assertHistogramState(CollectorRegistry.defaultRegistry, "resident-default", 1, 0.005);
+ }
+ }
+ assertHistogramState(CollectorRegistry.defaultRegistry, "resident-default", 0, 0);
+ }
+
+ private void assertDisabledDefaultRegistryCycle(String firstCluster, String secondCluster) {
+ Meter disabled = OpenTelemetry.noop().getMeter("disabled");
+ try (var first = new ScalableTopicControllerMetrics(disabled, firstCluster,
+ () -> new ScalableTopicControllerMetrics.Counts(1, 2, 3));
+ var second = new ScalableTopicControllerMetrics(disabled, secondCluster,
+ () -> new ScalableTopicControllerMetrics.Counts(4, 5, 6))) {
+ first.recordBlocked("max_segments");
+ first.recordTakeover(0.5, true);
+ second.recordOperation("merge", "auto", "partial", 1);
+ var samples = Collections.list(CollectorRegistry.defaultRegistry.metricFamilySamples());
+ assertThat(sample(samples, "controller_count", Map.of("cluster", firstCluster))).isEqualTo(1);
+ assertThat(sample(samples, "controller_count", Map.of("cluster", secondCluster))).isEqualTo(4);
+ assertThat(sample(samples, "auto_scale_blocked_total",
+ Map.of("cluster", firstCluster, "auto_scale_reason", "max_segments"))).isEqualTo(1);
+ assertThat(sample(samples, "controller_takeover_duration_seconds_count",
+ Map.of("cluster", firstCluster, "operation_status", "success"))).isEqualTo(1);
+ assertThat(sample(samples, "layout_operation_total", Map.of("cluster", secondCluster,
+ "operation", "merge", "operation_trigger", "auto", "operation_status", "partial"))).isEqualTo(1);
+ assertThat(reader.collectAllMetrics()).isEmpty();
+ }
+ assertThat(Collections.list(CollectorRegistry.defaultRegistry.metricFamilySamples()))
+ .flatExtracting(family -> family.samples)
+ .noneSatisfy(value -> assertThat(labels(value)).containsEntry("cluster", firstCluster));
+ }
+
+ @DataProvider
+ public Object[][] disabledProviderLayouts() {
+ return new Object[][]{{true, true}, {true, false}, {false, true}, {false, false}};
+ }
+
+ @Test(dataProvider = "disabledProviderLayouts")
+ public void testDisabledProvidersOwnHistogramLifetimes(boolean sameCluster, boolean sameRegistry) {
+ Meter disabled = OpenTelemetry.noop().getMeter("resident");
+ assertThat(OpenTelemetry.noop().getMeter("transient")).isSameAs(disabled);
+ String residentCluster = "resident-" + sameCluster + "-" + sameRegistry;
+ String transientCluster = sameCluster ? residentCluster : "transient-" + sameRegistry;
+ CollectorRegistry transientRegistry = sameRegistry ? registry : new CollectorRegistry();
+ try (var resident = new ScalableTopicControllerMetrics(disabled, residentCluster, () -> EMPTY, registry)) {
+ recordHistograms(resident, 0.005);
+ assertHistogramState(registry, residentCluster, 1, 0.005);
+ for (int attempt = 0; attempt < 2; attempt++) {
+ try (var transientMetrics =
+ new ScalableTopicControllerMetrics(disabled, transientCluster,
+ () -> EMPTY, transientRegistry)) {
+ recordHistograms(transientMetrics, 0.01);
+ boolean aggregate = sameCluster && sameRegistry;
+ assertHistogramState(registry, residentCluster, aggregate ? 2 : 1, aggregate ? 0.015 : 0.005);
+ assertHistogramState(transientRegistry, transientCluster,
+ aggregate ? 2 : 1, aggregate ? 0.015 : 0.01);
+ if (!sameCluster && !sameRegistry) {
+ assertHistogramState(registry, transientCluster, 0, 0);
+ assertHistogramState(transientRegistry, residentCluster, 0, 0);
+ }
+ }
+ assertHistogramState(registry, residentCluster, 1, 0.005);
+ if (!sameCluster || !sameRegistry) {
+ assertHistogramState(transientRegistry, transientCluster, 0, 0);
+ }
+ }
+ }
+ assertHistogramState(registry, residentCluster, 0, 0);
+ }
+
+ private static void recordHistograms(ScalableTopicControllerMetrics metrics, double seconds) {
+ metrics.recordTakeover(seconds, true);
+ metrics.recordOperation("split", "manual", "success", seconds);
+ metrics.recordSeal("split", seconds);
+ }
+
+ private static void assertSdkHistogramState(Collection<MetricData> metrics, long count, double sum) {
+ for (String name : List.of("controller.takeover.duration", "layout.operation.duration",
+ "layout.seal.duration")) {
+ assertThat(metric(metrics, name).getHistogramData().getPoints()).singleElement().satisfies(point -> {
+ assertThat(point.getCount()).as("%s SDK count", name).isEqualTo(count);
+ assertThat(point.getSum()).as("%s SDK sum", name).isCloseTo(sum, offset(0.000001));
+ });
+ }
+ }
+
+ private static void assertHistogramState(CollectorRegistry registry, String cluster, long count, double sum) {
+ Map<String, Map<String, String>> series = Map.of(
+ "controller_takeover_duration_seconds", Map.of("cluster", cluster, "operation_status", "success"),
+ "layout_operation_duration_seconds",
+ Map.of("cluster", cluster, "operation", "split", "operation_status", "success"),
+ "layout_seal_duration_seconds", Map.of("cluster", cluster, "operation", "split"));
+ series.forEach((name, labels) -> {
+ List<String> names = List.of("cluster", "operation", "operation_status").stream()
+ .filter(labels::containsKey).toList();
+ String[] labelNames = names.toArray(String[]::new);
+ String[] labelValues = names.stream().map(labels::get).toArray(String[]::new);
+ Double actualCount = registry.getSampleValue(PROM_PREFIX + name + "_count", labelNames, labelValues);
+ Double actualSum = registry.getSampleValue(PROM_PREFIX + name + "_sum", labelNames, labelValues);
+ if (count == 0) {
+ assertThat(actualCount).as("%s count for closed provider %s", name, cluster).isNull();
+ assertThat(actualSum).as("%s sum for closed provider %s", name, cluster).isNull();
+ } else {
+ assertThat(actualCount).as("%s count for %s", name, cluster).isEqualTo((double) count);
+ assertThat(actualSum).as("%s sum for %s", name, cluster).isCloseTo(sum, offset(0.000001));
+ }
+ });
+ }
+
+ @Test
+ public void testCountersAreRegistryLocalAndNoopNeverRegisters() {
+ var otherRegistry = new CollectorRegistry();
+ try (var first = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry);
+ var second = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, otherRegistry)) {
+ first.recordBlocked("invalid_policy");
+ assertThat(sample(Collections.list(registry.metricFamilySamples()), "auto_scale_blocked_total",
+ Map.of("cluster", CLUSTER, "auto_scale_reason", "invalid_policy"))).isEqualTo(1);
+ assertThat(Collections.list(otherRegistry.metricFamilySamples())).hasSize(2);
+ ScalableTopicControllerMetrics.NOOP.recordBlocked("arbitrary");
+ ScalableTopicControllerMetrics.NOOP.recordOperation(null, null, null, Double.NaN);
+ ScalableTopicControllerMetrics.NOOP.recordTakeover(Double.NaN, false);
+ ScalableTopicControllerMetrics.NOOP.recordSeal(null, Double.NaN);
+ ScalableTopicControllerMetrics.NOOP.close();
+ assertThat(Collections.list(otherRegistry.metricFamilySamples())).hasSize(2);
+ }
+ }
+
+ @Test
+ public void testRejectsUnboundedLabelsAndInvalidDurations() {
+ try (var metrics = new ScalableTopicControllerMetrics(meter, CLUSTER, () -> EMPTY, registry)) {
+ assertThatThrownBy(() -> metrics.recordOperation("topic-name", "auto", "success", 1))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordOperation("split", "unknown", "success", 1))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordOperation("split", "auto", "unknown", 1))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordBlocked("topic-name")).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordSeal("prune", 1)).isInstanceOf(IllegalArgumentException.class);
+ for (double duration : new double[]{-1, Double.NaN, Double.POSITIVE_INFINITY}) {
+ assertThatThrownBy(() -> metrics.recordTakeover(duration, true))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordOperation("split", "auto", "success", duration))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> metrics.recordSeal("split", duration))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ assertThat(Collections.list(registry.metricFamilySamples())).hasSize(2);
+ }
+ }
+
+ private static void assertCounterParity(Collection<MetricData> otel, List<MetricFamilySamples> prometheus,
+ String otelName, String prometheusName, long expectedTotal) {
+ var data = metric(otel, otelName).getLongSumData();
+ assertThat(data.isMonotonic()).isTrue();
+ long total = 0;
+ for (var point : data.getPoints()) {
+ assertThat(sample(prometheus, prometheusName, labels(point.getAttributes()))).isEqualTo(point.getValue());
+ total += point.getValue();
+ }
+ assertThat(total).isEqualTo(expectedTotal);
+ }
+
+ private static void assertHistogramParity(Collection<MetricData> otel, List<MetricFamilySamples> prometheus,
+ String otelName, String prometheusName, long expectedCount,
+ double expectedSum) {
+ var metric = metric(otel, otelName);
+ assertThat(metric.getUnit()).isEqualTo("s");
+ long totalCount = 0;
+ double totalSum = 0;
+ for (var point : metric.getHistogramData().getPoints()) {
+ assertThat(point.getBoundaries())
+ .containsExactlyElementsOf(ScalableTopicControllerMetrics.DURATION_BUCKETS);
+ Map<String, String> labels = labels(point.getAttributes());
+ assertThat(sample(prometheus, prometheusName + "_count", labels)).isEqualTo(point.getCount());
+ assertThat(sample(prometheus, prometheusName + "_sum", labels)).isEqualTo(point.getSum());
+ long cumulative = 0;
+ for (int i = 0; i < point.getCounts().size(); i++) {
+ cumulative += point.getCounts().get(i);
+ Map<String, String> bucketLabels = new HashMap<>(labels);
+ bucketLabels.put("le", i < point.getBoundaries().size()
+ ? Collector.doubleToGoString(point.getBoundaries().get(i)) : "+Inf");
+ assertThat(sample(prometheus, prometheusName + "_bucket", bucketLabels)).isEqualTo(cumulative);
+ }
+ totalCount += point.getCount();
+ totalSum += point.getSum();
+ }
+ assertThat(totalCount).isEqualTo(expectedCount);
+ assertThat(totalSum).isCloseTo(expectedSum, offset(0.000001));
+ }
+
+ private static void assertGauges(Collection<MetricData> otel, List<MetricFamilySamples> prometheus,
+ long controllers, long active, long sealed) {
+ var controllerData = metric(otel, "controller.count").getLongSumData();
+ assertThat(controllerData.isMonotonic()).isFalse();
+ assertThat(controllerData.getPoints()).singleElement().satisfies(point -> {
+ assertThat(point.getAttributes()).isEqualTo(Attributes.empty());
+ assertThat(point.getValue()).isEqualTo(controllers);
+ });
+ var segmentData = metric(otel, "segment.count").getLongSumData();
+ assertThat(segmentData.isMonotonic()).isFalse();
+ assertThat(segmentData.getPoints()).hasSize(2).allSatisfy(point -> assertThat(point.getValue())
+ .isEqualTo(point.getAttributes().equals(Attributes.builder()
+ .put(ATTRIBUTE_PREFIX + "segment.state", "active").build()) ? active : sealed));
+ assertThat(sample(prometheus, "controller_count", Map.of("cluster", CLUSTER))).isEqualTo(controllers);
+ assertThat(sample(prometheus, "segment_count", Map.of("cluster", CLUSTER, "segment_state", "active")))
+ .isEqualTo(active);
+ assertThat(sample(prometheus, "segment_count", Map.of("cluster", CLUSTER, "segment_state", "sealed")))
+ .isEqualTo(sealed);
+ }
+
+ private static MetricData metric(Collection<MetricData> metrics, String name) {
+ return metrics.stream().filter(metric -> metric.getName().equals(OTEL_PREFIX + name)).findFirst().orElseThrow();
+ }
+
+ private static Map<String, String> labels(Attributes attributes) {
+ Map<String, String> result = new HashMap<>(Map.of("cluster", CLUSTER));
+ attributes.forEach((key, value) -> {
+ assertThat(key.getKey()).startsWith(ATTRIBUTE_PREFIX);
+ result.put(key.getKey().substring(ATTRIBUTE_PREFIX.length()).replace('.', '_'), value.toString());
+ });
+ return result;
+ }
+
+ private static Map<String, String> labels(Sample sample) {
+ Map<String, String> result = new HashMap<>();
+ for (int i = 0; i < sample.labelNames.size(); i++) {
+ result.put(sample.labelNames.get(i), sample.labelValues.get(i));
+ }
+ return result;
+ }
+
+ private static double sample(List<MetricFamilySamples> families, String name, Map<String, String> labels) {
+ return families.stream().flatMap(family -> family.samples.stream())
+ .filter(sample -> sample.name.equals(PROM_PREFIX + name) && labels(sample).equals(labels))
+ .mapToDouble(sample -> sample.value).findFirst().orElseThrow();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
index 2f2cc626693b..43b461a0ccf9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
@@ -20,13 +20,16 @@ package org.apache.pulsar.broker.service.scalable;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyDouble;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -107,6 +110,7 @@ public class ScalableTopicControllerTest {
private TopicName topicName;
private ScalableTopicController controller;
+ private ScalableTopicControllerMetrics metrics;
@BeforeMethod
public void setUp() throws Exception {
@@ -160,6 +164,7 @@ public class ScalableTopicControllerTest {
when(scalableTopics.getSegmentStatsAsync(anyString(), anyLong()))
.thenAnswer(inv -> CompletableFuture.completedFuture(segmentTopicStats()));
+ metrics = mock(ScalableTopicControllerMetrics.class);
controller = newController(topicName);
}
@@ -188,14 +193,192 @@ public class ScalableTopicControllerTest {
}
private ScalableTopicController newController(TopicName tn) {
- return new ScalableTopicController(tn, resources, brokerService, coordinationService);
+ ScalableTopicController result = new ScalableTopicController(tn, resources, brokerService,
+ coordinationService);
+ result.setMetrics(metrics);
+ return result;
}
private ScalableTopicController newControllerWithClock(TopicName tn,
java.time.Clock clock,
java.time.Duration gcInterval) {
- return new ScalableTopicController(tn, resources, brokerService, coordinationService,
+ ScalableTopicController result = new ScalableTopicController(tn, resources, brokerService, coordinationService,
clock, gcInterval);
+ result.setMetrics(metrics);
+ return result;
+ }
+
+ @Test(timeOut = 10_000)
+ public void testMetricCountsDoNotWaitForControllerMonitor() throws Exception {
+ synchronized (controller) {
+ assertEquals(CompletableFuture.supplyAsync(controller::metricCounts, scheduler).get(5, TimeUnit.SECONDS),
+ new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+ }
+ }
+
+ @Test
+ public void testLayoutMetricsAndLeaderCounts() throws Exception {
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+ controller.initialize().get();
+ verify(metrics).recordTakeover(anyDouble(), eq(true));
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(1, 4, 0));
+
+ controller.splitSegment(0).get();
+ verify(metrics).recordOperation(eq("split"), eq("manual"), eq("success"), anyDouble());
+ verify(metrics).recordSeal(eq("split"), anyDouble());
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(1, 5, 1));
+
+ controller.mergeSegments(1, 2).get();
+ verify(metrics).recordOperation(eq("merge"), eq("manual"), eq("success"), anyDouble());
+ verify(metrics).recordSeal(eq("merge"), anyDouble());
+ controller.rebucketSegment(3, 2).get();
+ verify(metrics).recordOperation(eq("rebucket"), eq("manual"), eq("success"), anyDouble());
+ verify(metrics).recordSeal(eq("rebucket"), anyDouble());
+ controller.close().get();
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+ }
+
+ @Test
+ public void testLayoutMetricFailureBeforeSealing() throws Exception {
+ controller.initialize().get();
+ when(scalableTopics.createSegmentAsync(anyString(), any()))
+ .thenReturn(CompletableFuture.failedFuture(new IllegalStateException("create failed")));
+ assertThrows(ExecutionException.class, () -> controller.splitSegment(0).get());
+ verify(metrics).recordOperation(eq("split"), eq("manual"), eq("failure"), anyDouble());
+ verify(metrics, never()).recordSeal(anyString(), anyDouble());
+ }
+
+ @Test
+ public void testLayoutMetricPartialAfterSealBeforeCommit() throws Exception {
+ controller.close().get();
+ resources = spy(resources);
+ controller = newController(topicName);
+ controller.initialize().get();
+ doReturn(CompletableFuture.failedFuture(new IllegalStateException("CAS failed")))
+ .when(resources).updateScalableTopicAsync(eq(topicName), any());
+
+ assertThrows(ExecutionException.class, () -> controller.splitSegment(0).get());
+ verify(scalableTopics).terminateSegmentAsync(anyString());
+ verify(metrics).recordOperation(eq("split"), eq("manual"), eq("partial"), anyDouble());
+ verify(metrics, never()).recordSeal(anyString(), anyDouble());
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(1, 4, 0));
+ }
+
+ private record PendingTakeover(CompletableFuture<Void> initialization,
+ CompletableFuture<List<String>> restore) { }
+
+ private PendingTakeover startPendingTakeover() throws Exception {
+ controller.close().get(5, TimeUnit.SECONDS);
+ resources = spy(resources);
+ CompletableFuture<List<String>> restore = new CompletableFuture<>();
+ CompletableFuture<Void> entered = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ entered.complete(null);
+ return restore;
+ }).when(resources).listSubscriptionsAsync(topicName);
+ controller = newController(topicName);
+ CompletableFuture<Void> initialization = controller.initialize();
+ entered.get(5, TimeUnit.SECONDS);
+ return new PendingTakeover(initialization, restore);
+ }
+
+ @Test
+ public void testClosedTakeoverIsNotRecordedAsFailure() throws Exception {
+ PendingTakeover pending = startPendingTakeover();
+ try {
+ controller.close().get(5, TimeUnit.SECONDS);
+ pending.restore().complete(List.of());
+ pending.initialization().handle((__, ex) -> null).get(5, TimeUnit.SECONDS);
+ verify(metrics, never()).recordTakeover(anyDouble(), anyBoolean());
+ } finally {
+ pending.restore().complete(List.of());
+ }
+ }
+
+ @Test
+ public void testSuccessfulTakeoverRecordedBeforeWaitersResume() throws Exception {
+ PendingTakeover pending = startPendingTakeover();
+ try {
+ CompletableFuture<Void> closedByWaiter = controller.takeOverInProgress()
+ .thenCompose(__ -> controller.close());
+ pending.restore().complete(List.of());
+ pending.initialization().get(5, TimeUnit.SECONDS);
+ closedByWaiter.get(5, TimeUnit.SECONDS);
+ verify(metrics).recordTakeover(anyDouble(), eq(true));
+ verify(metrics, never()).recordTakeover(anyDouble(), eq(false));
+ assertTrue(controller.isClosed());
+ } finally {
+ pending.restore().complete(List.of());
+ }
+ }
+
+ @Test
+ public void testFailedTakeoverMetric() throws Exception {
+ controller.close().get();
+ resources = spy(resources);
+ controller = newController(topicName);
+ doReturn(CompletableFuture.failedFuture(new IllegalStateException("takeover failed")))
+ .when(resources).listSubscriptionsAsync(topicName);
+ assertThrows(ExecutionException.class, () -> controller.initialize().get());
+ verify(metrics).recordTakeover(anyDouble(), eq(false));
+ assertEquals(controller.metricCounts(), new ScalableTopicControllerMetrics.Counts(0, 0, 0));
+ }
+
+ @Test
+ public void testPruneMetricStorageDeleteFailureIsPartial() throws Exception {
+ installGcMocks(0);
+ controller.initialize().get();
+ controller.splitSegment(0).get();
+ when(scalableTopics.deleteSegmentAsync(anyString(), anyBoolean()))
+ .thenReturn(CompletableFuture.failedFuture(new IllegalStateException("storage failed")));
+ controller.runGcTickAsync().get();
+ verify(metrics).recordOperation(eq("prune"), eq("auto"), eq("partial"), anyDouble());
+ assertFalse(controller.getLayout().get().getAllSegments().containsKey(0L));
+ }
+
+ @Test
+ public void testPruneLoadCleanupFailureIsNotPartial() throws Exception {
+ installGcMocks(0);
+ controller.close().get();
+ resources = spy(resources);
+ controller = newController(topicName);
+ controller.initialize().get();
+ controller.splitSegment(0).get();
+ doReturn(CompletableFuture.failedFuture(new IllegalStateException("load cleanup failed")))
+ .when(resources).deleteSegmentLoadAsync(topicName, 0);
+ assertThrows(ExecutionException.class, () -> controller.runGcTickAsync().get());
+ verify(scalableTopics).deleteSegmentAsync(anyString(), anyBoolean());
+ verify(metrics).recordOperation(eq("prune"), eq("auto"), eq("failure"), anyDouble());
+ verify(metrics, never()).recordOperation(eq("prune"), eq("auto"), eq("partial"), anyDouble());
+ }
+
+ @Test
+ public void testOverlappingGcTicksCountOnlyLayoutChanges() throws Exception {
+ installGcMocks(0);
+ controller.close().get();
+ resources = spy(resources);
+ controller = newController(topicName);
+ controller.initialize().get();
+ controller.splitSegment(0).get();
+ CompletableFuture<List<String>> subscriptions = new CompletableFuture<>();
+ doReturn(subscriptions).when(resources).listSubscriptionsAsync(topicName);
+ CompletableFuture<Void> first = controller.runGcTickAsync();
+ CompletableFuture<Void> second = controller.runGcTickAsync();
+ // Both ticks captured the same layout, and reach their CAS only after this completes.
+ subscriptions.complete(List.of());
+ CompletableFuture.allOf(first, second).get();
+ verify(metrics).recordOperation(eq("prune"), eq("auto"), eq("success"), anyDouble());
+ }
+
+ @Test
+ public void testMergeSecondSealFailureIsPartial() throws Exception {
+ controller.initialize().get();
+ when(scalableTopics.terminateSegmentAsync(anyString()))
+ .thenReturn(CompletableFuture.completedFuture(null))
+ .thenReturn(CompletableFuture.failedFuture(new IllegalStateException("second seal failed")));
+ assertThrows(ExecutionException.class, () -> controller.mergeSegments(0, 1).get());
+ verify(metrics).recordOperation(eq("merge"), eq("manual"), eq("partial"), anyDouble());
+ verify(metrics, never()).recordSeal(anyString(), anyDouble());
}
// --- initialize() ---
@@ -425,8 +608,8 @@ public class ScalableTopicControllerTest {
public void testWriteOperationThrowsWhenNotLeader() {
// Controller not initialized → not leader.
assertFalse(controller.isLeader());
- assertThrows(IllegalStateException.class, () -> controller.splitSegment(0));
- assertThrows(IllegalStateException.class, () -> controller.mergeSegments(0, 1));
+ assertThrows(ExecutionException.class, () -> controller.splitSegment(0).get());
+ assertThrows(ExecutionException.class, () -> controller.mergeSegments(0, 1).get());
assertThrows(IllegalStateException.class,
() -> controller.createSubscription("sub", SubscriptionType.STREAM));
assertThrows(IllegalStateException.class, () -> controller.deleteSubscription("sub"));
@@ -994,6 +1177,7 @@ public class ScalableTopicControllerTest {
"tick past retention must prune the sealed segment");
// Backing topic delete was issued via the segment-aware admin call.
verify(scalableTopics).deleteSegmentAsync(anyString(), anyBoolean());
+ verify(metrics).recordOperation(eq("prune"), eq("auto"), eq("success"), anyDouble());
// The pruned segment's load record is deleted along with it.
assertFalse(resources.getSegmentLoadAsync(topicName, 0).get().isPresent(),
"prune must delete the segment's load record");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
index 33230de6f970..2e673236ca9d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
@@ -18,14 +18,17 @@
*/
package org.apache.pulsar.broker.service.scalable;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -184,6 +187,78 @@ public class ScalableTopicServiceTest {
return TopicName.get("topic://tenant/ns/" + name);
}
+ @Test
+ public void testBrokerMetricsAggregateAndReleaseControllers() throws Exception {
+ TopicName first = scalableTopic("metrics-first");
+ TopicName second = scalableTopic("metrics-second");
+ service.createScalableTopic(first, 2).get();
+ service.createScalableTopic(second, 3).get();
+ ScalableTopicController leader = service.getOrCreateController(first).get();
+ service.getOrCreateController(second).get();
+ assertThat(prometheusValue("controller_count")).isEqualTo(2.0);
+ assertThat(prometheusSegments("active")).isEqualTo(5.0);
+ assertThat(prometheusSegments("sealed")).isZero();
+
+ leader.splitSegment(0).get();
+ assertThat(prometheusSegments("active")).isEqualTo(6.0);
+ assertThat(prometheusSegments("sealed")).isEqualTo(1.0);
+ service.releaseController(first).get();
+ assertThat(prometheusValue("controller_count")).isEqualTo(1.0);
+ assertThat(prometheusSegments("active")).isEqualTo(3.0);
+ assertThat(prometheusSegments("sealed")).isZero();
+ service.close();
+ assertThat(prometheusValue("controller_count")).isNull();
+ }
+
+ @Test
+ public void testMetricsCleanupDoesNotChangeControllerLifecycle() throws Exception {
+ service.close();
+ ScalableTopicResources observedResources = spy(resources);
+ service = new ScalableTopicService(brokerService, observedResources, coordinationService);
+ TopicName topic = scalableTopic("metrics-cleanup");
+ service.createScalableTopic(topic, 1).get();
+ ScalableTopicController controller = service.getOrCreateController(topic).get();
+ Object[] tasks = scheduler.getQueue().toArray();
+ clearInvocations(observedResources, scalableTopicsAdmin);
+
+ service.closeMetrics();
+
+ verifyNoInteractions(observedResources, scalableTopicsAdmin);
+ assertThat(prometheusValue("controller_count")).isNull();
+ assertThat(controller.isClosed()).isFalse();
+ assertThat(controller.isLeader()).isTrue();
+ assertThat(store.exists(resources.controllerLockPath(topic)).get()).isTrue();
+ assertThat(scheduler.getQueue().toArray()).containsExactlyInAnyOrder(tasks);
+ }
+
+ @Test(timeOut = 10_000)
+ public void testMetricsCleanupDoesNotWaitForControllerInitialization() throws Exception {
+ TopicName topic = scalableTopic("metrics-cleanup-pending-init");
+ service.createScalableTopic(topic, 1).get();
+ CompletableFuture<Void> create = new CompletableFuture<>();
+ when(scalableTopicsAdmin.createSegmentAsync(anyString(), any())).thenReturn(create);
+ CompletableFuture<ScalableTopicController> initializing = service.getOrCreateController(topic);
+ try {
+ assertThat(initializing).isNotDone();
+ service.closeMetrics();
+ assertThat(prometheusValue("controller_count")).isNull();
+ assertThat(initializing).isNotDone();
+ } finally {
+ create.complete(null);
+ initializing.get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ private Double prometheusValue(String suffix) {
+ return CollectorRegistry.defaultRegistry.getSampleValue("pulsar_scalable_topic_" + suffix,
+ new String[]{"cluster"}, new String[]{clusterName});
+ }
+
+ private Double prometheusSegments(String state) {
+ return CollectorRegistry.defaultRegistry.getSampleValue("pulsar_scalable_topic_segment_count",
+ new String[]{"cluster", "segment_state"}, new String[]{clusterName, state});
+ }
+
// --- createScalableTopic ---
@Test