optimize: RM TM startup connect server fail fast

changes/en-us/2.0.0.md

@@ -153,10 +153,10 @@ The version is updated as follows:
 - [[#5938](https://github.com/seata/seata/pull/5938)] support jmx port in seata
 - [[#5951](https://github.com/seata/seata/pull/5951)] remove un support config in jdk17
 - [[#5959](https://github.com/seata/seata/pull/5959)] modify code style and remove unused import
+- [[#6004](https://github.com/seata/seata/pull/6004)] optimize RM TM startup connect server fail fast
 - [[#6002](https://github.com/seata/seata/pull/6002)] remove fst serialization
 - [[#6045](https://github.com/seata/seata/pull/6045)] optimize derivative product check base on mysql
 
-
 ### security:
 - [[#5642](https://github.com/seata/seata/pull/5642)] add Hessian Serializer WhiteDenyList
 - [[#5694](https://github.com/seata/seata/pull/5694)] fix several node.js security vulnerabilities

changes/zh-cn/2.0.0.md

@@ -154,6 +154,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
 - [[#5938](https://github.com/seata/seata/pull/5938)] 支持 jmx 监控配置
 - [[#5951](https://github.com/seata/seata/pull/5951)] 删除在 jdk17 中不支持的配置项
 - [[#5959](https://github.com/seata/seata/pull/5959)] 修正代码风格问题及去除无用的类引用
+- [[#6004](https://github.com/seata/seata/pull/6004)] 优化RM,TM连接server快速失败
 - [[#6002](https://github.com/seata/seata/pull/6002)] 移除fst序列化模块
 - [[#6045](https://github.com/seata/seata/pull/6045)] 优化MySQL衍生数据库判断逻辑
 

common/src/main/java/io/seata/common/ConfigurationKeys.java

@@ -638,6 +638,16 @@ public interface ConfigurationKeys {
      */
     String ENABLE_TM_CLIENT_BATCH_SEND_REQUEST = TRANSPORT_PREFIX + "enableTmClientBatchSendRequest";
 
+    /**
+     * The constant ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST
+     */
+    String ENABLE_TM_CLIENT_CHANNEL_CHECK_FAIL_FAST = TRANSPORT_PREFIX + "enableTmClientChannelCheckFailFast";
+
+    /**
+     * The constant ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST
+     */
+    String ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST = TRANSPORT_PREFIX + "enableRmClientChannelCheckFailFast";
+
     /**
      * The constant ENABLE_RM_CLIENT_BATCH_SEND_REQUEST
      */

common/src/main/java/io/seata/common/DefaultValues.java

@@ -60,6 +60,8 @@ public interface DefaultValues {
     boolean DEFAULT_ENABLE_RM_CLIENT_BATCH_SEND_REQUEST = true;
     boolean DEFAULT_ENABLE_TC_SERVER_BATCH_SEND_RESPONSE = false;
 
+    boolean DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST = true;
+
     String DEFAULT_BOSS_THREAD_PREFIX = "NettyBoss";
     String DEFAULT_NIO_WORKER_THREAD_PREFIX = "NettyServerNIOWorker";
     String DEFAULT_EXECUTOR_THREAD_PREFIX = "NettyServerBizHandler";

common/src/main/java/io/seata/common/util/StringUtils.java

@@ -25,6 +25,7 @@
 import java.text.SimpleDateFormat;
 import java.util.Collection;
 import java.util.Date;
+import java.util.Iterator;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -399,4 +400,31 @@ public static boolean hasUpperCase(String str) {
         return false;
     }
 
+    public static String join(Iterator iterator, String separator) {
+        if (iterator == null) {
+            return null;
+        }
+        if (!iterator.hasNext()) {
+            return EMPTY;
+        }
+        Object first = iterator.next();
+        if (!iterator.hasNext()) {
+            return first == null ? "" : first.toString();
+        }
+        StringBuilder builder = new StringBuilder(256);
+        if (first != null) {
+            builder.append(first);
+        }
+        while (iterator.hasNext()) {
+            if (separator != null) {
+                builder.append(separator);
+            }
+            Object obj = iterator.next();
+            if (obj != null) {
+                builder.append(obj);
+            }
+        }
+        return builder.toString();
+    }
+
 }

core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java

@@ -58,6 +58,7 @@
 public abstract class AbstractNettyRemoting implements Disposable {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyRemoting.class);
+
     /**
      * The Timer executor.
      */

core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -104,7 +104,13 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting
 
     @Override
     public void init() {
-        timerExecutor.scheduleAtFixedRate(() -> clientChannelManager.reconnect(getTransactionServiceGroup()), SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
+        timerExecutor.scheduleAtFixedRate(() -> {
+            try {
+                clientChannelManager.reconnect(getTransactionServiceGroup());
+            } catch (Exception ex) {
+                LOGGER.warn("reconnect server failed. {}", ex.getMessage());
+            }
+        }, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
         if (this.isEnableClientBatchSendRequest()) {
             mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                 MAX_MERGE_SEND_THREAD,

core/src/main/java/io/seata/core/rpc/netty/NettyClientChannelManager.java

@@ -158,13 +158,73 @@ void destroyChannel(String serverAddress, Channel channel) {
         }
     }
 
+    /**
+     * Reconnect to remote server of current transaction service group.
+     *
+     * @param transactionServiceGroup transaction service group
+     */
+    void reconnect(String transactionServiceGroup) {
+        doReconnect(transactionServiceGroup, false);
+    }
+
+    /**
+     * Init reconnect to remote server of current transaction service group.
+     * @param transactionServiceGroup
+     * @param failFast
+     */
+    void initReconnect(String transactionServiceGroup, boolean failFast) {
+        doReconnect(transactionServiceGroup, failFast);
+    }
+
+    /**
+     * reconnect to remote server of current transaction service group.
+     * @param transactionServiceGroup
+     * @param failFast
+     */
+    void doReconnect(String transactionServiceGroup, boolean failFast) {
+        List<String> availList;
+        try {
+            availList = getAvailServerList(transactionServiceGroup);
+        } catch (Exception e) {
+            LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);
+            throwFailFastException(failFast, "Failed to get available servers");
+            return;
+        }
+        if (CollectionUtils.isEmpty(availList)) {
+            RegistryService registryService = RegistryFactory.getInstance();
+            String clusterName = registryService.getServiceGroup(transactionServiceGroup);
+
+            if (StringUtils.isBlank(clusterName)) {
+                LOGGER.error("can not get cluster name in registry config '{}{}', please make sure registry config correct",
+                        ConfigurationKeys.SERVICE_GROUP_MAPPING_PREFIX,
+                        transactionServiceGroup);
+                throwFailFastException(failFast, "can not get cluster name in registry config.");
+                return;
+            }
+
+            if (!(registryService instanceof FileRegistryServiceImpl)) {
+                LOGGER.error("no available service found in cluster '{}', please make sure registry config correct and keep your seata server running", clusterName);
+            }
+            throwFailFastException(failFast, "no available service found in cluster.");
+            return;
+        }
+        try {
+            doReconnect(availList, transactionServiceGroup);
+        } catch (Exception e) {
+            if (failFast) {
+                throw e;
+            }
+            LOGGER.error("connect server failed. {}", e.getMessage(), e);
+        }
+    }
+
     /**
      * Reconnect to remote server of current transaction service group.
      *
      * @param availList avail list
      * @param transactionServiceGroup transaction service group
      */
-    void reconnect(List<String> availList, String transactionServiceGroup) {
+    void doReconnect(List<String> availList, String transactionServiceGroup) {
         Set<String> channelAddress = new HashSet<>(availList.size());
         Map<String, Exception> failedMap = new HashMap<>();
         try {
@@ -178,12 +238,17 @@ void reconnect(List<String> availList, String transactionServiceGroup) {
             }
             if (failedMap.size() > 0) {
                 if (LOGGER.isInfoEnabled()) {
-                    LOGGER.error("{} can not connect to {} cause:{}", FrameworkErrorCode.NetConnect.getErrCode(), failedMap.keySet(), failedMap.values().stream().map(Throwable::getMessage).collect(Collectors.toSet()));
+                    LOGGER.error("{} can not connect to {} cause:{}", FrameworkErrorCode.NetConnect.getErrCode(),
+                            failedMap.keySet(),
+                            failedMap.values().stream().map(Throwable::getMessage).collect(Collectors.toSet()));
                 } else if (LOGGER.isDebugEnabled()) {
                     failedMap.forEach((key, value) -> {
-                        LOGGER.error("{} can not connect to {} cause:{} trace information:{}", FrameworkErrorCode.NetConnect.getErrCode(), key, value.getMessage(), value);
+                        LOGGER.error("{} can not connect to {} cause:{} trace information:{}",
+                                FrameworkErrorCode.NetConnect.getErrCode(), key, value.getMessage(), value);
                     });
                 }
+                String invalidAddress = StringUtils.join(failedMap.keySet().iterator(), ", ");
+                throw new FrameworkException("can not connect to [" + invalidAddress + "]");
             }
         } finally {
             if (CollectionUtils.isNotEmpty(channelAddress)) {
@@ -199,38 +264,6 @@ void reconnect(List<String> availList, String transactionServiceGroup) {
         }
     }
 
-    /**
-     * Reconnect to remote server of current transaction service group.
-     *
-     * @param transactionServiceGroup transaction service group
-     */
-    void reconnect(String transactionServiceGroup) {
-        List<String> availList;
-        try {
-            availList = getAvailServerList(transactionServiceGroup);
-        } catch (Exception e) {
-            LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);
-            return;
-        }
-        if (CollectionUtils.isEmpty(availList)) {
-            RegistryService registryService = RegistryFactory.getInstance();
-            String clusterName = registryService.getServiceGroup(transactionServiceGroup);
-
-            if (StringUtils.isBlank(clusterName)) {
-                LOGGER.error("can not get cluster name in registry config '{}{}', please make sure registry config correct",
-                        ConfigurationKeys.SERVICE_GROUP_MAPPING_PREFIX,
-                        transactionServiceGroup);
-                return;
-            }
-
-            if (!(registryService instanceof FileRegistryServiceImpl)) {
-                LOGGER.error("no available service found in cluster '{}', please make sure registry config correct and keep your seata server running", clusterName);
-            }
-            return;
-        }
-        reconnect(availList, transactionServiceGroup);
-    }
-
     void invalidateObject(final String serverAddress, final Channel channel) throws Exception {
         nettyClientKeyPool.invalidateObject(poolKeyMap.get(serverAddress), channel);
     }
@@ -297,5 +330,12 @@ private Channel getExistAliveChannel(Channel rmChannel, String serverAddress) {
         }
         return null;
     }
+
+    private void throwFailFastException(boolean failFast, String message) {
+        if (failFast) {
+            throw new FrameworkException(message);
+        }
+    }
+
 }
 

core/src/main/java/io/seata/core/rpc/netty/NettyClientConfig.java

@@ -455,4 +455,5 @@ public String getRmDispatchThreadPrefix() {
     public static boolean isEnableClientBatchSendRequest() {
         return ENABLE_CLIENT_BATCH_SEND_REQUEST;
     }
+
 }

core/src/main/java/io/seata/core/rpc/netty/RmNettyRemotingClient.java

@@ -80,7 +80,10 @@ public void init() {
             if (resourceManager != null
                     && !resourceManager.getManagedResources().isEmpty()
                     && StringUtils.isNotBlank(transactionServiceGroup)) {
-                getClientChannelManager().reconnect(transactionServiceGroup);
+                boolean failFast = ConfigurationFactory.getInstance().getBoolean(
+                        ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
+                        DefaultValues.DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST);
+                getClientChannelManager().initReconnect(transactionServiceGroup, failFast);
             }
         }
     }
@@ -214,7 +217,10 @@ public void registerResource(String resourceGroupId, String resourceId) {
         }
 
         if (getClientChannelManager().getChannels().isEmpty()) {
-            getClientChannelManager().reconnect(transactionServiceGroup);
+            boolean failFast = ConfigurationFactory.getInstance().getBoolean(
+                    ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
+                    DefaultValues.DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST);
+            getClientChannelManager().initReconnect(transactionServiceGroup, failFast);
             return;
         }
         synchronized (getClientChannelManager().getChannels()) {

core/src/main/java/io/seata/core/rpc/netty/TmNettyRemotingClient.java

@@ -283,7 +283,10 @@ private String getExtraData() {
     }
 
     private void initConnection() {
-        getClientChannelManager().reconnect(transactionServiceGroup);
+        boolean failFast = ConfigurationFactory.getInstance().getBoolean(
+                ConfigurationKeys.ENABLE_TM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
+                DefaultValues.DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST);
+        getClientChannelManager().initReconnect(transactionServiceGroup, failFast);
     }
 
 }

core/src/test/resources/file.conf

@@ -22,4 +22,8 @@ service {
   default.grouplist = "127.0.0.1:8091"
   #disable seata
   disableGlobalTransaction = false
+}
+transport {
+  enableRmClientChannelCheckFailFast = false
+  enableTmClientChannelCheckFailFast = false
 }

test/src/test/resources/file.conf

@@ -46,8 +46,12 @@ transport {
   }
   serialization = "seata"
   compressor = "none"
+
+  enableRmClientChannelCheckFailFast = false
+  enableTmClientChannelCheckFailFast = false
 }
 
+
 service {
   #transaction service group mapping
   vgroupMapping.default_tx_group = "default"

tm/src/test/resources/file.conf

@@ -22,4 +22,8 @@ service {
   default.grouplist = "127.0.0.1:8091"
   #disable seata
   disableGlobalTransaction = false
-}
+}
+transport {
+  enableRmClientChannelCheckFailFast = false
+  enableTmClientChannelCheckFailFast = false
+}