[branch-2.1] pick some pr from master (#44556)

pick #40468 #40535  #40763 #40855

be/src/util/doris_metrics.cpp

@@ -335,7 +335,7 @@ void DorisMetrics::_update_process_thread_num() {
     std::filesystem::directory_iterator dict_iter("/proc/self/task/", ec);
     if (ec) {
         LOG(WARNING) << "failed to count thread num: " << ec.message();
-        process_fd_num_used->set_value(0);
+        process_thread_num->set_value(0);
         return;
     }
     int64_t count =

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

@@ -35,7 +35,6 @@ import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -76,13 +75,28 @@ public class TopicPublisherThread extends MasterDaemon {
         // because it may means workload group/policy is dropped
 
         // step 2: publish topic info to all be
-        Collection<Backend> nodesToPublish = clusterInfoService.getIdToBackend().values();
+        List<Backend> nodesToPublish = new ArrayList<>();
+        try {
+            for (Backend be : clusterInfoService.getIdToBackend().values()) {
+                if (be.isAlive()) {
+                    nodesToPublish.add(be);
+                }
+            }
+        } catch (Exception e) {
+            LOG.warn("get backends failed", e);
+            return;
+        }
+        if (nodesToPublish.isEmpty()) {
+            LOG.info("no alive backend, skip publish topic");
+            return;
+        }
         AckResponseHandler handler = new AckResponseHandler(nodesToPublish);
         for (Backend be : nodesToPublish) {
             executor.submit(new TopicPublishWorker(request, be, handler));
         }
         try {
             int timeoutMs = Config.publish_topic_info_interval_ms / 3 * 2;
+            timeoutMs = timeoutMs <= 0 ? 3000 : timeoutMs;
             if (!handler.awaitAllInMs(timeoutMs)) {
                 Backend[] backends = handler.pendingNodes();
                 if (backends.length > 0) {

fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java

@@ -446,7 +446,10 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
         // user need to reset user property first
         Pair<Boolean, String> ret = Env.getCurrentEnv().getAuth().isWorkloadGroupInUse(workloadGroupName);
         if (ret.first) {
-            throw new DdlException("workload group " + workloadGroupName + " is set for user " + ret.second);
+            throw new DdlException("workload group " + workloadGroupName + " is set for user " + ret.second
+                    + ", you can reset the user's property(eg, "
+                    + "set property for " + ret.second + " 'default_workload_group'='xxx'; ), "
+                    + "then you can drop the group.");
         }
 
         // A group with related policies should not be deleted.