You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/07/14 04:19:57 UTC
(pinot) branch master updated: Add server-level consumption rate limit observability metrics and fix byte-mode overflow (#18942)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ecb288d666d Add server-level consumption rate limit observability metrics and fix byte-mode overflow (#18942)
ecb288d666d is described below
commit ecb288d666d4918b1a30e8dc185e0e761036db55
Author: NOOB <ha...@startree.ai>
AuthorDate: Tue Jul 14 09:49:51 2026 +0530
Add server-level consumption rate limit observability metrics and fix byte-mode overflow (#18942)
* Add server-level consumption rate limit observability metrics and fix byte-mode overflow
Server-level realtime consumption rate limiting had two observability gaps:
1. The server-wide quota utilization was published through the per-table
CONSUMPTION_QUOTA_UTILIZATION gauge keyed by the meter name "realtimeRowsConsumed".
Being a server-wide value, exporters mapped it to a misleading, per-table-shaped
series that server-level dashboards could not match.
2. QuotaUtilizationTracker/AsyncMetricEmitter aggregated the per-minute unit count in
an int and cast the LongAdder sum to int in emit(). In byte-based throttling mode
the per-minute byte sum for a busy server exceeds Integer.MAX_VALUE (~35.8 MB/s over
the 60s window), overflowing to a negative utilization.
There was also no metric exposing the configured rate limit, so a dashboard could not
show the cap or whether limiting was enabled.
Changes:
- Add SERVER_CONSUMPTION_QUOTA_UTILIZATION (global) and emit the server-level
utilization through it instead of the per-table gauge.
- Add SERVER_CONSUMPTION_RATE_LIMIT (global) and CONSUMPTION_RATE_LIMIT (per-table),
emitting the configured caps on limiter setup/config change (server gauge is set to
-1 when disabled; server utilization is reset to 0 on disable so the two server
gauges stay consistent).
- Widen the per-minute aggregate and the emit() read to long to fix the byte-mode
overflow.
Backward-incompatible metric change: the server-wide quota utilization moves from
consumptionQuotaUtilization{table="realtimeRowsConsumed"} to the new global gauge
serverConsumptionQuotaUtilization; dashboards keyed on the old series should switch.
* Add listener-level test verifying the rate-limit gauge updates on cluster config change
* Do not reset the server utilization gauge on disable
The utilization gauge is a live measurement that stops updating once the
limiter is closed; on a rate-limit update the emitter keeps running so it
self-corrects, and on disable the -1 cap already signals any last value is
stale. Leaving it untouched keeps it consistent with the per-partition
utilization gauge, which is likewise not reset on removal.
* Remove per-partition rate limit gauges when the limit is removed
A partition rate limit change only takes effect when the next consuming
segment creates its rate limiter. When that creation finds no configured
limit (i.e. the limit was removed), remove the per-partition cap and
utilization gauges so they do not linger with stale values. Removing a
never-emitted gauge is a no-op, so this is safe for partitions that never
had a limit and creates no new series.
* Add lifecycle test: partition rate limit set, removed, then set again
Uses a real metrics registry to verify the cap gauge is registered with the
configured value, removed when a consumer is created without a limit, and
cleanly re-registered with the new value when the limit is configured again.
---
.../apache/pinot/common/metrics/ServerGauge.java | 8 +
.../realtime/RealtimeConsumptionRateManager.java | 84 +++++++++--
.../RealtimeConsumptionRateManagerTest.java | 161 ++++++++++++++++++++-
.../ServerRateLimitConfigChangeListenerTest.java | 17 +++
4 files changed, 254 insertions(+), 16 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
index 0a909ce4341..312ace46ee4 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java
@@ -48,6 +48,14 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
// Dedup metrics
DEDUP_PRIMARY_KEYS_COUNT("dedupPrimaryKeysCount", false),
CONSUMPTION_QUOTA_UTILIZATION("ratio", false),
+ // Server-level consumption rate limiting is server-wide, so these are global gauges (no table/partition
+ // dimension), kept distinct from the per-partition CONSUMPTION_QUOTA_UTILIZATION above.
+ SERVER_CONSUMPTION_QUOTA_UTILIZATION("ratio", true),
+ // Configured consumption rate limits, emitted when the limiter is created or its config changes; the server gauge
+ // is set to -1 when server-level rate limiting is disabled. Units follow the active throttling strategy
+ // (bytes/sec in byte mode, messages/sec in message mode).
+ SERVER_CONSUMPTION_RATE_LIMIT("perSecond", true),
+ CONSUMPTION_RATE_LIMIT("perSecond", false),
JVM_HEAP_USED_BYTES("bytes", true),
NETTY_POOLED_USED_DIRECT_MEMORY("bytes", true),
NETTY_POOLED_USED_HEAP_MEMORY("bytes", true),
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java
index 56d54ffdd3b..1c0ce0577dd 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java
@@ -35,7 +35,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.LongAdder;
import org.apache.pinot.common.metrics.ServerGauge;
-import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.stream.MessageBatch;
@@ -65,8 +64,6 @@ public class RealtimeConsumptionRateManager {
private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeConsumptionRateManager.class);
private static final int CACHE_ENTRY_EXPIRATION_TIME_IN_MINUTES = 10;
- private static final String SERVER_CONSUMPTION_RATE_METRIC_KEY_NAME =
- ServerMeter.REALTIME_ROWS_CONSUMED.getMeterName();
private volatile ConsumptionRateLimiter _serverRateLimiter = NOOP_RATE_LIMITER;
// stream config object is required for fetching the partition count from the stream
@@ -147,6 +144,11 @@ public class RealtimeConsumptionRateManager {
// result of it, But the metric related to throttling won't be emitted since as a result of here above the
// AsyncMetricEmitter will be closed. It's recommended to forceCommit segments to avoid this.
}
+ // Expose the configured cap (-1 when disabled) so operators can see that rate limiting is off. The server
+ // utilization gauge is intentionally not touched here: it is a global gauge seeded at server startup, so
+ // removing it would be undone on restart, and rate-limit updates keep the emitter running so it self-corrects
+ // on the common update path. The -1 cap already signals that any last utilization value is stale.
+ emitServerRateLimit(serverMetrics, -1);
return;
}
@@ -154,11 +156,42 @@ public class RealtimeConsumptionRateManager {
ServerRateLimiter existingLimiter = (ServerRateLimiter) _serverRateLimiter;
existingLimiter.updateRateLimit(serverRateLimitConfig._serverRateLimit,
serverRateLimitConfig._throttlingStrategy);
+ emitServerRateLimit(serverMetrics, serverRateLimitConfig._serverRateLimit);
return;
}
_serverRateLimiter = new ServerRateLimiter(serverRateLimitConfig._serverRateLimit, serverMetrics,
- SERVER_CONSUMPTION_RATE_METRIC_KEY_NAME, serverRateLimitConfig._throttlingStrategy);
+ serverRateLimitConfig._throttlingStrategy);
+ emitServerRateLimit(serverMetrics, serverRateLimitConfig._serverRateLimit);
+ }
+
+ // Exposes the configured server-level consumption rate limit as a global gauge so operators can see the cap
+ // (and whether it is enabled) without inspecting server config. Set on setup and on each config change, mirroring
+ // the configured-threshold gauge pattern used by SegmentOperationsThrottler.
+ private static void emitServerRateLimit(ServerMetrics serverMetrics, double rateLimit) {
+ if (serverMetrics != null) {
+ // Round rather than truncate so a sub-1 limit does not report 0 and collide with the -1 "disabled" sentinel.
+ serverMetrics.setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, Math.round(rateLimit));
+ }
+ }
+
+ // Exposes the configured per-partition consumption rate limit alongside its utilization gauge.
+ private static void emitPartitionRateLimit(ServerMetrics serverMetrics, String metricKeyName,
+ double partitionRateLimit) {
+ if (serverMetrics != null && metricKeyName != null) {
+ serverMetrics.setValueOfTableGauge(metricKeyName, ServerGauge.CONSUMPTION_RATE_LIMIT,
+ Math.round(partitionRateLimit));
+ }
+ }
+
+ // Removes the per-partition rate limit gauges when a consumer is created without a rate limit, so that a
+ // previously configured limit that has since been removed does not linger as stale series. Removing a gauge that
+ // was never emitted is a no-op, so this is safe (and cheap) for partitions that never had a limit.
+ private static void removePartitionRateLimitGauges(ServerMetrics serverMetrics, String metricKeyName) {
+ if (serverMetrics != null && metricKeyName != null) {
+ serverMetrics.removeTableGauge(metricKeyName, ServerGauge.CONSUMPTION_RATE_LIMIT);
+ serverMetrics.removeTableGauge(metricKeyName, ServerGauge.CONSUMPTION_QUOTA_UTILIZATION);
+ }
}
public void updateServerRateLimiter(ServerRateLimitConfig serverRateLimitConfig, ServerMetrics serverMetrics) {
@@ -177,10 +210,14 @@ public class RealtimeConsumptionRateManager {
if (partitionRateLimit > 0) {
LOGGER.info("A consumption rate limiter is set up for topic {} in table {} with partition rate limit: {}",
streamConfig.getTopicName(), tableName, partitionRateLimit);
+ emitPartitionRateLimit(serverMetrics, metricKeyName, partitionRateLimit);
return new PartitionRateLimiter(partitionRateLimit, serverMetrics, metricKeyName);
}
double topicRateLimit = streamConfig.getTopicConsumptionRateLimit();
if (topicRateLimit <= 0) {
+ // No rate limit configured (possibly removed since the previous consuming segment): clean up any gauges left
+ // behind by a previous limiter for this partition so the metrics reflect the current config.
+ removePartitionRateLimitGauges(serverMetrics, metricKeyName);
return NOOP_RATE_LIMITER;
}
int partitionCount;
@@ -194,6 +231,7 @@ public class RealtimeConsumptionRateManager {
LOGGER.info("A consumption rate limiter is set up for topic {} in table {} with rate limit: {} "
+ "(topic rate limit: {}, partition count: {})", streamConfig.getTopicName(), tableName, partitionRateLimit,
topicRateLimit, partitionCount);
+ emitPartitionRateLimit(serverMetrics, metricKeyName, partitionRateLimit);
return new PartitionRateLimiter(partitionRateLimit, serverMetrics, metricKeyName);
}
@@ -243,19 +281,33 @@ public class RealtimeConsumptionRateManager {
*/
static class QuotaUtilizationTracker {
private long _previousMinute = -1;
- private int _aggregateUnits = 0;
+ // Aggregated over a minute; must be long because in byte-throttling mode this counts bytes/minute, which for a
+ // busy server easily exceeds Integer.MAX_VALUE (~35.8 MB/s over the 60s window) and would otherwise overflow.
+ private long _aggregateUnits = 0;
private final ServerMetrics _serverMetrics;
private final String _metricKeyName;
+ private final boolean _serverLevel;
+ // Partition-level: emits the per-table/partition CONSUMPTION_QUOTA_UTILIZATION gauge.
public QuotaUtilizationTracker(ServerMetrics serverMetrics, String metricKeyName) {
+ this(serverMetrics, metricKeyName, false);
+ }
+
+ // Server-level: emits the server-wide (global) SERVER_CONSUMPTION_QUOTA_UTILIZATION gauge.
+ public QuotaUtilizationTracker(ServerMetrics serverMetrics) {
+ this(serverMetrics, null, true);
+ }
+
+ private QuotaUtilizationTracker(ServerMetrics serverMetrics, String metricKeyName, boolean serverLevel) {
_serverMetrics = serverMetrics;
_metricKeyName = metricKeyName;
+ _serverLevel = serverLevel;
}
/**
* Update count and return utilization ratio percentage (0 if not enough data yet).
*/
- public int update(int unitsConsumed, double rateLimit, Instant now) {
+ public int update(long unitsConsumed, double rateLimit, Instant now) {
int ratioPercentage = 0;
long nowInMinutes = now.getEpochSecond() / 60;
if (nowInMinutes == _previousMinute) {
@@ -264,8 +316,12 @@ public class RealtimeConsumptionRateManager {
if (_previousMinute != -1) { // not first time
double actualRate = _aggregateUnits / ((nowInMinutes - _previousMinute) * 60.0); // units per second
ratioPercentage = (int) Math.round(actualRate / rateLimit * 100);
- _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.CONSUMPTION_QUOTA_UTILIZATION,
- ratioPercentage);
+ if (_serverLevel) {
+ _serverMetrics.setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_QUOTA_UTILIZATION, ratioPercentage);
+ } else {
+ _serverMetrics.setValueOfTableGauge(_metricKeyName, ServerGauge.CONSUMPTION_QUOTA_UTILIZATION,
+ ratioPercentage);
+ }
}
_aggregateUnits = unitsConsumed;
_previousMinute = nowInMinutes;
@@ -274,7 +330,7 @@ public class RealtimeConsumptionRateManager {
}
@VisibleForTesting
- int getAggregateUnits() {
+ long getAggregateUnits() {
return _aggregateUnits;
}
}
@@ -383,10 +439,10 @@ public class RealtimeConsumptionRateManager {
private final AsyncMetricEmitter _metricEmitter;
private ThrottlingStrategy _throttlingStrategy;
- public ServerRateLimiter(double initialRateLimit, ServerMetrics serverMetrics, String metricKeyName,
+ public ServerRateLimiter(double initialRateLimit, ServerMetrics serverMetrics,
ThrottlingStrategy throttlingStrategy) {
_rateLimiter = RateLimiter.create(initialRateLimit);
- _metricEmitter = new AsyncMetricEmitter(serverMetrics, metricKeyName, initialRateLimit);
+ _metricEmitter = new AsyncMetricEmitter(serverMetrics, initialRateLimit);
_throttlingStrategy = throttlingStrategy;
_metricEmitter.start(); // start background emission
}
@@ -477,9 +533,9 @@ public class RealtimeConsumptionRateManager {
private final AtomicBoolean _running = new AtomicBoolean(false);
private final QuotaUtilizationTracker _tracker;
- public AsyncMetricEmitter(ServerMetrics serverMetrics, String metricKeyName, double initialRateLimit) {
+ public AsyncMetricEmitter(ServerMetrics serverMetrics, double initialRateLimit) {
_rateLimit = new AtomicDouble(initialRateLimit);
- _tracker = new QuotaUtilizationTracker(serverMetrics, metricKeyName);
+ _tracker = new QuotaUtilizationTracker(serverMetrics);
_executor = Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "server-rate-limit-metric-emitter");
t.setDaemon(true);
@@ -512,7 +568,7 @@ public class RealtimeConsumptionRateManager {
try {
double rateLimit = _rateLimit.get();
Instant now = Instant.now();
- int count = (int) _messageCount.sumThenReset();
+ long count = _messageCount.sumThenReset();
_tracker.update(count, rateLimit, now);
} catch (Exception e) {
LOGGER.warn("Encountered an error while emitting the metrics.", e);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManagerTest.java
index bcc40a7d9ca..93acf64c3ec 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManagerTest.java
@@ -26,19 +26,26 @@ import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.metrics.PinotMetricUtils;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.util.TestUtils;
import org.testng.annotations.Test;
import static org.apache.pinot.core.data.manager.realtime.RealtimeConsumptionRateManager.*;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
public class RealtimeConsumptionRateManagerTest {
@@ -260,7 +267,7 @@ public class RealtimeConsumptionRateManagerTest {
@Test
public void testAsyncMetricEmitter()
throws InterruptedException {
- AsyncMetricEmitter emitter = new AsyncMetricEmitter(mock(ServerMetrics.class), "testMetric", 10.0);
+ AsyncMetricEmitter emitter = new AsyncMetricEmitter(mock(ServerMetrics.class), 10.0);
try {
emitter.start(0, 1);
Thread.sleep(1500); // Let emitter run at-least once
@@ -276,7 +283,7 @@ public class RealtimeConsumptionRateManagerTest {
emitter.close();
}
- AsyncMetricEmitter emitter1 = new AsyncMetricEmitter(mock(ServerMetrics.class), "testMetric", 10.0);
+ AsyncMetricEmitter emitter1 = new AsyncMetricEmitter(mock(ServerMetrics.class), 10.0);
try {
emitter1.start(10, 10);
for (int i = 0; i < 20; i++) {
@@ -287,11 +294,161 @@ public class RealtimeConsumptionRateManagerTest {
== 0)), 5000,
"Expected messageCount to be greater than zero because messageCount will reset post initial delay (first "
+ "run).");
+ } finally {
+ emitter1.close();
+ }
+ }
+
+ @Test
+ public void testAsyncMetricEmitterByteOverflow()
+ throws InterruptedException {
+ // emit() previously cast the per-minute LongAdder sum to int; in byte mode the sum can exceed Integer.MAX_VALUE
+ // and would overflow to a negative aggregate. This exercises the fixed emit() path end to end.
+ AsyncMetricEmitter emitter = new AsyncMetricEmitter(mock(ServerMetrics.class), 100_000_000.0);
+ try {
+ emitter.record(2_000_000_000);
+ emitter.record(2_000_000_000); // total 4e9, exceeds Integer.MAX_VALUE
+ emitter.start(0, 3600); // fire emit() once, ~immediately
+ TestUtils.waitForCondition(
+ aVoid -> emitter.getMessageCount().sum() == 0 && emitter.getTracker().getAggregateUnits() == 4_000_000_000L,
+ 5000, "emit() must read the LongAdder sum as long without int truncation");
} finally {
emitter.close();
}
}
+ @Test
+ public void testQuotaUtilizationTrackerByteOverflow() {
+ // In byte-throttling mode the per-minute aggregate counts bytes and can exceed Integer.MAX_VALUE. This is a
+ // regression for a (int) overflow that previously turned high byte rates into a negative utilization.
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ QuotaUtilizationTracker tracker = new QuotaUtilizationTracker(serverMetrics); // server-level
+ double byteRateLimit = 100_000_000; // 100 MB/s
+ Instant now = Clock.fixed(Instant.parse("2022-08-10T12:00:02Z"), ZoneOffset.UTC).instant();
+ assertEquals(tracker.update(2_000_000_000, byteRateLimit, now), 0); // first minute, no emit
+ now = Clock.fixed(Instant.parse("2022-08-10T12:00:40Z"), ZoneOffset.UTC).instant();
+ assertEquals(tracker.update(4_000_000_000L, byteRateLimit, now), 0); // same minute: aggregate = 6e9 (> 2^32)
+ assertEquals(tracker.getAggregateUnits(), 6_000_000_000L); // long: not overflowed to a negative int
+ // next minute: 6e9 bytes / 60s = 100 MB/s == limit -> 100% (pre-fix the overflow produced a negative ratio)
+ now = Clock.fixed(Instant.parse("2022-08-10T12:01:02Z"), ZoneOffset.UTC).instant();
+ assertEquals(tracker.update(0, byteRateLimit, now), 100);
+ }
+
+ @Test
+ public void testQuotaUtilizationTrackerServerVsPartitionGauge() {
+ Instant firstMinute = Clock.fixed(Instant.parse("2022-08-10T12:00:02Z"), ZoneOffset.UTC).instant();
+ Instant secondMinute = Clock.fixed(Instant.parse("2022-08-10T12:01:02Z"), ZoneOffset.UTC).instant();
+
+ // Server-level tracker emits the server-wide (global) gauge, never the per-table gauge.
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ QuotaUtilizationTracker serverTracker = new QuotaUtilizationTracker(serverMetrics);
+ serverTracker.update(3000, 50, firstMinute);
+ serverTracker.update(0, 50, secondMinute);
+ verify(serverMetrics).setValueOfGlobalGauge(eq(ServerGauge.SERVER_CONSUMPTION_QUOTA_UTILIZATION), anyLong());
+ verify(serverMetrics, never()).setValueOfTableGauge(anyString(),
+ eq(ServerGauge.CONSUMPTION_QUOTA_UTILIZATION), anyLong());
+
+ // Partition-level tracker emits the per-table/partition gauge, never the server-wide gauge.
+ ServerMetrics partitionMetrics = mock(ServerMetrics.class);
+ QuotaUtilizationTracker partitionTracker =
+ new QuotaUtilizationTracker(partitionMetrics, "tableA_REALTIME-topicA-0");
+ partitionTracker.update(3000, 50, firstMinute);
+ partitionTracker.update(0, 50, secondMinute);
+ verify(partitionMetrics).setValueOfTableGauge(eq("tableA_REALTIME-topicA-0"),
+ eq(ServerGauge.CONSUMPTION_QUOTA_UTILIZATION), anyLong());
+ verify(partitionMetrics, never()).setValueOfGlobalGauge(
+ eq(ServerGauge.SERVER_CONSUMPTION_QUOTA_UTILIZATION), anyLong());
+ }
+
+ @Test
+ public void testServerRateLimitGaugeEmitted() {
+ RealtimeConsumptionRateManager manager = new RealtimeConsumptionRateManager(mock(LoadingCache.class));
+
+ // Byte rate limit configured -> gauge exposes the configured cap.
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ PinotConfiguration config = mock(PinotConfiguration.class);
+ when(config.getProperty(CommonConstants.Server.CONFIG_OF_SERVER_CONSUMPTION_RATE_LIMIT,
+ CommonConstants.Server.DEFAULT_SERVER_CONSUMPTION_RATE_LIMIT)).thenReturn(0.0);
+ when(config.getProperty(CommonConstants.Server.CONFIG_OF_SERVER_CONSUMPTION_RATE_LIMIT_BYTES,
+ CommonConstants.Server.DEFAULT_SERVER_CONSUMPTION_RATE_LIMIT)).thenReturn(112_000_000.0);
+ ServerRateLimiter limiter = (ServerRateLimiter) manager.createServerRateLimiter(config, metrics);
+ try {
+ verify(metrics).setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, 112_000_000L);
+ } finally {
+ limiter.close();
+ }
+
+ // Disabled -> gauge is set to -1 so operators can still see rate limiting is off.
+ ServerMetrics disabledMetrics = mock(ServerMetrics.class);
+ PinotConfiguration disabledConfig = mock(PinotConfiguration.class);
+ when(disabledConfig.getProperty(CommonConstants.Server.CONFIG_OF_SERVER_CONSUMPTION_RATE_LIMIT,
+ CommonConstants.Server.DEFAULT_SERVER_CONSUMPTION_RATE_LIMIT)).thenReturn(0.0);
+ when(disabledConfig.getProperty(CommonConstants.Server.CONFIG_OF_SERVER_CONSUMPTION_RATE_LIMIT_BYTES,
+ CommonConstants.Server.DEFAULT_SERVER_CONSUMPTION_RATE_LIMIT)).thenReturn(0.0);
+ manager.createServerRateLimiter(disabledConfig, disabledMetrics);
+ verify(disabledMetrics).setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, -1L);
+ }
+
+ @Test
+ public void testPartitionRateLimitGaugeEmitted() {
+ // Topic-derived: STREAM_CONFIG_A topic rate limit 50 over 10 partitions -> per-partition rate limit 5.
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_A, TABLE_NAME, metrics, "tableA_REALTIME-topicA-0");
+ verify(metrics).setValueOfTableGauge("tableA_REALTIME-topicA-0", ServerGauge.CONSUMPTION_RATE_LIMIT, 5L);
+
+ // Direct partition rate limit: STREAM_CONFIG_D specifies 4.0 per partition, used as-is.
+ ServerMetrics directMetrics = mock(ServerMetrics.class);
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_D, TABLE_NAME, directMetrics, "tableD_REALTIME-topicD-0");
+ verify(directMetrics).setValueOfTableGauge("tableD_REALTIME-topicD-0", ServerGauge.CONSUMPTION_RATE_LIMIT, 4L);
+
+ // No rate limit configured (STREAM_CONFIG_C) -> NOOP limiter, no gauge emitted, and any gauges left behind by a
+ // previously configured (since removed) limit are cleaned up so they do not linger as stale series.
+ ServerMetrics noneMetrics = mock(ServerMetrics.class);
+ ConsumptionRateLimiter limiter =
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_C, TABLE_NAME, noneMetrics, "keyC");
+ assertEquals(limiter, NOOP_RATE_LIMITER);
+ verify(noneMetrics, never()).setValueOfTableGauge(anyString(), eq(ServerGauge.CONSUMPTION_RATE_LIMIT), anyLong());
+ verify(noneMetrics).removeTableGauge("keyC", ServerGauge.CONSUMPTION_RATE_LIMIT);
+ verify(noneMetrics).removeTableGauge("keyC", ServerGauge.CONSUMPTION_QUOTA_UTILIZATION);
+ }
+
+ @Test
+ public void testPartitionRateLimitGaugeLifecycle() {
+ // set -> removed -> set again across consuming segments, against a real metrics registry to verify the gauge is
+ // registered, removed and cleanly re-registered as each new consumer picks up the current config.
+ ServerMetrics metrics = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+ String key = "tableL_REALTIME-topicL-3";
+ String capGaugeName = ServerGauge.CONSUMPTION_RATE_LIMIT.getGaugeName() + "." + key;
+
+ // Limit set (STREAM_CONFIG_D: direct partition limit 4): cap gauge registered with the configured value.
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_D, TABLE_NAME, metrics, key);
+ assertEquals(metrics.getGaugeValue(capGaugeName), Long.valueOf(4));
+
+ // Limit removed (STREAM_CONFIG_C: no limit): the next consumer creation removes the series.
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_C, TABLE_NAME, metrics, key);
+ assertNull(metrics.getGaugeValue(capGaugeName));
+
+ // Limit set again (STREAM_CONFIG_A: topic limit 50 over 10 partitions): gauge re-registers with the new value.
+ CONSUMPTION_RATE_MANAGER.createRateLimiter(STREAM_CONFIG_A, TABLE_NAME, metrics, key);
+ assertEquals(metrics.getGaugeValue(capGaugeName), Long.valueOf(5));
+ }
+
+ @Test
+ public void testServerRateLimitGaugeEmittedOnUpdate() {
+ RealtimeConsumptionRateManager manager = new RealtimeConsumptionRateManager(mock(LoadingCache.class));
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ // Create an active server limiter, then update its cap -> gauge re-emitted with the new value.
+ manager.updateServerRateLimiter(
+ new ServerRateLimitConfig(100_000_000, ByteCountThrottlingStrategy.INSTANCE), metrics);
+ try {
+ manager.updateServerRateLimiter(
+ new ServerRateLimitConfig(200_000_000, ByteCountThrottlingStrategy.INSTANCE), metrics);
+ verify(metrics).setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, 200_000_000L);
+ } finally {
+ ((ServerRateLimiter) manager.getServerRateLimiter()).close();
+ }
+ }
+
private int calcExpectedRatio(double rateLimitInMinutes, int sumOfMsgsInPrevMinute) {
return (int) Math.round(sumOfMsgsInPrevMinute / rateLimitInMinutes * 100);
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ServerRateLimitConfigChangeListenerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ServerRateLimitConfigChangeListenerTest.java
index 4f8327c0e68..c1c99e1ecf8 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ServerRateLimitConfigChangeListenerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ServerRateLimitConfigChangeListenerTest.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.stream.MessageBatch;
@@ -33,6 +34,7 @@ import org.mockito.Mockito;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -113,6 +115,21 @@ public class ServerRateLimitConfigChangeListenerTest {
}
}
+ @Test
+ public void testRateLimitConfigChangeUpdatesGauge() {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ ServerRateLimitConfigChangeListener listener = new ServerRateLimitConfigChangeListener(metrics);
+ String byteKey = CommonConstants.Server.CONFIG_OF_SERVER_CONSUMPTION_RATE_LIMIT_BYTES;
+
+ // Enabling/updating the limit via cluster config re-emits the configured-cap gauge with the new value.
+ listener.onChange(new HashSet<>(List.of(byteKey)), Map.of(byteKey, "150000000"));
+ verify(metrics).setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, 150_000_000L);
+
+ // Disabling via cluster config resets the gauge to -1.
+ listener.onChange(new HashSet<>(List.of(byteKey)), Map.of(byteKey, "0"));
+ verify(metrics).setValueOfGlobalGauge(ServerGauge.SERVER_CONSUMPTION_RATE_LIMIT, -1L);
+ }
+
private void simulateThrottling(AtomicReference<Throwable> errorRef) {
// A helper method to test side effects of throttling during serverRateLimit config change.
MessageBatch messageBatch = Mockito.mock(MessageBatch.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org