bugfix: thread safety issue when adding and removing instances (apache#6668)

funky-eyes · web-flow · commit e8ea2cce35b2 · 2024-07-13T14:07:42.000+08:00
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
@@ -13,6 +13,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#6626](https://github.com/apache/incubator-seata/pull/6626)] fix hsf ConsumerModel convert error
 - [[#6642](https://github.com/apache/incubator-seata/pull/6642)] codecov token not found
 - [[#6661](https://github.com/apache/incubator-seata/pull/6661)] fix `tableMeta` cache scheduled refresh issue
+- [[#6668](https://github.com/apache/incubator-seata/pull/6668)] thread safety issue when adding and removing instances
 
 ### optimize:
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] split the task thread pool for committing and rollbacking statuses
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
@@ -14,6 +14,8 @@
 - [[#6640](https://github.com/apache/incubator-seata/pull/6640)] 优化codecov相关配置
 - [[#6642](https://github.com/apache/incubator-seata/pull/6642)] 修复codecov token找不到导致无法提交单测覆盖度报告
 - [[#6661](https://github.com/apache/incubator-seata/pull/6661)] 修复`tableMeta`缓存定时刷新失效问题
+- [[#6668](https://github.com/apache/incubator-seata/pull/6668)] 解决namingserver同一个集群下instance添加和删除时的线程安全问题
+
 
 ### optimize:
 - [[#6499](https://github.com/apache/incubator-seata/pull/6499)] 拆分 committing 和 rollbacking 状态的任务线程池
diff --git a/common/src/main/java/org/apache/seata/common/metadata/namingserver/Unit.java b/common/src/main/java/org/apache/seata/common/metadata/namingserver/Unit.java
@@ -49,20 +49,18 @@ public void removeInstance(Node node) {
     }
 
     /**
-     * @param node
-     * @return true if the node has changed, false if there is no change.
+     * @param node node
      */
-    public boolean addInstance(NamingServerNode node) {
+    public void addInstance(NamingServerNode node) {
         if (nodeList.contains(node)) {
             Node node1 = nodeList.get(nodeList.indexOf(node));
             if (node.isTotalEqual(node1)) {
-                return false;
+                return;
             } else {
                 nodeList.remove(node1);
             }
         }
         nodeList.add(node);
-        return true;
 
     }
 
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/controller/HealthController.java b/namingserver/src/main/java/org/apache/seata/namingserver/controller/HealthController.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.namingserver.controller;
+
+import org.apache.seata.common.result.Result;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+public class HealthController {
+
+    @GetMapping("/health")
+    public Result<?> healthCheck() {
+        return new Result<>();
+    }
+
+}
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java b/namingserver/src/main/java/org/apache/seata/namingserver/controller/NamingController.java
@@ -24,8 +24,8 @@
 import org.apache.seata.namingserver.listener.Watcher;
 import org.apache.seata.namingserver.manager.ClusterWatcherManager;
 import org.apache.seata.namingserver.manager.NamingManager;
-import org.apache.seata.namingserver.vo.monitor.ClusterVO;
-import org.apache.seata.namingserver.vo.monitor.WatcherVO;
+import org.apache.seata.namingserver.entity.vo.monitor.ClusterVO;
+import org.apache.seata.namingserver.entity.vo.monitor.WatcherVO;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -56,11 +56,11 @@ public class NamingController {
     private ClusterWatcherManager clusterWatcherManager;
 
     @PostMapping("/register")
-    public Result<?> registerInstance(@RequestParam String namespace,
+    public Result<String> registerInstance(@RequestParam String namespace,
                                       @RequestParam String clusterName,
                                       @RequestParam String unit,
                                       @RequestBody NamingServerNode registerBody) {
-        Result result = new Result();
+        Result<String> result = new Result<>();
         boolean isSuccess = namingManager.registerInstance(registerBody, namespace, clusterName, unit);
         if (isSuccess) {
             result.setMessage("node has registered successfully!");
@@ -72,9 +72,9 @@ public Result<?> registerInstance(@RequestParam String namespace,
     }
 
     @PostMapping("/unregister")
-    public Result<?> unregisterInstance(@RequestParam String unit,
+    public Result<String> unregisterInstance(@RequestParam String unit,
                                         @RequestBody Node registerBody) {
-        Result result = new Result();
+        Result<String> result = new Result<>();
         boolean isSuccess = namingManager.unregisterInstance(unit, registerBody);
         if (isSuccess) {
             result.setMessage("node has unregistered successfully!");
@@ -90,29 +90,24 @@ public List<ClusterVO> monitorCluster(String namespace) {
         return namingManager.monitorCluster(namespace);
     }
 
-    @GetMapping("/health")
-    public Result<?> healthCheck() {
-        return new Result<>();
-    }
-
     @GetMapping("/discovery")
     public MetaResponse discovery(@RequestParam String vGroup, @RequestParam String namespace) {
         return new MetaResponse(namingManager.getClusterListByVgroup(vGroup, namespace),
                 clusterWatcherManager.getTermByvGroup(vGroup));
     }
 
     @PostMapping("/changeGroup")
-    public Result<?> changeGroup(@RequestParam String namespace,
+    public Result<String> changeGroup(@RequestParam String namespace,
                                  @RequestParam String clusterName,
                                  @RequestParam String unitName,
                                  @RequestParam String vGroup) {
 
-        Result<?> addGroupResult = namingManager.addGroup(namespace, vGroup, clusterName, unitName);
+        Result<String> addGroupResult = namingManager.addGroup(namespace, vGroup, clusterName, unitName);
         if (!addGroupResult.isSuccess()) {
             return addGroupResult;
         }
         // remove vGroup in old cluster
-        Result<?> removeGroupResult = namingManager.removeGroup(namespace, vGroup, unitName);
+        Result<String> removeGroupResult = namingManager.removeGroup(namespace, vGroup, unitName);
         if (!removeGroupResult.isSuccess()) {
             return removeGroupResult;
         }
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/entity/pojo/ClusterData.java b/namingserver/src/main/java/org/apache/seata/namingserver/entity/pojo/ClusterData.java
@@ -14,31 +14,35 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.seata.namingserver.pojo;
-
-
-import org.apache.seata.common.metadata.Cluster;
-import org.apache.seata.common.metadata.Node;
-import org.apache.seata.common.metadata.namingserver.NamingServerNode;
-import org.apache.seata.common.metadata.namingserver.Unit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
+package org.apache.seata.namingserver.entity.pojo;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
-@Component
-public class ClusterData extends AbstractClusterData {
+import org.apache.seata.common.metadata.Cluster;
+import org.apache.seata.common.metadata.Node;
+import org.apache.seata.common.metadata.namingserver.NamingServerNode;
+import org.apache.seata.common.metadata.namingserver.Unit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+public class ClusterData {
     private static final Logger LOGGER = LoggerFactory.getLogger(ClusterData.class);
     private String clusterName;
     private String clusterType;
     private final Map<String, Unit> unitData;
+    
+    private Lock lock = new ReentrantLock();
 
 
     public ClusterData() {
@@ -84,8 +88,14 @@ public void removeInstance(Node node, String unitName) {
             return;
         }
         unit.removeInstance(node);
-        if (unit.getNamingInstanceList() == null || unit.getNamingInstanceList().size() == 0) {
-            unitData.remove(unitName);
+        // remove unit if unit has no instance
+        lock.lock();
+        try {
+            if (CollectionUtils.isEmpty(unit.getNamingInstanceList())) {
+                unitData.remove(unitName);
+            }
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -101,7 +111,7 @@ public Cluster getClusterByUnit(String unitName) {
         Cluster clusterResponse = new Cluster();
         clusterResponse.setClusterName(clusterName);
         clusterResponse.setClusterType(clusterType);
-        if (unitName == null || unitName.equals("")) {
+        if (!StringUtils.hasLength(unitName)) {
             clusterResponse.setUnitData(new ArrayList<>(unitData.values()));
         } else {
             List<Unit> unitList = new ArrayList<>();
@@ -112,26 +122,28 @@ public Cluster getClusterByUnit(String unitName) {
         return clusterResponse;
     }
 
-    @Override
     public boolean registerInstance(NamingServerNode instance, String unitName) {
         // refresh node weight
         Object weightValue = instance.getMetadata().get("weight");
         if (weightValue != null) {
             instance.setWeight(Double.parseDouble(String.valueOf(weightValue)));
             instance.getMetadata().remove("weight");
         }
-
-        Unit unit = unitData.get(unitName);
-        if (unit == null) {
-            unit = new Unit();
-            List<Node> instances = new ArrayList<>();
-            instances.add(instance);
+        Unit currentUnit = unitData.computeIfAbsent(unitName, value -> {
+            Unit unit = new Unit();
+            List<Node> instances = new CopyOnWriteArrayList<>();
             unit.setUnitName(unitName);
             unit.setNamingInstanceList(instances);
-            unitData.put(unitName, unit);
-            return true;
+            return unit;
+        });
+        // ensure that when adding an instance, the remove side will not delete the unit.
+        lock.lock();
+        try {
+            currentUnit.addInstance(instance);
+        } finally {
+            lock.unlock();
         }
-        return unit.addInstance(instance);
+        return true;
     }
 
 
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/entity/vo/monitor/ClusterVO.java b/namingserver/src/main/java/org/apache/seata/namingserver/entity/vo/monitor/ClusterVO.java
@@ -14,10 +14,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.seata.namingserver.vo.monitor;
+package org.apache.seata.namingserver.entity.vo.monitor;
 
 import org.apache.seata.common.metadata.namingserver.Unit;
-import org.apache.seata.namingserver.pojo.ClusterData;
+import org.apache.seata.namingserver.entity.pojo.ClusterData;
 
 import java.util.ArrayList;
 import java.util.List;
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/entity/vo/monitor/WatcherVO.java b/namingserver/src/main/java/org/apache/seata/namingserver/entity/vo/monitor/WatcherVO.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.seata.namingserver.vo.monitor;
+package org.apache.seata.namingserver.entity.vo.monitor;
 
 import java.util.List;
 
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/manager/NamingManager.java b/namingserver/src/main/java/org/apache/seata/namingserver/manager/NamingManager.java
diff --git a/namingserver/src/main/java/org/apache/seata/namingserver/pojo/AbstractClusterData.java b/namingserver/src/main/java/org/apache/seata/namingserver/pojo/AbstractClusterData.java
diff --git a/namingserver/src/test/java/org/apache/seata/namingserver/NamingControllerTest.java b/namingserver/src/test/java/org/apache/seata/namingserver/NamingControllerTest.java
diff --git a/server/src/main/java/org/apache/seata/server/controller/HealthController.java b/server/src/main/java/org/apache/seata/server/controller/HealthController.java

Original file line number	Diff line number	Diff line change
`@@ -49,20 +49,18 @@ public void removeInstance(Node node) {`
`49`	`49`	`}`
`50`	`50`
`51`	`51`	`/**`
`52`		`- * @param node`
`53`		`- * @return true if the node has changed, false if there is no change.`
	`52`	`+ * @param node node`
`54`	`53`	`*/`
`55`		`- public boolean addInstance(NamingServerNode node) {`
	`54`	`+ public void addInstance(NamingServerNode node) {`
`56`	`55`	`if (nodeList.contains(node)) {`
`57`	`56`	`Node node1 = nodeList.get(nodeList.indexOf(node));`
`58`	`57`	`if (node.isTotalEqual(node1)) {`
`59`		`- return false;`
	`58`	`+ return;`
`60`	`59`	`} else {`
`61`	`60`	`nodeList.remove(node1);`
`62`	`61`	`}`
`63`	`62`	`}`
`64`	`63`	`nodeList.add(node);`
`65`		`- return true;`
`66`	`64`
`67`	`65`	`}`
`68`	`66`