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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
