gavinchou commented on code in PR #65694:
URL: https://github.com/apache/doris/pull/65694#discussion_r3629171701
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java:
##########
@@ -331,40 +380,26 @@ public void onSuccess(Cloud.GetVersionResponse result) {
@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);
}
}
},
com.google.common.util.concurrent.MoreExecutors.directExecutor());
}
return future;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
Review Comment:
I confirmed that this is still present at the current head (`a7bb3c5`) and
that the impact is broader than `VersionHelper`. The limiter rejects before
sending any RPC, but `getVisibleVersionInternal()` catches every `RpcException`
and returns `null`; the outer loop then retries up to the default 200 times.
With `meta_service_rpc_rate_limit_wait_timeout_ms=0`, one logical operation can
therefore turn into roughly 52–120 seconds of backoff plus 200 local
acquisition attempts and warning logs. Several DDL paths in
`CloudInternalCatalog` also retry every `RpcException`, and query retry logic
treats `RpcException` as retryable. Because `MetaServiceRateLimitException` is
package-private, callers outside `org.apache.doris.cloud.rpc` cannot reliably
distinguish it. Please expose a stable non-retryable signal (for example, a
public subtype or error code), handle it explicitly across the real caller
paths, and add caller-level tests that assert one prompt rejection, no sleep,
and no MetaS
ervice RPC.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java:
##########
@@ -0,0 +1,253 @@
+// 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.limitForPeriod);
Review Comment:
I rechecked the current head. The default overrides are now `500` per core
for both `getPartitionVersion` and `getTableVersion`, so the earlier 4-core/800
example in this thread is no longer current. The correctness issue itself
remains: `acquire()` silently clamps the original batch weight to
`maxPermitsInTimeout`. On a 1-core FE, the default get-version period capacity
is `500 * 1 * 2 = 1000` permits while `cloud_get_version_task_batch_size`
defaults to 2000; larger query batches or a lower configured QPS are
undercharged as well. Resilience4j can calculate a full reservation across
refresh cycles, so please either reserve the complete item count and let the
configured timeout reject it, or split the request into bounded chunks. The
regression test should verify that an above-capacity batch is fully charged or
rejected, rather than capped and accepted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
