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

Author: mymeiyi

be/src/olap/wal/wal_table.cpp

@@ -172,8 +172,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;
@@ -193,10 +192,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

@@ -408,6 +408,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) {
@@ -425,7 +426,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);
         DBUG_EXECUTE_IF("LoadBlockQueue._finish_group_commit_load.commit_success_and_rpc_error",
                         { result_status = Status::InternalError("commit_success_and_rpc_error"); });
@@ -442,8 +443,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";