fix replica

Author: wangbo

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java

@@ -212,41 +212,42 @@ private String getCurrentClusterId() throws ComputeGroupException {
         String cluster = null;
         ConnectContext context = ConnectContext.get();
         if (context != null) {
-            if (!Strings.isNullOrEmpty(context.getSessionVariable().getCloudCluster())) {
-                cluster = context.getSessionVariable().getCloudCluster();
+            // TODO(wb) rethinking whether should update err status.
+            cluster = context.getCloudCluster();
+
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("get compute group by context {}", cluster);
+            }
+
+            if (context.getCurrentUserIdentity() != null) {
                 try {
                     ((CloudEnv) Env.getCurrentEnv()).checkCloudClusterPriv(cluster);
                 } catch (Exception e) {
-                    LOG.warn("get compute group by session context exception");
-                    throw new ComputeGroupException(String.format("session context compute group %s check auth failed",
-                            cluster),
-                        ComputeGroupException.FailedTypeEnum.CURRENT_USER_NO_AUTH_TO_USE_DEFAULT_COMPUTE_GROUP);
-                }
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("get compute group by session context compute group: {}", cluster);
-                }
-            } else {
-                cluster = context.getCloudCluster(false);
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("get compute group by context {}", cluster);
-                }
-                String clusterStatus = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
-                        .getCloudStatusByName(cluster);
-                if (!Strings.isNullOrEmpty(clusterStatus)
-                        && Cloud.ClusterStatus.valueOf(clusterStatus)
-                        == Cloud.ClusterStatus.MANUAL_SHUTDOWN) {
-                    LOG.warn("auto start compute group {} in manual shutdown status", cluster);
+                    LOG.warn("check cloud cluster {} for {} auth failed.", cluster,
+                            context.getCurrentUserIdentity().toString());
                     throw new ComputeGroupException(
+                            String.format("context compute group %s check auth failed, user is %s",
+                                    cluster, context.getCurrentUserIdentity().toString()),
+                            ComputeGroupException.FailedTypeEnum.CURRENT_USER_NO_AUTH_TO_USE_DEFAULT_COMPUTE_GROUP);
+                }
+            }
+
+            String clusterStatus = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
+                    .getCloudStatusByName(cluster);
+            if (!Strings.isNullOrEmpty(clusterStatus)
+                    && Cloud.ClusterStatus.valueOf(clusterStatus)
+                    == Cloud.ClusterStatus.MANUAL_SHUTDOWN) {
+                LOG.warn("auto start compute group {} in manual shutdown status", cluster);
+                throw new ComputeGroupException(
                         String.format("The current compute group %s has been manually shutdown", cluster),
                         ComputeGroupException.FailedTypeEnum.CURRENT_COMPUTE_GROUP_BEEN_MANUAL_SHUTDOWN);
-                }
             }
         } else {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("connect context is null in getBackendId");
             }
-            throw new ComputeGroupException("connect context not set",
-                ComputeGroupException.FailedTypeEnum.CONNECT_CONTEXT_NOT_SET);
+            throw new ComputeGroupException("connect context not set cluster ",
+                    ComputeGroupException.FailedTypeEnum.CONNECT_CONTEXT_NOT_SET);
         }
 
         return getCloudClusterIdByName(cluster);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/BackendDistributedPlanWorkerManager.java

@@ -88,14 +88,12 @@ private ImmutableMap<Long, Backend> checkAndInitClusterBackends(
         // 1 get compute group
         ComputeGroup cg = context.getComputeGroup();
 
-        // 2 load job has been checked priv when submitting task, so no need check priv here.
-        if (!isLoadJob) {
-            try {
-                ((CloudEnv) Env.getCurrentEnv()).checkCloudClusterPriv(cg.getName());
-            } catch (Exception e) {
-                LOG.warn("cluster priv check failed", e);
-                throw new UserException("cluster priv check failed", e);
-            }
+        // 2 check priv
+        try {
+            ((CloudEnv) Env.getCurrentEnv()).checkCloudClusterPriv(cg.getName());
+        } catch (Exception e) {
+            LOG.warn("cluster priv check failed", e);
+            throw new UserException("cluster priv check failed", e);
         }
 
         // 3 return be list