From 2ced17f5041d419e8e3b051388a3756962515fb5 Mon Sep 17 00:00:00 2001
From: Pxl <pxl290@qq.com>
Date: Thu, 18 Jul 2024 15:59:34 +0800
Subject: [PATCH] =?UTF-8?q?[Improvement](rpc)=20set=20grpc=20channel's=20k?=
 =?UTF-8?q?eepAliveTime=20and=20remove=20proxy=20on=20InterruptedExcep?=
 =?UTF-8?q?=E2=80=A6=20(#37304)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## Proposed changes
1. set grpc channel's keepAliveTime
2. remove proxy on InterruptedException/TimeoutException to avoid
channel unavailable
---
 .../src/main/java/org/apache/doris/common/Config.java     | 8 ++++++++
 .../src/main/java/org/apache/doris/qe/Coordinator.java    | 2 ++
 .../java/org/apache/doris/rpc/BackendServiceClient.java   | 2 +-
 3 files changed, 11 insertions(+), 1 deletion(-)

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 5a811fddd6c985..e32359761dbbf4 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
@@ -1464,6 +1464,14 @@ public class Config extends ConfigBase {
     @ConfField
     public static int grpc_threadmgr_threads_nums = 4096;
 
+    /**
+     * sets the time without read activity before sending a keepalive ping
+     * the smaller the value, the sooner the channel is unavailable, but it will increase network io
+     */
+    @ConfField(description = { "设置grpc连接发送 keepalive ping 之前没有数据传输的时间。",
+            "The time without grpc read activity before sending a keepalive ping" })
+    public static int grpc_keep_alive_second = 10;
+
     /**
      * Used to set minimal number of replication per tablet.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 2904cfdb5e47f2..f85a5b38cbf7a8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -1199,6 +1199,7 @@ private void waitPipelineRpc(List<Triple<PipelineExecContexts, BackendServicePro
             } catch (InterruptedException e) {
                 exception = e;
                 code = TStatusCode.INTERNAL_ERROR;
+                triple.getMiddle().removeProxy(triple.getLeft().brpcAddr);
             } catch (TimeoutException e) {
                 exception = e;
                 errMsg = String.format(
@@ -1206,6 +1207,7 @@ private void waitPipelineRpc(List<Triple<PipelineExecContexts, BackendServicePro
                                             operation, queryOptions.getExecutionTimeout(), timeoutMs / 1000);
                 LOG.warn("Query {} {}", DebugUtil.printId(queryId), errMsg);
                 code = TStatusCode.TIMEOUT;
+                triple.getMiddle().removeProxy(triple.getLeft().brpcAddr);
             }
 
             if (code != TStatusCode.OK) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
index 3b7780591d6edb..924955e662c443 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java
@@ -46,7 +46,7 @@ public class BackendServiceClient {
     public BackendServiceClient(TNetworkAddress address, Executor executor) {
         this.address = address;
         channel = NettyChannelBuilder.forAddress(address.getHostname(), address.getPort())
-                .executor(executor)
+                .executor(executor).keepAliveTime(Config.grpc_keep_alive_second, TimeUnit.SECONDS)
                 .flowControlWindow(Config.grpc_max_message_size_bytes)
                 .keepAliveWithoutCalls(true)
                 .maxInboundMessageSize(Config.grpc_max_message_size_bytes).enableRetry().maxRetryAttempts(MAX_RETRY_NUM)