[fix](thrift)limit be and fe thrift server max pkg size,avoid accepting error or too large package causing OOM

be/src/common/config.cpp

@@ -1199,6 +1199,10 @@ DEFINE_mInt32(max_s3_client_retry, "10");
 
 DEFINE_String(trino_connector_plugin_dir, "${DORIS_HOME}/connectors");
 
+// the max package bytes be thrift server can receive
+// avoid accepting error or too large package causing OOM,default 20000000(20M)
+DEFINE_Int32(be_thrift_max_pkg_bytes, "20000000");
+
 // clang-format off
 #ifdef BE_TEST
 // test s3

be/src/common/config.h

@@ -1277,6 +1277,10 @@ DECLARE_String(tmp_file_dir);
 // the directory for storing the trino-connector plugins.
 DECLARE_String(trino_connector_plugin_dir);
 
+// the max package bytes be thrift server can receive
+// avoid accepting error or too large package causing OOM,default 20000000(20M)
+DECLARE_Int32(be_thrift_max_pkg_bytes);
+
 #ifdef BE_TEST
 // test s3
 DECLARE_String(test_s3_resource);

be/src/util/thrift_server.cpp

@@ -308,6 +308,11 @@ Status ThriftServer::start() {
     DCHECK(!_started);
     std::shared_ptr<apache::thrift::protocol::TProtocolFactory> protocol_factory(
             new apache::thrift::protocol::TBinaryProtocolFactory());
+    // add binary_protocal_factory to call TBinaryProtocolFactory's member function:setStringSizeLimit
+    std::shared_ptr<apache::thrift::protocol::TBinaryProtocolFactory> binary_protocal_factory =
+            std::dynamic_pointer_cast<apache::thrift::protocol::TBinaryProtocolFactory>(
+                    protocol_factory);
+    binary_protocal_factory->setStringSizeLimit(config::be_thrift_max_pkg_bytes);
     std::shared_ptr<apache::thrift::concurrency::ThreadManager> thread_mgr;
     std::shared_ptr<apache::thrift::concurrency::ThreadFactory> thread_factory =
             std::make_shared<apache::thrift::concurrency::ThreadFactory>();

docs/zh-CN/docs/admin-manual/config/be-config.md

@@ -1535,3 +1535,8 @@ load tablets from header failed, failed tablets size: xxx, path=xxx
 
 * 描述：是否直接使用 Doris 自带的时区数据库。开启后不再尝试查找系统目录。
 * 默认值：false
+
+#### `be_thrift_max_pkg_bytes`
+
+* 描述: be节点thrift端口最大接收包大小(字节)
+* 默认值: 20000000

docs/zh-CN/docs/admin-manual/config/fe-config.md

@@ -2772,3 +2772,13 @@ Doris 为了兼用 mysql 周边工具生态，会内置一个名为 mysql 的数
 默认值：2000
 
 对于自动分区表，防止用户意外创建大量分区，每个OLAP表允许的分区数量为`max_auto_partition_num`。默认2000。
+
+#### `fe_thrift_max_pkg_bytes`
+
+默认值：20000000
+
+是否可以动态配置：false
+
+是否为 Master FE 节点独有的配置项：false
+
+用于限制fe节点thrift端口可以接收的最大包长度，避免接收到过大或者错误的包导致OOM

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2531,6 +2531,12 @@ public class Config extends ConfigBase {
             "Default storage format of inverted index, the default value is V1."
     })
     public static String inverted_index_storage_format = "V1";
+
+    @ConfField(description = {
+            "限制fe节点thrift server可以接收的最大包大小,默认20M,设置为-1表示不限制",
+            "the max package size fe thrift server can receive,avoid accepting error"
+            + "or too large package causing OOM,default 20000000(20M),set -1 for unlimited. "})
+    public static int fe_thrift_max_pkg_bytes = 20000000;
 
     //==========================================================================
     //                    begin of cloud config

fe/fe-core/src/main/java/org/apache/doris/common/ThriftServer.java

@@ -92,14 +92,14 @@ public ThriftServerType getType() {
 
     private void createSimpleServer() throws TTransportException {
         TServer.Args args = new TServer.Args(new TServerSocket(port)).protocolFactory(
-                new TBinaryProtocol.Factory()).processor(processor);
+                new TBinaryProtocol.Factory(Config.fe_thrift_max_pkg_bytes, -1)).processor(processor);
         server = new TSimpleServer(args);
     }
 
     private void createThreadedServer() throws TTransportException {
         TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(
                 new TNonblockingServerSocket(port, Config.thrift_client_timeout_ms)).protocolFactory(
-                        new TBinaryProtocol.Factory()).processor(processor);
+                        new TBinaryProtocol.Factory(Config.fe_thrift_max_pkg_bytes, -1)).processor(processor);
         ThreadPoolExecutor threadPoolExecutor = ThreadPoolManager.newDaemonCacheThreadPool(
                 Config.thrift_server_max_worker_threads, "thrift-server-pool", true);
         args.executorService(threadPoolExecutor);
@@ -123,7 +123,7 @@ private void createThreadPoolServer() throws TTransportException {
 
         TThreadPoolServer.Args serverArgs =
                 new TThreadPoolServer.Args(new TServerSocket(socketTransportArgs)).protocolFactory(
-                        new TBinaryProtocol.Factory()).processor(processor);
+                        new TBinaryProtocol.Factory(Config.fe_thrift_max_pkg_bytes, -1)).processor(processor);
         ThreadPoolExecutor threadPoolExecutor = ThreadPoolManager.newDaemonCacheThreadPool(
                 Config.thrift_server_max_worker_threads, "thrift-server-pool", true);
         serverArgs.executorService(threadPoolExecutor);