[fix](group commit) Modify group commit commit/abort txn timeout as stream load (#39003)

Author: mymeiyi

be/src/olap/wal/wal_table.cpp

@@ -174,8 +174,7 @@ Status WalTable::_try_abort_txn(int64_t db_id, std::string& label) {
             master_addr.hostname, master_addr.port,
             [&request, &result](FrontendServiceConnection& client) {
                 client->loadTxnRollback(result, request);
-            },
-            10000L);
+            });
     auto result_status = Status::create<false>(result.status);
     LOG(INFO) << "abort label " << label << ", st:" << st << ", result_status:" << result_status;
     return result_status;
@@ -195,10 +194,8 @@ Status WalTable::_replay_wal_internal(const std::string& wal) {
         [[maybe_unused]] auto st = _try_abort_txn(_db_id, label);
     }
 #endif
-    DBUG_EXECUTE_IF("WalTable.replay_wals.stop", {
-        // LOG(INFO) << "WalTable.replay_wals.stop";
-        return Status::InternalError("WalTable.replay_wals.stop");
-    });
+    DBUG_EXECUTE_IF("WalTable.replay_wals.stop",
+                    { return Status::InternalError("WalTable.replay_wals.stop"); });
     return _replay_one_wal_with_streamload(wal_id, wal, label);
 }
 

be/src/runtime/group_commit_mgr.cpp

@@ -444,6 +444,7 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
         request.__set_db_id(db_id);
         request.__set_table_id(table_id);
         request.__set_txnId(txn_id);
+        request.__set_thrift_rpc_timeout_ms(config::txn_commit_rpc_timeout_ms);
         request.__set_groupCommit(true);
         request.__set_receiveBytes(state->num_bytes_load_total());
         if (_exec_env->master_info()->__isset.backend_id) {
@@ -463,7 +464,7 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
                     [&request, &result](FrontendServiceConnection& client) {
                         client->loadTxnCommit(result, request);
                     },
-                    10000L);
+                    config::txn_commit_rpc_timeout_ms);
             result_status = Status::create(result.status);
             // DELETE_BITMAP_LOCK_ERROR will be retried
             if (result_status.ok() || !result_status.is<ErrorCode::DELETE_BITMAP_LOCK_ERROR>()) {
@@ -491,8 +492,7 @@ Status GroupCommitTable::_finish_group_commit_load(int64_t db_id, int64_t table_
                 master_addr.hostname, master_addr.port,
                 [&request, &result](FrontendServiceConnection& client) {
                     client->loadTxnRollback(result, request);
-                },
-                10000L);
+                });
         result_status = Status::create<false>(result.status);
         DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.err_status", {
             std ::string msg = "abort txn";