Copilot commented on code in PR #65378:
URL: https://github.com/apache/doris/pull/65378#discussion_r3543118986
##########
fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java:
##########
@@ -119,6 +125,97 @@ public void testGetVisibleVersionAsyncShutdownOnFailure()
throws RpcException {
Mockito.verify(client).shutdown(true);
}
+ @Test
+ public void testExecuteRequestNoShutdownOnTooBusyFailure() throws
RpcException {
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+
+ Map<String, MetaServiceClient> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(Config.meta_service_endpoint, client);
+ Queue<Long> lastConnTimeMs = Deencapsulation.getField(proxy,
"lastConnTimeMs");
+ lastConnTimeMs.clear();
+ lastConnTimeMs.add(0L);
+ lastConnTimeMs.add(0L);
+ lastConnTimeMs.add(0L);
+
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ Cloud.MetaServiceResponseStatus status =
Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.MS_TOO_BUSY)
+ .setMsg("server is overloaded")
+ .build();
+ Cloud.GetVersionResponse response =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(status)
+ .build();
+
+ try {
+ wrapper.executeRequest("ignored", (ignored) -> response,
Cloud.GetVersionResponse::getStatus);
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertEquals("server is overloaded", e.getMessage());
+ }
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
Review Comment:
These assertions pin the exception message to the exact server msg. If the
production code is updated to include method/context in the RpcException
message (recommended for debuggability), assert on substring instead of full
equality to keep the test stable.
##########
fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java:
##########
@@ -119,6 +125,97 @@ public void testGetVisibleVersionAsyncShutdownOnFailure()
throws RpcException {
Mockito.verify(client).shutdown(true);
}
+ @Test
+ public void testExecuteRequestNoShutdownOnTooBusyFailure() throws
RpcException {
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+
+ Map<String, MetaServiceClient> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(Config.meta_service_endpoint, client);
+ Queue<Long> lastConnTimeMs = Deencapsulation.getField(proxy,
"lastConnTimeMs");
+ lastConnTimeMs.clear();
+ lastConnTimeMs.add(0L);
+ lastConnTimeMs.add(0L);
+ lastConnTimeMs.add(0L);
+
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ Cloud.MetaServiceResponseStatus status =
Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.MS_TOO_BUSY)
+ .setMsg("server is overloaded")
+ .build();
+ Cloud.GetVersionResponse response =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(status)
+ .build();
+
+ try {
+ wrapper.executeRequest("ignored", (ignored) -> response,
Cloud.GetVersionResponse::getStatus);
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertEquals("server is overloaded", e.getMessage());
+ }
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testExecuteRequestRetryOnTooBusy() throws RpcException {
+ Config.meta_service_rpc_retry_cnt = 2;
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+
+ Map<String, MetaServiceClient> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(Config.meta_service_endpoint, client);
+
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ Cloud.GetVersionResponse tooBusyResponse =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.MS_TOO_BUSY)
+ .setMsg("server is overloaded"))
+ .build();
+ Cloud.GetVersionResponse okResponse =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.OK))
+ .build();
+ AtomicInteger callCount = new AtomicInteger();
+
+ Cloud.GetVersionResponse result = wrapper.executeRequest("ignored",
(ignored) ->
+ callCount.incrementAndGet() == 1 ? tooBusyResponse :
okResponse, Cloud.GetVersionResponse::getStatus);
+
+ Assert.assertEquals(Cloud.MetaServiceCode.OK,
result.getStatus().getCode());
+ Assert.assertEquals(2, callCount.get());
+ Mockito.verify(client, Mockito.never()).shutdown(Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testExecuteRequestFailureAfterTooBusyRetries() throws
RpcException {
+ Config.meta_service_rpc_retry_cnt = 2;
+ MetaServiceProxy proxy = new MetaServiceProxy();
+ MetaServiceClient client = mockNormalClient();
+
+ Map<String, MetaServiceClient> serviceMap =
Deencapsulation.getField(proxy, "serviceMap");
+ serviceMap.put(Config.meta_service_endpoint, client);
+
+ MetaServiceProxy.MetaServiceClientWrapper wrapper =
Deencapsulation.getField(proxy, "w");
+ Cloud.GetVersionResponse tooBusyResponse =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.MS_TOO_BUSY)
+ .setMsg("server is overloaded"))
+ .build();
+ AtomicInteger callCount = new AtomicInteger();
+
+ try {
+ wrapper.executeRequest("ignored", (ignored) -> {
+ callCount.incrementAndGet();
+ return tooBusyResponse;
+ }, Cloud.GetVersionResponse::getStatus);
+ Assert.fail("should throw RpcException");
+ } catch (RpcException e) {
+ Assert.assertEquals("server is overloaded", e.getMessage());
+ }
+
Review Comment:
Same as above: this assertion is brittle if RpcException messages are
augmented with method/context. Prefer asserting that the message contains the
server-provided text.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java:
##########
@@ -212,7 +212,16 @@ public <Response> Response executeRequest(String
methodName, Function<MetaServic
CloudMetrics.META_SERVICE_RPC_ALL_RETRY.increase(1L);
CloudMetrics.META_SERVICE_RPC_RETRY.getOrAdd(methodName).increase(1L);
}
- return function.apply(client);
+ Response response = function.apply(client);
+ Cloud.MetaServiceResponseStatus status =
statusExtractor.apply(response);
+ if (status.getCode() != Cloud.MetaServiceCode.MS_TOO_BUSY)
{
+ return response;
+ }
+ LOG.warn("meta service is too busy, method: {}, msg {},
trycnt {}", methodName, status.getMsg(),
+ tried);
+ if (tried >= maxRetries) {
+ throw new RpcException("", status.getMsg());
+ }
Review Comment:
Optional: when exhausting MS_TOO_BUSY retries, RpcException currently only
contains the server msg. Including the method name (and/or code) in the message
would make caller-side debugging easier when different RPCs can return the same
overload text.
--
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]
