This is an automated email from the ASF dual-hosted git repository.
mymeiyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c345e92d93d [feature](fe) Add meta service RPC rate limiting in FE
(#65694)
c345e92d93d is described below
commit c345e92d93d086da11326d449bc1935829a8d072
Author: meiyi <[email protected]>
AuthorDate: Thu Jul 23 15:17:01 2026 +0800
[feature](fe) Add meta service RPC rate limiting in FE (#65694)
Add configurable FE-side rate limiting for meta service RPCs. The change
introduces per-method Resilience4j rate limiters, dynamic config
validation, weighted get-version limiting for batch requests,
rate-limit-specific metrics, and profile/audit visibility for
get-version limiter wait time.
---
.../main/java/org/apache/doris/common/Config.java | 33 +++
.../MetaServiceRpcRateLimitConfigValidator.java | 103 ++++++++
fe/fe-core/pom.xml | 4 +
.../apache/doris/cloud/rpc/MetaServiceProxy.java | 97 ++++---
.../cloud/rpc/MetaServiceRateLimitException.java | 27 ++
.../doris/cloud/rpc/MetaServiceRpcRateLimiter.java | 278 +++++++++++++++++++++
.../doris/common/profile/SummaryProfile.java | 44 +++-
.../java/org/apache/doris/metric/CloudMetrics.java | 17 ++
.../doris/cloud/rpc/MetaServiceProxyTest.java | 190 ++++++++++++++
.../cloud/rpc/MetaServiceRpcRateLimiterTest.java | 275 ++++++++++++++++++++
.../doris/common/profile/SummaryProfileTest.java | 30 +++
fe/pom.xml | 6 +
12 files changed, 1064 insertions(+), 40 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 440e5599735..655af570cfb 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3489,6 +3489,38 @@ public class Config extends ConfigBase {
"In cloud mode, the retry count when the FE request to meta
service times out. Default is 1."})
public static int meta_service_rpc_timeout_retry_times = 1;
+ @ConfField(mutable = true, description = {
+ "Whether to enable QPS rate limit for RPC requests to meta
service."})
+ public static boolean meta_service_rpc_rate_limit_enabled = false;
+
+ @ConfField(mutable = true, description = {
+ "Default QPS limit for each method (requests per second) in each
cpu core, "
+ + "non-positive value (<= 0) means no limit"})
+ public static int meta_service_rpc_rate_limit_default_qps_per_core = 50;
+
+ @ConfField(mutable = true,
+ callback =
MetaServiceRpcRateLimitConfigValidator.QpsConfigHandler.class,
+ description = {
+ "QPS limit config per rpc method to meta service in per cpu
core, "
+ + "format: method1:qps1;method2:qps2, "
+ + "e.g.:
getPartitionVersion:100;getTableVersion:100;getTabletStats:50, "
+ + "non-positive value (<= 0) means no limit"})
+ public static String meta_service_rpc_rate_limit_qps_per_core_config
+ =
"getPartitionVersion:500;getTableVersion:500;getTabletStats:50;beginTxn:50";
+
+ @ConfField(mutable = true,
+ callback =
MetaServiceRpcRateLimitConfigValidator.PositiveIntConfigHandler.class,
+ description = {
+ "Burst window for meta service RPC rate limit in seconds. "
+ + "The long-term average QPS is unchanged, while calls can
burst within this window."})
+ public static int meta_service_rpc_rate_limit_burst_seconds = 2;
+
+ @ConfField(mutable = true, callback =
MetaServiceRpcRateLimitConfigValidator.NonNegativeLongConfigHandler.class,
+ description = {
+ "Max wait time in milliseconds when meta service RPC is rate
limited, "
+ + "zero means fail fast."})
+ public static long meta_service_rpc_rate_limit_wait_timeout_ms = 1000;
+
@ConfField(mutable = true, description = {
"In cloud mode, the auto start and stop ignores the databases used
by internal jobs, "
+ "such as those used for statistics. "
@@ -3735,4 +3767,5 @@ public class Config extends ConfigBase {
+ "（持有主副本的桶），并在单个 tablet 写入量超过阈值（默认 200 MB）后在本地桶之间轮转。"
+ "可降低导入内存压力并提升随机分桶表的吞吐量，覆盖所有导入类型。"})
public static boolean enable_adaptive_random_bucket_load = true;
+
}
diff --git
a/fe/fe-common/src/main/java/org/apache/doris/common/MetaServiceRpcRateLimitConfigValidator.java
b/fe/fe-common/src/main/java/org/apache/doris/common/MetaServiceRpcRateLimitConfigValidator.java
new file mode 100644
index 00000000000..54ac359934c
--- /dev/null
+++
b/fe/fe-common/src/main/java/org/apache/doris/common/MetaServiceRpcRateLimitConfigValidator.java
@@ -0,0 +1,103 @@
+// 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.doris.common;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public final class MetaServiceRpcRateLimitConfigValidator {
+ private MetaServiceRpcRateLimitConfigValidator() {
+ }
+
+ public static Map<String, Integer> parseQpsPerCoreConfig(String config)
throws ConfigException {
+ if (config == null || config.trim().isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, Integer> qpsPerCore = new HashMap<>();
+ Set<String> methods = new HashSet<>();
+ for (String item : config.split(";")) {
+ String trimmedItem = item.trim();
+ if (trimmedItem.isEmpty()) {
+ continue;
+ }
+ int separatorIndex = trimmedItem.indexOf(':');
+ if (separatorIndex <= 0 || separatorIndex !=
trimmedItem.lastIndexOf(':')
+ || separatorIndex == trimmedItem.length() - 1) {
+ throw new ConfigException("Invalid format, expected
method1:qps1;method2:qps2");
+ }
+
+ String methodName = trimmedItem.substring(0,
separatorIndex).trim();
+ String qpsText = trimmedItem.substring(separatorIndex + 1).trim();
+ if (methodName.isEmpty() || qpsText.isEmpty()) {
+ throw new ConfigException("Invalid format, expected
method1:qps1;method2:qps2");
+ }
+ if (!methods.add(methodName)) {
+ throw new ConfigException("Duplicate method: " + methodName);
+ }
+ try {
+ qpsPerCore.put(methodName, Integer.parseInt(qpsText));
+ } catch (NumberFormatException e) {
+ throw new ConfigException("Invalid qps for method: " +
methodName, e);
+ }
+ }
+ return qpsPerCore;
+ }
+
+ public static void validatePositive(String fieldName, int value) throws
ConfigException {
+ if (value <= 0) {
+ throw new ConfigException(fieldName + " must be positive");
+ }
+ }
+
+ public static void validateNonNegative(String fieldName, long value)
throws ConfigException {
+ if (value < 0) {
+ throw new ConfigException(fieldName + " must be non-negative");
+ }
+ }
+
+ public static class QpsConfigHandler extends ConfigBase.DefaultConfHandler
{
+ @Override
+ public void handle(Field field, String confVal) throws Exception {
+ parseQpsPerCoreConfig(confVal);
+ super.handle(field, confVal);
+ }
+ }
+
+ public static class PositiveIntConfigHandler extends
ConfigBase.DefaultConfHandler {
+ @Override
+ public void handle(Field field, String confVal) throws Exception {
+ String trimmedVal = confVal.trim();
+ validatePositive(field.getName(), Integer.parseInt(trimmedVal));
+ super.handle(field, trimmedVal);
+ }
+ }
+
+ public static class NonNegativeLongConfigHandler extends
ConfigBase.DefaultConfHandler {
+ @Override
+ public void handle(Field field, String confVal) throws Exception {
+ String trimmedVal = confVal.trim();
+ validateNonNegative(field.getName(), Long.parseLong(trimmedVal));
+ super.handle(field, trimmedVal);
+ }
+ }
+}
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index ec53a24a75e..d7a7acb8387 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -228,6 +228,10 @@ under the License.
<artifactId>guava-testlib</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-ratelimiter</artifactId>
+ </dependency>
<!--
https://mvnrepository.com/artifact/com.googlecode.java-ipv6/java-ipv6 -->
<dependency>
<groupId>com.googlecode.java-ipv6</groupId>
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
index 25fcfbcd0d7..6b81f084717 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
@@ -19,6 +19,7 @@ package org.apache.doris.cloud.rpc;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.common.Config;
+import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.metric.CloudMetrics;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.rpc.RpcException;
@@ -39,6 +40,7 @@ import java.util.function.Function;
public class MetaServiceProxy {
private static final Logger LOG =
LogManager.getLogger(MetaServiceProxy.class);
+ private static final MetaServiceRpcRateLimiter
META_SERVICE_RPC_RATE_LIMITER = new MetaServiceRpcRateLimiter();
// use exclusive lock to make sure only one thread can add or remove
client from
// serviceMap.
@@ -105,6 +107,7 @@ public class MetaServiceProxy {
}
try {
+ acquireRateLimit(methodName);
final MetaServiceClient client = getProxy();
Cloud.GetInstanceResponse response = client.getInstance(request);
if (MetricRepo.isInit && Config.isCloudMode()) {
@@ -112,17 +115,63 @@ public class MetaServiceProxy {
.update(System.currentTimeMillis() - startTime);
}
return response;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
} catch (Exception e) {
- if (MetricRepo.isInit && Config.isCloudMode()) {
- CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L);
-
CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L);
- CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
- .update(System.currentTimeMillis() - startTime);
- }
+ recordRpcFailed(methodName, startTime);
throw new RpcException("", e.getMessage(), e);
}
}
+ private static long acquireRateLimit(String methodName) throws
RpcException {
+ return META_SERVICE_RPC_RATE_LIMITER.acquire(methodName);
+ }
+
+ private static long acquireRateLimit(String methodName, int permits)
throws RpcException {
+ return META_SERVICE_RPC_RATE_LIMITER.acquire(methodName, permits);
+ }
+
+ private static int getGetVersionRateLimitPermits(Cloud.GetVersionRequest
request) {
+ if (!request.getBatchMode()) {
+ return 1;
+ }
+ int permits = request.hasIsTableVersion() &&
request.getIsTableVersion()
+ ? request.getTableIdsCount()
+ : request.getPartitionIdsCount();
+ return Math.max(permits, 1);
+ }
+
+ static void resetMetaServiceRpcRateLimitForTest() {
+ META_SERVICE_RPC_RATE_LIMITER.reset();
+ }
+
+ private static void recordRpcFailed(String methodName, long startTime) {
+ if (MetricRepo.isInit && Config.isCloudMode()) {
+ CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L);
+
CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L);
+ CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
+ .update(System.currentTimeMillis() - startTime);
+ }
+ }
+
+ private static void recordRpcRateLimited(String methodName) {
+ if (MetricRepo.isInit && Config.isCloudMode()) {
+ CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED.increase(1L);
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMITED.getOrAdd(methodName).increase(1L);
+ }
+ }
+
+ private static void recordGetVersionRateLimitWait(long waitNs) {
+ if (waitNs <= 0) {
+ return;
+ }
+ SummaryProfile profile = SummaryProfile.getSummaryProfile(null);
+ if (profile != null) {
+ profile.addGetMetaVersionRateLimitWaitTime(waitNs);
+ }
+ }
+
public void removeProxy(String address) {
LOG.warn("begin to remove proxy: {}", address);
MetaServiceClient service;
@@ -207,6 +256,7 @@ public class MetaServiceProxy {
MetaServiceClient client = null;
boolean requestFailed = false;
try {
+ acquireRateLimit(methodName, 1);
client = proxy.getProxy();
if (tried > 1 && MetricRepo.isInit &&
Config.isCloudMode()) {
CloudMetrics.META_SERVICE_RPC_ALL_RETRY.increase(1L);
@@ -289,13 +339,11 @@ public class MetaServiceProxy {
.update(System.currentTimeMillis() - startTime);
}
return response;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
} catch (RpcException e) {
- if (MetricRepo.isInit && Config.isCloudMode()) {
- CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L);
-
CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L);
- CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
- .update(System.currentTimeMillis() - startTime);
- }
+ recordRpcFailed(methodName, startTime);
throw e;
}
}
@@ -313,6 +361,7 @@ public class MetaServiceProxy {
}
try {
+ recordGetVersionRateLimitWait(acquireRateLimit(methodName,
getGetVersionRateLimitPermits(request)));
client = getProxy();
Future<Cloud.GetVersionResponse> future =
client.getVisibleVersionAsync(request);
if (future instanceof
com.google.common.util.concurrent.ListenableFuture) {
@@ -331,12 +380,7 @@ public class MetaServiceProxy {
@Override
public void onFailure(Throwable t) {
- if (MetricRepo.isInit && Config.isCloudMode())
{
-
CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L);
-
CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L);
-
CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
- .update(System.currentTimeMillis()
- startTime);
- }
+ recordRpcFailed(methodName, startTime);
if (finalClient != null) {
finalClient.shutdown(true);
}
@@ -344,13 +388,11 @@ public class MetaServiceProxy {
},
com.google.common.util.concurrent.MoreExecutors.directExecutor());
}
return future;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
} catch (Exception e) {
- if (MetricRepo.isInit && Config.isCloudMode()) {
- CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L);
-
CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L);
- CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
- .update(System.currentTimeMillis() - startTime);
- }
+ recordRpcFailed(methodName, startTime);
if (client != null) {
client.shutdown(true);
}
@@ -358,13 +400,6 @@ public class MetaServiceProxy {
}
}
- public Cloud.GetVersionResponse getVersion(Cloud.GetVersionRequest
request) throws RpcException {
- String methodName = request.hasIsTableVersion() &&
request.getIsTableVersion() ? "getTableVersion"
- : "getPartitionVersion";
- return executeWithMetrics(methodName, (client) ->
client.getVersion(request),
- Cloud.GetVersionResponse::getStatus);
- }
-
public Cloud.CreateTabletsResponse
createTablets(Cloud.CreateTabletsRequest request) throws RpcException {
return executeWithMetrics("createTablets", (client) ->
client.createTablets(request),
Cloud.CreateTabletsResponse::getStatus);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRateLimitException.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRateLimitException.java
new file mode 100644
index 00000000000..219a43d467b
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRateLimitException.java
@@ -0,0 +1,27 @@
+// 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.doris.cloud.rpc;
+
+import org.apache.doris.rpc.RpcException;
+
+class MetaServiceRateLimitException extends RpcException {
+ MetaServiceRateLimitException(String methodName, long waitTimeoutMs) {
+ super("", "meta service rpc rate limited, method: " + methodName
+ + ", wait timeout: " + waitTimeoutMs + "ms");
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
new file mode 100644
index 00000000000..cfcd9407641
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
@@ -0,0 +1,278 @@
+// 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.doris.cloud.rpc;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ConfigException;
+import org.apache.doris.common.MetaServiceRpcRateLimitConfigValidator;
+import org.apache.doris.metric.CloudMetrics;
+import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.rpc.RpcException;
+
+import com.google.common.collect.Maps;
+import io.github.resilience4j.ratelimiter.RateLimiter;
+import io.github.resilience4j.ratelimiter.RateLimiterConfig;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+class MetaServiceRpcRateLimiter {
+ private static final Logger LOG =
LogManager.getLogger(MetaServiceRpcRateLimiter.class);
+
+ private static final int CPU_CORES =
Runtime.getRuntime().availableProcessors();
+ private static final RateLimitConfigSnapshot EMPTY_SNAPSHOT =
+ new RateLimitConfigSnapshot(0, "", 1, 0);
+
+ private final ReentrantLock configLock = new ReentrantLock();
+ private final ConcurrentMap<String, RateLimiterHolder> rateLimiters =
Maps.newConcurrentMap();
+ private volatile RateLimitConfigSnapshot currentSnapshot = EMPTY_SNAPSHOT;
+ private volatile Map<String, Integer> methodQpsPerCore =
Collections.emptyMap();
+
+ long acquire(String methodName) throws RpcException {
+ return acquire(methodName, 1);
+ }
+
+ long acquire(String methodName, int permits) throws RpcException {
+ if (!Config.meta_service_rpc_rate_limit_enabled) {
+ return 0;
+ }
+
+ RateLimiterHolder holder = getRateLimiter(methodName);
+ if (holder == null) {
+ return 0;
+ }
+
+ int permitsToAcquire = Math.min(Math.max(permits, 1),
holder.maxPermitsInTimeout);
+ // Resilience4j returns negative when the estimated wait exceeds the
configured timeout.
+ // Otherwise the returned wait time is within
meta_service_rpc_rate_limit_wait_timeout_ms.
+ long nanosToWait =
holder.rateLimiter.reservePermission(permitsToAcquire);
+ if (nanosToWait < 0) {
+ throw new MetaServiceRateLimitException(methodName,
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms);
+ }
+ if (nanosToWait == 0) {
+ return 0;
+ }
+
+ long waitMs = TimeUnit.NANOSECONDS.toMillis(nanosToWait);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("meta service rpc rate limiter waits before acquiring
permission, method: {}, permits: {}, "
+ + "original permits: {}, max permits in timeout:
{}, limit for period: {}, "
+ + "burst seconds: {}, wait ms: {}",
+ methodName, permitsToAcquire, permits,
holder.maxPermitsInTimeout, holder.limitForPeriod,
+ holder.burstSeconds,
+ waitMs);
+ }
+ long waitStartNs = System.nanoTime();
+ try {
+ TimeUnit.NANOSECONDS.sleep(nanosToWait);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RpcException("", e.getMessage(), e);
+ }
+ long actualWaitNs = System.nanoTime() - waitStartNs;
+ if (MetricRepo.isInit && Config.isCloudMode()) {
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY.getOrAdd(methodName)
+ .update(TimeUnit.NANOSECONDS.toMillis(actualWaitNs));
+ }
+ return actualWaitNs;
+ }
+
+ private RateLimiterHolder getRateLimiter(String methodName) throws
RpcException {
+ refreshConfigIfNeeded();
+ RateLimitConfigSnapshot snapshot = currentSnapshot;
+ int qpsPerCore = methodQpsPerCore.getOrDefault(methodName,
snapshot.defaultQpsPerCore);
+ if (qpsPerCore <= 0) {
+ rateLimiters.remove(methodName);
+ return null;
+ }
+
+ int limitForPeriod = getLimitForPeriod(methodName, qpsPerCore,
snapshot.burstSeconds);
+ int maxPermitsInTimeout = getMaxPermitsInTimeout(methodName,
limitForPeriod, snapshot.burstSeconds,
+ snapshot.waitTimeoutMs);
+ RateLimiterHolder holder = rateLimiters.compute(methodName, (name,
existingHolder) -> {
+ if (existingHolder != null &&
existingHolder.matches(limitForPeriod, snapshot.burstSeconds,
+ snapshot.waitTimeoutMs)) {
+ return existingHolder;
+ }
+ return new RateLimiterHolder(createRateLimiter(methodName,
limitForPeriod,
+ snapshot.burstSeconds, snapshot.waitTimeoutMs),
limitForPeriod,
+ maxPermitsInTimeout, snapshot.burstSeconds,
snapshot.waitTimeoutMs);
+ });
+ return holder;
+ }
+
+ private void refreshConfigIfNeeded() throws RpcException {
+ RateLimitConfigSnapshot latestSnapshot =
RateLimitConfigSnapshot.current();
+ if (latestSnapshot.equals(currentSnapshot)) {
+ return;
+ }
+
+ configLock.lock();
+ try {
+ latestSnapshot = RateLimitConfigSnapshot.current();
+ if (latestSnapshot.equals(currentSnapshot)) {
+ return;
+ }
+ validateConfig(latestSnapshot);
+ methodQpsPerCore =
parseMethodQpsPerCore(latestSnapshot.qpsPerCoreConfig);
+ currentSnapshot = latestSnapshot;
+ } finally {
+ configLock.unlock();
+ }
+ }
+
+ private void validateConfig(RateLimitConfigSnapshot snapshot) throws
RpcException {
+ try {
+
MetaServiceRpcRateLimitConfigValidator.validatePositive("meta_service_rpc_rate_limit_burst_seconds",
+ snapshot.burstSeconds);
+
MetaServiceRpcRateLimitConfigValidator.validateNonNegative("meta_service_rpc_rate_limit_wait_timeout_ms",
+ snapshot.waitTimeoutMs);
+ } catch (ConfigException e) {
+ throw new RpcException("", e.getMessage(), e);
+ }
+ }
+
+ private Map<String, Integer> parseMethodQpsPerCore(String config) throws
RpcException {
+ try {
+ return
MetaServiceRpcRateLimitConfigValidator.parseQpsPerCoreConfig(config);
+ } catch (ConfigException e) {
+ throw new RpcException("", "invalid
meta_service_rpc_rate_limit_qps_per_core_config: " + config, e);
+ }
+ }
+
+ private int getLimitForPeriod(String methodName, int qpsPerCore, int
burstSeconds) throws RpcException {
+ long limitForPeriod = (long) qpsPerCore * CPU_CORES * burstSeconds;
+ if (limitForPeriod > Integer.MAX_VALUE) {
+ throw new RpcException("", "meta service rpc rate limit is too
large, method: " + methodName
+ + ", qps per core: " + qpsPerCore + ", cpu cores: " +
CPU_CORES
+ + ", burst seconds: " + burstSeconds);
+ }
+ return (int) limitForPeriod;
+ }
+
+ private int getMaxPermitsInTimeout(String methodName, int limitForPeriod,
int burstSeconds, long waitTimeoutMs)
+ throws RpcException {
+ if (waitTimeoutMs <= 0) {
+ return limitForPeriod;
+ }
+
+ long refreshPeriodMs = TimeUnit.SECONDS.toMillis(burstSeconds);
+ // Keep a small margin from the timeout boundary, so a capped large
batch does not fail repeatedly
+ // because of scheduling jitter or an already slow permit reservation
path.
+ long maxPermitsInTimeout = Math.max(limitForPeriod,
+ (long) Math.floor(limitForPeriod * (double) waitTimeoutMs /
refreshPeriodMs * 0.95D));
+ if (maxPermitsInTimeout > Integer.MAX_VALUE) {
+ throw new RpcException("", "meta service rpc rate limit permits in
timeout is too large, method: "
+ + methodName + ", limit for period: " + limitForPeriod
+ + ", burst seconds: " + burstSeconds + ", wait timeout ms:
" + waitTimeoutMs);
+ }
+ return (int) maxPermitsInTimeout;
+ }
+
+ private RateLimiter createRateLimiter(String methodName, int
limitForPeriod, int burstSeconds,
+ long waitTimeoutMs) {
+ RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
+ .limitRefreshPeriod(Duration.ofSeconds(burstSeconds))
+ .limitForPeriod(limitForPeriod)
+ .timeoutDuration(Duration.ofMillis(waitTimeoutMs))
+ .build();
+ return RateLimiter.of("meta_service_rpc_" + methodName,
rateLimiterConfig);
+ }
+
+ void reset() {
+ configLock.lock();
+ try {
+ rateLimiters.clear();
+ methodQpsPerCore = Collections.emptyMap();
+ currentSnapshot = EMPTY_SNAPSHOT;
+ } finally {
+ configLock.unlock();
+ }
+ }
+
+ private static class RateLimiterHolder {
+ private final RateLimiter rateLimiter;
+ private final int limitForPeriod;
+ private final int maxPermitsInTimeout;
+ private final int burstSeconds;
+ private final long waitTimeoutMs;
+
+ private RateLimiterHolder(RateLimiter rateLimiter, int limitForPeriod,
int maxPermitsInTimeout,
+ int burstSeconds, long waitTimeoutMs) {
+ this.rateLimiter = rateLimiter;
+ this.limitForPeriod = limitForPeriod;
+ this.maxPermitsInTimeout = maxPermitsInTimeout;
+ this.burstSeconds = burstSeconds;
+ this.waitTimeoutMs = waitTimeoutMs;
+ }
+
+ private boolean matches(int limitForPeriod, int burstSeconds, long
waitTimeoutMs) {
+ return this.limitForPeriod == limitForPeriod && this.burstSeconds
== burstSeconds
+ && this.waitTimeoutMs == waitTimeoutMs;
+ }
+ }
+
+ private static class RateLimitConfigSnapshot {
+ private final int defaultQpsPerCore;
+ private final String qpsPerCoreConfig;
+ private final int burstSeconds;
+ private final long waitTimeoutMs;
+
+ private RateLimitConfigSnapshot(int defaultQpsPerCore, String
qpsPerCoreConfig, int burstSeconds,
+ long waitTimeoutMs) {
+ this.defaultQpsPerCore = defaultQpsPerCore;
+ this.qpsPerCoreConfig = qpsPerCoreConfig == null ? "" :
qpsPerCoreConfig;
+ this.burstSeconds = burstSeconds;
+ this.waitTimeoutMs = waitTimeoutMs;
+ }
+
+ private static RateLimitConfigSnapshot current() {
+ return new
RateLimitConfigSnapshot(Config.meta_service_rpc_rate_limit_default_qps_per_core,
+ Config.meta_service_rpc_rate_limit_qps_per_core_config,
+ Config.meta_service_rpc_rate_limit_burst_seconds,
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof RateLimitConfigSnapshot)) {
+ return false;
+ }
+ RateLimitConfigSnapshot that = (RateLimitConfigSnapshot) o;
+ return defaultQpsPerCore == that.defaultQpsPerCore && burstSeconds
== that.burstSeconds
+ && waitTimeoutMs == that.waitTimeoutMs
+ && Objects.equals(qpsPerCoreConfig, that.qpsPerCoreConfig);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(defaultQpsPerCore, qpsPerCoreConfig,
burstSeconds, waitTimeoutMs);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java
b/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java
index 124be52b05b..7011e28df7e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java
@@ -106,6 +106,7 @@ public class SummaryProfile {
public static final String FETCH_RESULT_TIME = "Fetch Result Time";
public static final String WRITE_RESULT_TIME = "Write Result Time";
public static final String GET_META_VERSION_TIME = "Get Meta Version Time";
+ public static final String GET_META_VERSION_RATE_LIMIT_WAIT_TIME = "Get
Meta Version Rate Limit Wait Time";
public static final String GET_PARTITION_VERSION_TIME = "Get Partition
Version Time";
public static final String GET_PARTITION_VERSION_COUNT = "Get Partition
Version Count";
public static final String GET_PARTITION_VERSION_BY_HAS_DATA_COUNT = "Get
Partition Version Count (hasData)";
@@ -218,6 +219,7 @@ public class SummaryProfile {
PAIMON_SCAN_METRICS,
NEREIDS_DISTRIBUTE_TIME,
GET_META_VERSION_TIME,
+ GET_META_VERSION_RATE_LIMIT_WAIT_TIME,
GET_PARTITION_VERSION_TIME,
GET_PARTITION_VERSION_BY_HAS_DATA_COUNT,
GET_PARTITION_VERSION_COUNT,
@@ -275,6 +277,7 @@ public class SummaryProfile {
.put(CREATE_SCAN_RANGE_TIME, 2)
.put(ICEBERG_SCAN_METRICS, 3)
.put(PAIMON_SCAN_METRICS, 3)
+ .put(GET_META_VERSION_RATE_LIMIT_WAIT_TIME, 1)
.put(GET_PARTITION_VERSION_TIME, 1)
.put(GET_PARTITION_VERSION_COUNT, 1)
.put(GET_PARTITION_VERSION_BY_HAS_DATA_COUNT, 1)
@@ -403,6 +406,8 @@ public class SummaryProfile {
private long getTableVersionTime = 0;
@SerializedName(value = "getTableVersionCount")
private long getTableVersionCount = 0;
+ @SerializedName(value = "getMetaVersionRateLimitWaitTime")
+ private long getMetaVersionRateLimitWaitTime = 0;
@SerializedName(value = "transactionCommitBeginTime")
private long transactionCommitBeginTime = -1;
@SerializedName(value = "transactionCommitEndTime")
@@ -678,6 +683,8 @@ public class SummaryProfile {
if (Config.isCloudMode()) {
executionSummaryProfile.addInfoString(GET_META_VERSION_TIME,
getPrettyGetMetaVersionTime());
+
executionSummaryProfile.addInfoString(GET_META_VERSION_RATE_LIMIT_WAIT_TIME,
+ getPrettyGetMetaVersionRateLimitWaitTime());
executionSummaryProfile.addInfoString(GET_PARTITION_VERSION_TIME,
getPrettyGetPartitionVersionTime());
executionSummaryProfile.addInfoString(GET_PARTITION_VERSION_COUNT,
getPrettyGetPartitionVersionCount());
executionSummaryProfile.addInfoString(GET_PARTITION_VERSION_BY_HAS_DATA_COUNT,
@@ -882,6 +889,10 @@ public class SummaryProfile {
this.getTableVersionCount += 1;
}
+ public void addGetMetaVersionRateLimitWaitTime(long ns) {
+ this.getMetaVersionRateLimitWaitTime += ns;
+ }
+
public void incGetPartitionVersionByHasDataCount() {
this.getPartitionVersionByHasDataCount += 1;
}
@@ -1065,6 +1076,13 @@ public class SummaryProfile {
return RuntimeProfile.printCounter(getMetaVersionTime, TUnit.TIME_NS);
}
+ private String getPrettyGetMetaVersionRateLimitWaitTime() {
+ if (getMetaVersionRateLimitWaitTime == 0) {
+ return "N/A";
+ }
+ return RuntimeProfile.printCounter(getMetaVersionRateLimitWaitTime,
TUnit.TIME_NS);
+ }
+
private String getPrettyGetPartitionVersionTime() {
if (getPartitionVersionTime == 0) {
return "N/A";
@@ -1095,8 +1113,8 @@ public class SummaryProfile {
return RuntimeProfile.printCounter(getTableVersionCount, TUnit.UNIT);
}
- public long getGetPartitionVersionTime() {
- return getPartitionVersionTime;
+ public long getGetPartitionVersionTimeMs() {
+ return TimeUnit.NANOSECONDS.toMillis(getPartitionVersionTime);
}
public long getGetPartitionVersionCount() {
@@ -1107,14 +1125,18 @@ public class SummaryProfile {
return getPartitionVersionByHasDataCount;
}
- public long getGetTableVersionTime() {
- return getTableVersionTime;
+ public long getGetTableVersionTimeMs() {
+ return TimeUnit.NANOSECONDS.toMillis(getTableVersionTime);
}
public long getGetTableVersionCount() {
return getTableVersionCount;
}
+ public long getGetMetaVersionRateLimitWaitTime() {
+ return getMetaVersionRateLimitWaitTime;
+ }
+
private String getPrettyTime(long end, long start, TUnit unit) {
if (start == -1 || end == -1) {
return "N/A";
@@ -1383,13 +1405,17 @@ public class SummaryProfile {
}
public String getMetaTime() {
- return "{"
- + "\"get_partition_version_time_ms\"" + ":" +
this.getGetPartitionVersionTime() + ","
+ String metaTime = "{"
+ + "\"get_partition_version_time_ms\"" + ":" +
this.getGetPartitionVersionTimeMs() + ","
+ "\"get_partition_version_count_has_data\"" + ":" +
this.getGetPartitionVersionByHasDataCount() + ","
+ "\"get_partition_version_count\"" + ":" +
this.getGetPartitionVersionCount() + ","
- + "\"get_table_version_time_ms\"" + ":" +
this.getGetTableVersionTime() + ","
- + "\"get_table_version_count\"" + ":" +
this.getGetTableVersionCount()
- + "}";
+ + "\"get_table_version_time_ms\"" + ":" +
this.getGetTableVersionTimeMs() + ","
+ + "\"get_table_version_count\"" + ":" +
this.getGetTableVersionCount();
+ if (this.getGetMetaVersionRateLimitWaitTime() > 0) {
+ metaTime += ",\"get_meta_version_rate_limit_wait_time_ms\"" + ":"
+ +
TimeUnit.NANOSECONDS.toMillis(this.getGetMetaVersionRateLimitWaitTime());
+ }
+ return metaTime + "}";
}
public String getScheduleTime() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java
b/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java
index b88780a9f4c..7649b3b1c67 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/metric/CloudMetrics.java
@@ -58,13 +58,16 @@ public class CloudMetrics {
// Per-method meta-service RPC metrics
public static AutoMappedMetric<LongCounterMetric> META_SERVICE_RPC_TOTAL;
public static AutoMappedMetric<LongCounterMetric> META_SERVICE_RPC_FAILED;
+ public static AutoMappedMetric<LongCounterMetric>
META_SERVICE_RPC_RATE_LIMITED;
public static AutoMappedMetric<LongCounterMetric> META_SERVICE_RPC_RETRY;
public static AutoMappedMetric<GaugeMetricImpl<Double>>
META_SERVICE_RPC_PER_SECOND;
public static AutoMappedMetric<HistogramMetric> META_SERVICE_RPC_LATENCY;
+ public static AutoMappedMetric<HistogramMetric>
META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY;
// Aggregate meta-service metrics
public static LongCounterMetric META_SERVICE_RPC_ALL_TOTAL;
public static LongCounterMetric META_SERVICE_RPC_ALL_FAILED;
+ public static LongCounterMetric META_SERVICE_RPC_ALL_RATE_LIMITED;
public static LongCounterMetric META_SERVICE_RPC_ALL_RETRY;
public static GaugeMetricImpl<Double> META_SERVICE_RPC_ALL_PER_SECOND;
@@ -155,6 +158,9 @@ public class CloudMetrics {
META_SERVICE_RPC_FAILED = MetricRepo.addLabeledMetrics("method", () ->
new LongCounterMetric("meta_service_rpc_failed", MetricUnit.NOUNIT,
"failed meta service RPC calls"));
+ META_SERVICE_RPC_RATE_LIMITED = MetricRepo.addLabeledMetrics("method",
() ->
+ new LongCounterMetric("meta_service_rpc_rate_limited",
MetricUnit.NOUNIT,
+ "rate limited meta service RPC calls"));
META_SERVICE_RPC_RETRY = MetricRepo.addLabeledMetrics("method", () ->
new LongCounterMetric("meta_service_rpc_retry", MetricUnit.NOUNIT,
"meta service RPC retry attempts"));
@@ -170,6 +176,14 @@ public class CloudMetrics {
});
MetricRepo.DORIS_METRIC_REGISTER.addHistogramMetrics(
"meta_service_rpc_latency", META_SERVICE_RPC_LATENCY,
Config::isCloudMode);
+ META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY = new
AutoMappedMetric<>(methodName -> {
+ List<MetricLabel> labels = Collections.singletonList(new
MetricLabel("method", methodName));
+ return new HistogramMetric(MetricRegistry.name("meta_service",
"rpc", "rate_limit_wait",
+ "latency", "ms"), labels);
+ });
+ MetricRepo.DORIS_METRIC_REGISTER.addHistogramMetrics(
+ "meta_service_rpc_rate_limit_wait_latency",
META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY,
+ Config::isCloudMode);
// Aggregate meta-service metrics
META_SERVICE_RPC_ALL_TOTAL = new
LongCounterMetric("meta_service_rpc_all_total",
@@ -178,6 +192,9 @@ public class CloudMetrics {
META_SERVICE_RPC_ALL_FAILED = new
LongCounterMetric("meta_service_rpc_all_failed",
MetricUnit.NOUNIT, "total failed meta service RPC calls");
MetricRepo.DORIS_METRIC_REGISTER.addMetrics(META_SERVICE_RPC_ALL_FAILED);
+ META_SERVICE_RPC_ALL_RATE_LIMITED = new
LongCounterMetric("meta_service_rpc_all_rate_limited",
+ MetricUnit.NOUNIT, "total rate limited meta service RPC calls");
+
MetricRepo.DORIS_METRIC_REGISTER.addMetrics(META_SERVICE_RPC_ALL_RATE_LIMITED);
META_SERVICE_RPC_ALL_RETRY = new
LongCounterMetric("meta_service_rpc_all_retry",
MetricUnit.NOUNIT, "total meta service RPC retry attempts");
MetricRepo.DORIS_METRIC_REGISTER.addMetrics(META_SERVICE_RPC_ALL_RETRY);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
index c37bd8a76cd..7638bfa774d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
@@ -34,19 +34,37 @@ import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
public class MetaServiceProxyTest {
+ private static final int CPU_CORES =
Runtime.getRuntime().availableProcessors();
+
private String originEndpoint;
private long originReconnectIntervalMs;
private long originRetryCnt;
+ private boolean originRateLimitEnabled;
+ private int originRateLimitDefaultQpsPerCore;
+ private String originRateLimitQpsPerCoreConfig;
+ private int originRateLimitBurstSeconds;
+ private long originRateLimitWaitTimeoutMs;
@Before
public void setUp() {
originEndpoint = Config.meta_service_endpoint;
originReconnectIntervalMs =
Config.meta_service_rpc_reconnect_interval_ms;
originRetryCnt = Config.meta_service_rpc_retry_cnt;
+ originRateLimitEnabled = Config.meta_service_rpc_rate_limit_enabled;
+ originRateLimitDefaultQpsPerCore =
Config.meta_service_rpc_rate_limit_default_qps_per_core;
+ originRateLimitQpsPerCoreConfig =
Config.meta_service_rpc_rate_limit_qps_per_core_config;
+ originRateLimitBurstSeconds =
Config.meta_service_rpc_rate_limit_burst_seconds;
+ originRateLimitWaitTimeoutMs =
Config.meta_service_rpc_rate_limit_wait_timeout_ms;
Config.meta_service_endpoint = "127.0.0.1:12345";
Config.meta_service_rpc_reconnect_interval_ms = 0;
Config.meta_service_rpc_retry_cnt = 1;
+ Config.meta_service_rpc_rate_limit_enabled = false;
+ Config.meta_service_rpc_rate_limit_default_qps_per_core = 10;
+ Config.meta_service_rpc_rate_limit_qps_per_core_config = "";
+ Config.meta_service_rpc_rate_limit_burst_seconds = 1;
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms = 0;
+ MetaServiceProxy.resetMetaServiceRpcRateLimitForTest();
}
@After
@@ -54,6 +72,12 @@ public class MetaServiceProxyTest {
Config.meta_service_endpoint = originEndpoint;
Config.meta_service_rpc_reconnect_interval_ms =
originReconnectIntervalMs;
Config.meta_service_rpc_retry_cnt = originRetryCnt;
+ Config.meta_service_rpc_rate_limit_enabled = originRateLimitEnabled;
+ Config.meta_service_rpc_rate_limit_default_qps_per_core =
originRateLimitDefaultQpsPerCore;
+ Config.meta_service_rpc_rate_limit_qps_per_core_config =
originRateLimitQpsPerCoreConfig;
+ Config.meta_service_rpc_rate_limit_burst_seconds =
originRateLimitBurstSeconds;
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms =
originRateLimitWaitTimeoutMs;
+ MetaServiceProxy.resetMetaServiceRpcRateLimitForTest();
}
@Test
@@ -216,6 +240,172 @@ public class MetaServiceProxyTest {
Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
}
+ @Test
+ public void testExecuteRequestRateLimitedWithoutRetryOrShutdown() throws
RpcException {
+ enableRateLimit(1, "", 1, 0);
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+ putClient(proxy, client);
+
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ AtomicInteger callCount = new AtomicInteger();
+ Cloud.GetVersionResponse okResponse = okGetVersionResponse();
+ for (int i = 0; i < CPU_CORES; i++) {
+ wrapper.executeRequest("limited", (ignored) -> {
+ callCount.incrementAndGet();
+ return okResponse;
+ }, Cloud.GetVersionResponse::getStatus);
+ }
+
+ try {
+ wrapper.executeRequest("limited", (ignored) -> {
+ callCount.incrementAndGet();
+ return okResponse;
+ }, Cloud.GetVersionResponse::getStatus);
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains("meta service rpc rate
limited"));
+ }
+
+ Assert.assertEquals(CPU_CORES, callCount.get());
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testRateLimitSharedBetweenProxies() throws RpcException {
+ enableRateLimit(1, "", 1, 0);
+ MetaServiceProxy firstProxy = new MetaServiceProxy();
+ MetaServiceProxy secondProxy = new MetaServiceProxy();
+ putClient(firstProxy, mockNormalClient());
+ putClient(secondProxy, mockNormalClient());
+
+ MetaServiceProxy.MetaServiceClientWrapper firstWrapper =
Deencapsulation.getField(firstProxy, "w");
+ MetaServiceProxy.MetaServiceClientWrapper secondWrapper =
Deencapsulation.getField(secondProxy, "w");
+ Cloud.GetVersionResponse okResponse = okGetVersionResponse();
+ for (int i = 0; i < CPU_CORES; i++) {
+ firstWrapper.executeRequest("shared", (ignored) -> okResponse,
Cloud.GetVersionResponse::getStatus);
+ }
+
+ try {
+ secondWrapper.executeRequest("shared", (ignored) -> okResponse,
Cloud.GetVersionResponse::getStatus);
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains("meta service rpc rate
limited"));
+ }
+ }
+
+ @Test
+ public void testGetInstanceRateLimitedBeforeRpc() throws RpcException {
+ enableRateLimit(1, "", 1, 0);
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+ putClient(proxy, client);
+ consumeRateLimitPermits(proxy, "getInstance");
+
+ try {
+ proxy.getInstance(Cloud.GetInstanceRequest.newBuilder().build());
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains("meta service rpc rate
limited"));
+ }
+
+ Mockito.verify(client, Mockito.never()).getInstance(Mockito.any());
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testGetVisibleVersionAsyncRateLimitedBeforeRpc() throws
RpcException {
+ enableRateLimit(1, "", 1, 0);
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+ putClient(proxy, client);
+ consumeRateLimitPermits(proxy, "getPartitionVersion");
+
+ try {
+
proxy.getVisibleVersionAsync(Cloud.GetVersionRequest.newBuilder().build());
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains("meta service rpc rate
limited"));
+ }
+
+ Mockito.verify(client,
Mockito.never()).getVisibleVersionAsync(Mockito.any());
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
+ }
+
+ @Test
+ public void
testBatchGetVisibleVersionAsyncConsumesMultipleRateLimitPermits() throws
RpcException {
+ enableRateLimit(1, "", 1, 0);
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+ putClient(proxy, client);
+ SettableFuture<Cloud.GetVersionResponse> future =
SettableFuture.create();
+
Mockito.when(client.getVisibleVersionAsync(Mockito.any())).thenReturn(future);
+
+ proxy.getVisibleVersionAsync(buildBatchTableVersionRequest(CPU_CORES));
+
+ try {
+ proxy.getVisibleVersionAsync(Cloud.GetVersionRequest.newBuilder()
+ .setIsTableVersion(true)
+ .build());
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains("meta service rpc rate
limited"));
+ }
+ Mockito.verify(client,
Mockito.times(1)).getVisibleVersionAsync(Mockito.any());
+ }
+
+ private void enableRateLimit(int defaultQpsPerCore, String
qpsPerCoreConfig, int burstSeconds,
+ long waitTimeoutMs) {
+ Config.meta_service_rpc_rate_limit_enabled = true;
+ Config.meta_service_rpc_rate_limit_default_qps_per_core =
defaultQpsPerCore;
+ Config.meta_service_rpc_rate_limit_qps_per_core_config =
qpsPerCoreConfig;
+ Config.meta_service_rpc_rate_limit_burst_seconds = burstSeconds;
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms = waitTimeoutMs;
+ MetaServiceProxy.resetMetaServiceRpcRateLimitForTest();
+ }
+
+ private void putClient(MetaServiceProxy proxy, MetaServiceClient client) {
+ Map<String, MetaServiceClient> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(Config.meta_service_endpoint, client);
+ }
+
+ private void consumeRateLimitPermits(MetaServiceProxy proxy, String
methodName) throws RpcException {
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ Cloud.GetVersionResponse okResponse = okGetVersionResponse();
+ for (int i = 0; i < CPU_CORES; i++) {
+ wrapper.executeRequest(methodName, (ignored) -> okResponse,
Cloud.GetVersionResponse::getStatus);
+ }
+ }
+
+ private Cloud.GetVersionRequest buildBatchPartitionVersionRequest(int
partitionNum) {
+ Cloud.GetVersionRequest.Builder builder =
Cloud.GetVersionRequest.newBuilder()
+ .setBatchMode(true);
+ for (int i = 0; i < partitionNum; i++) {
+ builder.addDbIds(1L);
+ builder.addTableIds(1L);
+ builder.addPartitionIds(i);
+ }
+ return builder.build();
+ }
+
+ private Cloud.GetVersionRequest buildBatchTableVersionRequest(int
tableNum) {
+ Cloud.GetVersionRequest.Builder builder =
Cloud.GetVersionRequest.newBuilder()
+ .setBatchMode(true)
+ .setIsTableVersion(true);
+ for (int i = 0; i < tableNum; i++) {
+ builder.addDbIds(1L);
+ builder.addTableIds(i);
+ }
+ return builder.build();
+ }
+
+ private Cloud.GetVersionResponse okGetVersionResponse() {
+ return Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.OK))
+ .build();
+ }
+
private MetaServiceClient mockNormalClient() {
MetaServiceClient client = Mockito.mock(MetaServiceClient.class);
Mockito.when(client.isNormalState()).thenReturn(true);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
new file mode 100644
index 00000000000..e54db40f595
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
@@ -0,0 +1,275 @@
+// 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.doris.cloud.rpc;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ConfigBase;
+import org.apache.doris.common.MetaServiceRpcRateLimitConfigValidator;
+import org.apache.doris.rpc.RpcException;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.TimeUnit;
+
+public class MetaServiceRpcRateLimiterTest {
+ private static final int CPU_CORES =
Runtime.getRuntime().availableProcessors();
+
+ private boolean originRateLimitEnabled;
+ private int originRateLimitDefaultQpsPerCore;
+ private String originRateLimitQpsPerCoreConfig;
+ private int originRateLimitBurstSeconds;
+ private long originRateLimitWaitTimeoutMs;
+
+ private MetaServiceRpcRateLimiter rateLimiter;
+
+ @Before
+ public void setUp() {
+ originRateLimitEnabled = Config.meta_service_rpc_rate_limit_enabled;
+ originRateLimitDefaultQpsPerCore =
Config.meta_service_rpc_rate_limit_default_qps_per_core;
+ originRateLimitQpsPerCoreConfig =
Config.meta_service_rpc_rate_limit_qps_per_core_config;
+ originRateLimitBurstSeconds =
Config.meta_service_rpc_rate_limit_burst_seconds;
+ originRateLimitWaitTimeoutMs =
Config.meta_service_rpc_rate_limit_wait_timeout_ms;
+
+ rateLimiter = new MetaServiceRpcRateLimiter();
+ enableRateLimit(1, "", 1, 0);
+ }
+
+ @After
+ public void tearDown() {
+ Config.meta_service_rpc_rate_limit_enabled = originRateLimitEnabled;
+ Config.meta_service_rpc_rate_limit_default_qps_per_core =
originRateLimitDefaultQpsPerCore;
+ Config.meta_service_rpc_rate_limit_qps_per_core_config =
originRateLimitQpsPerCoreConfig;
+ Config.meta_service_rpc_rate_limit_burst_seconds =
originRateLimitBurstSeconds;
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms =
originRateLimitWaitTimeoutMs;
+ rateLimiter.reset();
+ }
+
+ @Test
+ public void testMethodRateLimitCanBeDisabledByConfig() throws RpcException
{
+ enableRateLimit(1, "unlimited:0", 1, 0);
+
+ for (int i = 0; i <= CPU_CORES; i++) {
+ rateLimiter.acquire("unlimited");
+ }
+ }
+
+ @Test
+ public void testRateLimitDisabledBySwitch() throws RpcException {
+ consumePermits("disabledSwitch", CPU_CORES);
+ assertRateLimited("disabledSwitch");
+
+ Config.meta_service_rpc_rate_limit_enabled = false;
+ rateLimiter.acquire("disabledSwitch");
+ }
+
+ @Test
+ public void testMethodOverrideQpsTakesEffect() throws RpcException {
+ enableRateLimit(1, "fast:2", 1, 0);
+
+ consumePermits("normal", CPU_CORES);
+ assertRateLimited("normal");
+
+ consumePermits("fast", CPU_CORES * 2);
+ assertRateLimited("fast");
+ }
+
+ @Test
+ public void testRateLimitIsIsolatedBetweenMethods() throws RpcException {
+ consumePermits("firstMethod", CPU_CORES);
+ assertRateLimited("firstMethod");
+
+ rateLimiter.acquire("secondMethod");
+ }
+
+ @Test
+ public void testWeightedAcquireConsumesMultiplePermits() throws
RpcException {
+ rateLimiter.acquire("weighted", CPU_CORES);
+
+ assertRateLimited("weighted");
+ }
+
+ @Test
+ public void
testWeightedAcquireLargerThanTimeoutCapacityIsCappedAndSucceeds() throws
RpcException {
+ rateLimiter.acquire("largeWeighted", CPU_CORES + 1);
+
+ assertRateLimited("largeWeighted");
+ }
+
+ @Test
+ public void testWeightedAcquireUsesTimeoutCapacity() throws RpcException {
+ enableRateLimit(1, "", 1, 3000);
+
+ long waitNs = rateLimiter.acquire("timeoutCapacity", CPU_CORES * 2);
+
+ Assert.assertTrue(waitNs > 0);
+ }
+
+ @Test
+ public void testDynamicConfigUpdate() throws RpcException {
+ consumePermits("dynamic", CPU_CORES);
+ assertRateLimited("dynamic");
+
+ Config.meta_service_rpc_rate_limit_default_qps_per_core = 0;
+ rateLimiter.acquire("dynamic");
+ }
+
+ @Test
+ public void testBurstWindow() throws RpcException {
+ enableRateLimit(1, "burst:1", 2, 0);
+
+ consumePermits("burst", CPU_CORES * 2);
+ assertRateLimited("burst");
+ }
+
+ @Test
+ public void testWaitTimeoutRejectsWithoutWaitingForNextRefreshPeriod()
throws RpcException {
+ enableRateLimit(1, "", 3, 1);
+ consumePermits("waitTimeout", CPU_CORES * 3);
+
+ long startTimeMs = System.currentTimeMillis();
+ assertRateLimited("waitTimeout");
+ Assert.assertTrue(System.currentTimeMillis() - startTimeMs <
TimeUnit.SECONDS.toMillis(1));
+ }
+
+ @Test
+ public void testAcquireReturnsActualWaitTime() throws RpcException {
+ enableRateLimit(1, "", 1, 2000);
+ consumePermits("wait", CPU_CORES);
+
+ long waitNs = rateLimiter.acquire("wait");
+
+ Assert.assertTrue(waitNs > 0);
+ }
+
+ @Test
+ public void testInvalidQpsConfig() throws Exception {
+ Field field =
Config.class.getField("meta_service_rpc_rate_limit_qps_per_core_config");
+ MetaServiceRpcRateLimitConfigValidator.QpsConfigHandler handler =
+ new MetaServiceRpcRateLimitConfigValidator.QpsConfigHandler();
+
+ assertConfigHandlerRejects(handler, field, "invalid", "Invalid");
+ assertConfigHandlerRejects(handler, field, "method:", "Invalid");
+ assertConfigHandlerRejects(handler, field, ":1", "Invalid");
+ assertConfigHandlerRejects(handler, field, "method:abc", "Invalid");
+ assertConfigHandlerRejects(handler, field, "method:1;method:2",
"Duplicate");
+
+ Config.meta_service_rpc_rate_limit_qps_per_core_config = "invalid";
+ assertRpcException("invalid
meta_service_rpc_rate_limit_qps_per_core_config",
+ () -> rateLimiter.acquire("invalid"));
+
+ Config.meta_service_rpc_rate_limit_qps_per_core_config =
"method:1;method:2";
+ assertRpcException("invalid
meta_service_rpc_rate_limit_qps_per_core_config",
+ () -> rateLimiter.acquire("method"));
+ }
+
+ @Test
+ public void testInvalidBurstSecondsConfigHandler() throws Exception {
+ Field field =
Config.class.getField("meta_service_rpc_rate_limit_burst_seconds");
+ MetaServiceRpcRateLimitConfigValidator.PositiveIntConfigHandler
handler =
+ new
MetaServiceRpcRateLimitConfigValidator.PositiveIntConfigHandler();
+
+ assertConfigHandlerRejects(handler, field, "0", "must be positive");
+ assertConfigHandlerRejects(handler, field, "-1", "must be positive");
+
+ handler.handle(field, " 2 ");
+ Assert.assertEquals(2,
Config.meta_service_rpc_rate_limit_burst_seconds);
+ }
+
+ @Test
+ public void testInvalidBurstSeconds() throws RpcException {
+ enableRateLimit(1, "", 0, 0);
+
+ assertRpcException("meta_service_rpc_rate_limit_burst_seconds must be
positive",
+ () -> rateLimiter.acquire("invalidBurst"));
+ }
+
+ @Test
+ public void testInvalidWaitTimeoutMsConfigHandler() throws Exception {
+ Field field =
Config.class.getField("meta_service_rpc_rate_limit_wait_timeout_ms");
+ MetaServiceRpcRateLimitConfigValidator.NonNegativeLongConfigHandler
handler =
+ new
MetaServiceRpcRateLimitConfigValidator.NonNegativeLongConfigHandler();
+
+ assertConfigHandlerRejects(handler, field, "-1", "must be
non-negative");
+
+ handler.handle(field, " 0 ");
+ Assert.assertEquals(0,
Config.meta_service_rpc_rate_limit_wait_timeout_ms);
+ }
+
+ @Test
+ public void testInvalidWaitTimeoutMs() throws RpcException {
+ enableRateLimit(1, "", 1, -1);
+
+ assertRpcException("meta_service_rpc_rate_limit_wait_timeout_ms must
be non-negative",
+ () -> rateLimiter.acquire("invalidWaitTimeout"));
+ }
+
+ @Test
+ public void testLimitForPeriodOverflow() throws RpcException {
+ enableRateLimit(Integer.MAX_VALUE, "", 2, 0);
+
+ assertRpcException("meta service rpc rate limit is too large",
+ () -> rateLimiter.acquire("overflow"));
+ }
+
+ private void enableRateLimit(int defaultQpsPerCore, String
qpsPerCoreConfig, int burstSeconds,
+ long waitTimeoutMs) {
+ Config.meta_service_rpc_rate_limit_enabled = true;
+ Config.meta_service_rpc_rate_limit_default_qps_per_core =
defaultQpsPerCore;
+ Config.meta_service_rpc_rate_limit_qps_per_core_config =
qpsPerCoreConfig;
+ Config.meta_service_rpc_rate_limit_burst_seconds = burstSeconds;
+ Config.meta_service_rpc_rate_limit_wait_timeout_ms = waitTimeoutMs;
+ rateLimiter.reset();
+ }
+
+ private void consumePermits(String methodName, int permitNum) throws
RpcException {
+ for (int i = 0; i < permitNum; i++) {
+ rateLimiter.acquire(methodName);
+ }
+ }
+
+ private void assertConfigHandlerRejects(ConfigBase.ConfHandler handler,
Field field, String config,
+ String expectedMessage) throws Exception {
+ try {
+ handler.handle(field, config);
+ Assert.fail("should throw exception");
+ } catch (Exception e) {
+ Assert.assertTrue(e.getMessage().contains(expectedMessage));
+ }
+ }
+
+ private void assertRateLimited(String methodName) throws RpcException {
+ assertRpcException("meta service rpc rate limited", () ->
rateLimiter.acquire(methodName));
+ }
+
+ private void assertRpcException(String expectedMessage, RpcCall rpcCall)
throws RpcException {
+ try {
+ rpcCall.run();
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertTrue(e.getMessage().contains(expectedMessage));
+ }
+ }
+
+ private interface RpcCall {
+ void run() throws RpcException;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java
b/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java
index afe40a68dc8..f6e8d5a677d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/common/profile/SummaryProfileTest.java
@@ -17,6 +17,8 @@
package org.apache.doris.common.profile;
+import org.apache.doris.common.Config;
+
import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -78,6 +80,34 @@ public class SummaryProfileTest {
Assertions.assertEquals("20ms",
profile.getPrettyNereidsPreloadExternalMetadataTime());
}
+ @Test
+ public void testMetaVersionRateLimitWaitTime() {
+ String originalCloudUniqueId = Config.cloud_unique_id;
+ Config.cloud_unique_id = "test_cloud";
+ try {
+ SummaryProfile profile = new SummaryProfile();
+ profile.addGetPartitionVersionTime(1_000_000);
+ profile.addGetTableVersionTime(2_000_000);
+ profile.addGetMetaVersionRateLimitWaitTime(1_000_000);
+ profile.addGetMetaVersionRateLimitWaitTime(2_000_000);
+
+ profile.update(ImmutableMap.of());
+
+ RuntimeProfile executionSummary = profile.getExecutionSummary();
+ Assertions.assertEquals(3_000_000,
profile.getGetMetaVersionRateLimitWaitTime());
+ Assertions.assertEquals("3.0ms", executionSummary.getInfoString(
+ SummaryProfile.GET_META_VERSION_RATE_LIMIT_WAIT_TIME));
+ String metaTime = profile.getMetaTime();
+
Assertions.assertTrue(metaTime.contains("\"get_partition_version_time_ms\":1"));
+
Assertions.assertTrue(metaTime.contains("\"get_table_version_time_ms\":2"));
+
Assertions.assertTrue(metaTime.contains("\"get_meta_version_rate_limit_wait_time_ms\":3"));
+ Assertions.assertFalse(new SummaryProfile().getMetaTime().contains(
+ "get_meta_version_rate_limit_wait_time_ms"));
+ } finally {
+ Config.cloud_unique_id = originalCloudUniqueId;
+ }
+ }
+
@Test
public void testExternalTableMetaSummary() {
SummaryProfile profile = new SummaryProfile();
diff --git a/fe/pom.xml b/fe/pom.xml
index 2b44718723b..fb9b8088f69 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -291,6 +291,7 @@ under the License.
<mqtt.version>1.2.5</mqtt.version>
<slf4j.version>2.0.17</slf4j.version>
<metrics-core.version>4.0.2</metrics-core.version>
+ <resilience4j.version>2.4.0</resilience4j.version>
<!-- Keep Netty compatible with Arrow Flight SQL 19 and other
transitive Netty users. -->
<netty-all.version>4.2.15.Final</netty-all.version>
<!--The dependence of transitive dependence cannot be ruled out, only
Saving the nation through twisted ways.-->
@@ -918,6 +919,11 @@ under the License.
<artifactId>cel</artifactId>
<version>${cel.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-ratelimiter</artifactId>
+ <version>${resilience4j.version}</version>
+ </dependency>
<!-- https://mvnrepository.com/artifact/org.javassist/javassist -->
<dependency>
<groupId>org.javassist</groupId>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
