From d38ce76566a0be474c85ecb57c0f31465d0be3e8 Mon Sep 17 00:00:00 2001
From: hui lai <1353307710@qq.com>
Date: Sun, 13 Oct 2024 01:59:38 +0800
Subject: [PATCH] [fix](cloud) fix multi table load can not plan after FE
 restart or leader change (#41677)

when master FE node restart, multi table load pause and can not resume:
```
ReasonOfStateChanged: ErrorReason{code=errCode = 2, msg='failed to get stream load plan, errCode = 2, detailMessage = the user is not granted permission to the compute group, ComputeGroupException: CURRENT_USER_NO_AUTH_TO_USE_ANY_COMPUTE_GROUP, you can contact the system administrator and request that they grant you the appropriate compute group permissions, use SQL `GRANT USAGE_PRIV ON COMPUTE GROUP {compute_group_name} TO {user}`'}
```

Due to the loss of cluster name after restart or leader change.
---
 .../doris/cloud/load/CloudRoutineLoadManager.java   | 13 +++++++++++++
 .../doris/load/routineload/RoutineLoadJob.java      |  5 +++++
 2 files changed, 18 insertions(+)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
index fd10a3bc467474..eff1c345e5a0ae 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudRoutineLoadManager.java
@@ -23,6 +23,7 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.load.routineload.RoutineLoadJob;
 import org.apache.doris.load.routineload.RoutineLoadManager;
+import org.apache.doris.persist.RoutineLoadOperation;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.system.Backend;
 
@@ -63,4 +64,16 @@ protected List<Long> getAvailableBackendIds(long jobId) throws LoadException {
                 .map(Backend::getId)
                 .collect(Collectors.toList());
     }
+
+    @Override
+    public void replayCreateRoutineLoadJob(RoutineLoadJob routineLoadJob) {
+        routineLoadJob.setCloudClusterById();
+        super.replayCreateRoutineLoadJob(routineLoadJob);
+    }
+
+    @Override
+    public void replayChangeRoutineLoadJob(RoutineLoadOperation operation) {
+        getJob(operation.getId()).setCloudClusterById();
+        super.replayChangeRoutineLoadJob(operation);
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index b9985e95d08fdb..46c330dca87e0e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -1563,6 +1563,11 @@ public String getCloudClusterId() {
         return cloudClusterId;
     }
 
+    public void setCloudClusterById() {
+        this.cloudCluster = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
+                        .getClusterNameByClusterId(cloudClusterId);
+    }
+
     // check the correctness of commit info
     protected abstract boolean checkCommitInfo(RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment,
                                                TransactionState txnState,