You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by ji...@apache.org on 2024/01/04 10:01:43 UTC
(incubator-seata) branch 2.x updated: optimize: RM TM startup connect server fail fast (#6004)
This is an automated email from the ASF dual-hosted git repository.
jianbin pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new 2363715c05 optimize: RM TM startup connect server fail fast (#6004)
2363715c05 is described below
commit 2363715c05529ddd6d5aa6c2472bd550e4c7cabf
Author: jsbxyyx <js...@163.com>
AuthorDate: Thu Jan 4 18:01:38 2024 +0800
optimize: RM TM startup connect server fail fast (#6004)
---
changes/en-us/2.0.0.md | 1 -
changes/en-us/2.x.md | 2 +
changes/zh-cn/2.x.md | 2 +
.../java/io/seata/common/ConfigurationKeys.java | 10 ++
.../main/java/io/seata/common/DefaultValues.java | 2 +
.../java/io/seata/common/util/StringUtils.java | 28 ++++++
.../core/rpc/netty/AbstractNettyRemoting.java | 1 +
.../rpc/netty/AbstractNettyRemotingClient.java | 8 +-
.../core/rpc/netty/NettyClientChannelManager.java | 110 ++++++++++++++-------
.../io/seata/core/rpc/netty/NettyClientConfig.java | 1 +
.../core/rpc/netty/RmNettyRemotingClient.java | 10 +-
.../core/rpc/netty/TmNettyRemotingClient.java | 5 +-
.../io/seata/core/rpc/netty/RmNettyClientTest.java | 50 ++++++++++
.../io/seata/core/rpc/netty/TmNettyClientTest.java | 31 ++++++
core/src/test/resources/file.conf | 4 +
test/src/test/resources/file.conf | 4 +
tm/src/test/resources/file.conf | 6 +-
17 files changed, 234 insertions(+), 41 deletions(-)
diff --git a/changes/en-us/2.0.0.md b/changes/en-us/2.0.0.md
index 6768f6311c..7b8757be94 100644
--- a/changes/en-us/2.0.0.md
+++ b/changes/en-us/2.0.0.md
@@ -156,7 +156,6 @@ The version is updated as follows:
- [[#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
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 5cca46050e..212713e365 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -46,6 +46,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#6201](https://github.com/apache/incubator-seata/pull/6201)] restore required_status_checks kept to remove context validation
- [[#6218](https://github.com/apache/incubator-seata/pull/6218)] remove Seata-Docker link
- [[#6227](https://github.com/apache/incubator-seata/pull/6227)] validate that the primary key is free of illegal characters
+- [[#6004](https://github.com/seata/seata/pull/6004)] optimize RM TM startup connect server fail fast
### security:
- [[#6069](https://github.com/apache/incubator-seata/pull/6069)] Upgrade Guava dependencies to fix security vulnerabilities
@@ -80,5 +81,6 @@ Thanks to these contributors for their code commits. Please report an unintended
- [PeppaO](https://github.com/PeppaO)
- [AlbumenJ](https://github.com/AlbumenJ)
- [dreamskyvision](https://github.com/dreamskyvision)
+- [jsbxyyx](https://github.com/jsbxyyx)
Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index a8d6b4e099..033e850c5a 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -46,6 +46,7 @@
- [[#6201](https://github.com/apache/incubator-seata/pull/6201)] 恢复required_status_checks但去除context校验
- [[#6218](https://github.com/apache/incubator-seata/pull/6218)] 移除Seata-Docker链接
- [[#6227](https://github.com/apache/incubator-seata/pull/6227)] 校验pk中不含逗号
+- [[#6004](https://github.com/seata/seata/pull/6004)] 优化RM,TM连接server快速失败
### security:
- [[#6069](https://github.com/apache/incubator-seata/pull/6069)] 升级Guava依赖版本，修复安全漏洞
@@ -79,5 +80,6 @@
- [PeppaO](https://github.com/PeppaO)
- [AlbumenJ](https://github.com/AlbumenJ)
- [dreamskyvision](https://github.com/dreamskyvision)
+- [jsbxyyx](https://github.com/jsbxyyx)
同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。
diff --git a/common/src/main/java/io/seata/common/ConfigurationKeys.java b/common/src/main/java/io/seata/common/ConfigurationKeys.java
index b54b3849cc..55ffe2d7d5 100644
--- a/common/src/main/java/io/seata/common/ConfigurationKeys.java
+++ b/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
*/
diff --git a/common/src/main/java/io/seata/common/DefaultValues.java b/common/src/main/java/io/seata/common/DefaultValues.java
index 56cc225a55..0db7db8b0d 100644
--- a/common/src/main/java/io/seata/common/DefaultValues.java
+++ b/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";
diff --git a/common/src/main/java/io/seata/common/util/StringUtils.java b/common/src/main/java/io/seata/common/util/StringUtils.java
index 94cc53d43f..a4a1e8e311 100644
--- a/common/src/main/java/io/seata/common/util/StringUtils.java
+++ b/common/src/main/java/io/seata/common/util/StringUtils.java
@@ -25,6 +25,7 @@ import java.nio.charset.StandardCharsets;
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 class StringUtils {
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();
+ }
+
}
diff --git a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java b/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java
index cad610c6ed..44f4547f7e 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java
+++ b/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemoting.java
@@ -58,6 +58,7 @@ import org.slf4j.MDC;
public abstract class AbstractNettyRemoting implements Disposable {
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyRemoting.class);
+
/**
* The Timer executor.
*/
diff --git a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java b/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java
index 2a9248828c..9f62732cc7 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingClient.java
+++ b/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,
diff --git a/core/src/main/java/io/seata/core/rpc/netty/NettyClientChannelManager.java b/core/src/main/java/io/seata/core/rpc/netty/NettyClientChannelManager.java
index 956f80b6bb..7cd79a76fe 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/NettyClientChannelManager.java
+++ b/core/src/main/java/io/seata/core/rpc/netty/NettyClientChannelManager.java
@@ -158,13 +158,73 @@ class NettyClientChannelManager {
}
}
+ /**
+ * 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 @@ class NettyClientChannelManager {
}
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 @@ class NettyClientChannelManager {
}
}
- /**
- * 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 @@ class NettyClientChannelManager {
}
return null;
}
+
+ private void throwFailFastException(boolean failFast, String message) {
+ if (failFast) {
+ throw new FrameworkException(message);
+ }
+ }
+
}
diff --git a/core/src/main/java/io/seata/core/rpc/netty/NettyClientConfig.java b/core/src/main/java/io/seata/core/rpc/netty/NettyClientConfig.java
index fbd5cba79d..fb5a151b16 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/NettyClientConfig.java
+++ b/core/src/main/java/io/seata/core/rpc/netty/NettyClientConfig.java
@@ -455,4 +455,5 @@ public class NettyClientConfig extends NettyBaseConfig {
public static boolean isEnableClientBatchSendRequest() {
return ENABLE_CLIENT_BATCH_SEND_REQUEST;
}
+
}
diff --git a/core/src/main/java/io/seata/core/rpc/netty/RmNettyRemotingClient.java b/core/src/main/java/io/seata/core/rpc/netty/RmNettyRemotingClient.java
index 5855ad3229..35a6f1eed5 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/RmNettyRemotingClient.java
+++ b/core/src/main/java/io/seata/core/rpc/netty/RmNettyRemotingClient.java
@@ -80,7 +80,10 @@ public final class RmNettyRemotingClient extends AbstractNettyRemotingClient {
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 final class RmNettyRemotingClient extends AbstractNettyRemotingClient {
}
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()) {
diff --git a/core/src/main/java/io/seata/core/rpc/netty/TmNettyRemotingClient.java b/core/src/main/java/io/seata/core/rpc/netty/TmNettyRemotingClient.java
index 76de2691b5..06579842a9 100644
--- a/core/src/main/java/io/seata/core/rpc/netty/TmNettyRemotingClient.java
+++ b/core/src/main/java/io/seata/core/rpc/netty/TmNettyRemotingClient.java
@@ -283,7 +283,10 @@ public final class TmNettyRemotingClient extends AbstractNettyRemotingClient {
}
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);
}
}
diff --git a/core/src/test/java/io/seata/core/rpc/netty/RmNettyClientTest.java b/core/src/test/java/io/seata/core/rpc/netty/RmNettyClientTest.java
index 676ecf2c63..692de3264d 100644
--- a/core/src/test/java/io/seata/core/rpc/netty/RmNettyClientTest.java
+++ b/core/src/test/java/io/seata/core/rpc/netty/RmNettyClientTest.java
@@ -16,10 +16,27 @@
*/
package io.seata.core.rpc.netty;
+import io.seata.common.ConfigurationKeys;
+import io.seata.common.DefaultValues;
+import io.seata.common.exception.FrameworkException;
+import io.seata.config.ConfigurationCache;
+import io.seata.config.ConfigurationChangeEvent;
+import io.seata.config.ConfigurationChangeListener;
+import io.seata.config.ConfigurationFactory;
+import io.seata.core.model.Resource;
+import io.seata.core.model.ResourceManager;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -31,6 +48,20 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
class RmNettyClientTest {
+ Logger logger = LoggerFactory.getLogger(getClass());
+
+ @BeforeAll
+ public static void beforeAll() {
+ RmNettyRemotingClient.getInstance().destroy();
+ System.setProperty(ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST, "true");
+ }
+
+ @AfterAll
+ public static void afterAll() {
+ RmNettyRemotingClient.getInstance().destroy();
+ System.setProperty(ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST, "false");
+ }
+
@Test
public void assertGetInstanceAfterDestroy() {
RmNettyRemotingClient oldClient = RmNettyRemotingClient.getInstance("ap", "group");
@@ -47,6 +78,25 @@ class RmNettyClientTest {
assertTrue(initialized.get());
newClient.destroy();
}
+
+ @Test
+ public void testCheckFailFast() throws Exception {
+ RmNettyRemotingClient newClient = RmNettyRemotingClient.getInstance("fail_fast", "default_tx_group");
+
+ ResourceManager resourceManager = Mockito.mock(ResourceManager.class);
+ Resource mockResource = Mockito.mock(Resource.class);
+ Map<String, Resource> resourceMap = new HashMap<>();
+ resourceMap.put("jdbc:xx://localhost/test", mockResource);
+ Mockito.when(resourceManager.getManagedResources()).thenReturn(resourceMap);
+ newClient.setResourceManager(resourceManager);
+ System.setProperty("file.listener.enabled", "true");
+ ConfigurationCache.addConfigListener(ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
+ event -> logger.info("dataId:{}, value: {}, oldValue: {}", event.getDataId(), event.getNewValue(),
+ event.getOldValue()));
+ System.setProperty(ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST, "true");
+ Thread.sleep(2000);
+ Assertions.assertThrows(FrameworkException.class, newClient::init);
+ }
private AtomicBoolean getInitializeStatus(final RmNettyRemotingClient rmNettyRemotingClient) {
try {
diff --git a/core/src/test/java/io/seata/core/rpc/netty/TmNettyClientTest.java b/core/src/test/java/io/seata/core/rpc/netty/TmNettyClientTest.java
index 28db3385cd..48f29fedff 100644
--- a/core/src/test/java/io/seata/core/rpc/netty/TmNettyClientTest.java
+++ b/core/src/test/java/io/seata/core/rpc/netty/TmNettyClientTest.java
@@ -21,9 +21,19 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.nio.NioSocketChannel;
+import io.seata.common.ConfigurationKeys;
+import io.seata.common.DefaultValues;
+import io.seata.common.exception.FrameworkException;
+import io.seata.config.ConfigurationCache;
+import io.seata.config.ConfigurationChangeEvent;
+import io.seata.config.ConfigurationFactory;
import org.apache.commons.pool.impl.GenericKeyedObjectPool;
+import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
import java.util.Map;
@@ -37,6 +47,8 @@ import java.util.concurrent.TimeUnit;
*/
public class TmNettyClientTest {
+ Logger logger = LoggerFactory.getLogger(getClass());
+
private static final ThreadPoolExecutor
workingThreads = new ThreadPoolExecutor(100, 500, 500, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(20000), new ThreadPoolExecutor.CallerRunsPolicy());
@@ -123,6 +135,25 @@ public class TmNettyClientTest {
}
+ @AfterAll
+ public static void afterAll() {
+ TmNettyRemotingClient.getInstance().destroy();
+ System.setProperty(ConfigurationKeys.ENABLE_TM_CLIENT_CHANNEL_CHECK_FAIL_FAST, "false");
+ }
+
+ @Test
+ public void testCheckFailFast() throws Exception {
+ TmNettyRemotingClient.getInstance().destroy();
+ TmNettyRemotingClient tmClient = TmNettyRemotingClient.getInstance("fail_fast", "default_tx_group");
+ System.setProperty("file.listener.enabled", "true");
+ ConfigurationCache.addConfigListener(ConfigurationKeys.ENABLE_TM_CLIENT_CHANNEL_CHECK_FAIL_FAST,
+ event -> logger.info("dataId:{}, value: {}, oldValue: {}", event.getDataId(), event.getNewValue(),
+ event.getOldValue()));
+ System.setProperty(ConfigurationKeys.ENABLE_TM_CLIENT_CHANNEL_CHECK_FAIL_FAST, "true");
+ Thread.sleep(2000);
+ Assertions.assertThrows(FrameworkException.class, tmClient::init);
+ }
+
/**
* get private field in parent class
*
diff --git a/core/src/test/resources/file.conf b/core/src/test/resources/file.conf
index 46c3e0401c..789450fc1e 100644
--- a/core/src/test/resources/file.conf
+++ b/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
}
\ No newline at end of file
diff --git a/test/src/test/resources/file.conf b/test/src/test/resources/file.conf
index 70d6dc3ab1..7cb8339968 100644
--- a/test/src/test/resources/file.conf
+++ b/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"
diff --git a/tm/src/test/resources/file.conf b/tm/src/test/resources/file.conf
index 46c3e0401c..0f73855d5d 100644
--- a/tm/src/test/resources/file.conf
+++ b/tm/src/test/resources/file.conf
@@ -22,4 +22,8 @@ service {
default.grouplist = "127.0.0.1:8091"
#disable seata
disableGlobalTransaction = false
-}
\ No newline at end of file
+}
+transport {
+ enableRmClientChannelCheckFailFast = false
+ enableTmClientChannelCheckFailFast = false
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org