bugfix: seata node refresh failure because consul crash (#3481)

changes/1.5.0.md

@@ -37,6 +37,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
   - [[#3431](https://github.com/seata/seata/pull/3431)] 修复在读取配置时Property Bean可能还未初始化
   - [[#3413](https://github.com/seata/seata/pull/3413)] 修复回滚到savepoint以及releaseSavepoint的逻辑
   - [[#3451](https://github.com/seata/seata/pull/3451)] 修复当不使用本地事务且设置自动提交为true时，全局锁竞争失败会使得rm退出全局事务，导致全局锁在rm重试时失效，数据被脏写
+  - [[#3481](https://github.com/seata/seata/pull/3481)] 修复当 consul client 获取集群信息报错时会导致刷新任务中断
 
 
   ### optimize： 
@@ -87,6 +88,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
   - [MentosL](https://github.com/MentosL)
   - [lian88jian](https://github.com/lian88jian)
   - [litianyu1992](https://github.com/litianyu1992)
+  - [xyz327](https://github.com/xyz327)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。
 

changes/en-us/1.5.0.md

@@ -38,6 +38,8 @@
   - [[#3413](https://github.com/seata/seata/pull/3413)] fix the logic of rollback to savepoint and release to savepoint	
   - [[#3443](https://github.com/seata/seata/pull/3443)] send the `seata-server` log to `logstash` or `kafka`	
   - [[#3451](https://github.com/seata/seata/pull/3451)] fix set auto-commit to true when local transactions are not being used. Failure to compete for a lock causes the global transaction to exit, invaliding the global row lock and dirty writing of the data.
+  - [[#3481](https://github.com/seata/seata/pull/3481)] fix seata node refresh failure because consul crash
+    
   ### optimize： 	
 
   - [[#3383](https://github.com/seata/seata/pull/3383)] optimize StatementProxyTest unit test 	
@@ -86,6 +88,7 @@
   - [MentosL](https://github.com/MentosL)	
   - [lian88jian](https://github.com/lian88jian)	
   - [litianyu1992](https://github.com/litianyu1992)
+  - [xyz327](https://github.com/xyz327)
 
   Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.	
 

discovery/seata-discovery-consul/src/main/java/io/seata/discovery/registry/consul/ConsulRegistryServiceImpl.java

@@ -26,6 +26,8 @@ import io.seata.common.util.NetUtil;
 import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
 import io.seata.discovery.registry.RegistryService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
 import java.util.Collections;
@@ -48,6 +50,7 @@ public class ConsulRegistryServiceImpl implements RegistryService<ConsulListener
     private static volatile ConsulRegistryServiceImpl instance;
     private static volatile ConsulClient client;
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ConsulRegistryServiceImpl.class);
     private static final Configuration FILE_CONFIG = ConfigurationFactory.CURRENT_FILE_INSTANCE;
     private static final String FILE_ROOT_REGISTRY = "registry";
     private static final String FILE_CONFIG_SPLIT_CHAR = ".";
@@ -282,6 +285,7 @@ public class ConsulRegistryServiceImpl implements RegistryService<ConsulListener
         private String cluster;
         private long consulIndex;
         private boolean running;
+        private boolean hasError = false;
 
         ConsulNotifier(String cluster, long consulIndex) {
             this.cluster = cluster;
@@ -292,14 +296,21 @@ public class ConsulRegistryServiceImpl implements RegistryService<ConsulListener
         @Override
         public void run() {
             while (this.running) {
-                processService();
+                try {
+                    processService();
+                } catch (Exception exception) {
+                    hasError = true;
+                    LOGGER.error("consul refresh services error:{}", exception.getMessage());
+                }
             }
         }
 
         private void processService() {
             Response<List<HealthService>> response = getHealthyServices(cluster, consulIndex, DEFAULT_WATCH_TIMEOUT);
             Long currentIndex = response.getConsulIndex();
-            if (currentIndex != null && currentIndex > consulIndex) {
+
+            if ((currentIndex != null && currentIndex > consulIndex) || hasError) {
+                hasError = false;
                 List<HealthService> services = response.getValue();
                 consulIndex = currentIndex;
                 for (ConsulListener listener : listenerMap.get(cluster)) {