This is an automated email from the ASF dual-hosted git repository.
funky-eyes 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 0b0101b7c9 optimize: supports switching http protocols based on server
versions in RaftRegistryService (#7952)
0b0101b7c9 is described below
commit 0b0101b7c9b46b17694e7863e684f393b01d090e
Author: Tsukilc <[email protected]>
AuthorDate: Wed Apr 29 14:26:00 2026 +0800
optimize: supports switching http protocols based on server versions in
RaftRegistryService (#7952)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 1 +
discovery/seata-discovery-raft/pom.xml | 7 +-
.../registry/raft/RaftRegistryServiceImpl.java | 421 ++++++++++++---
.../registry/raft/RaftRegistryServiceImplTest.java | 564 ++++++++++++++++++---
5 files changed, 862 insertions(+), 132 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 0fc495b560..db0376a775 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -32,6 +32,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#8020](https://github.com/apache/incubator-seata/pull/8020)] add
UnregisterRM protocol to notify server on client destroy
- [[#8044](https://github.com/apache/incubator-seata/pull/8044)] add protobuf
serialization support for UnregisterRM protocol
- [[#8046](https://github.com/apache/incubator-seata/pull/8046)] add fastjson2
and jackson3
+- [[#7952](https://github.com/apache/incubator-seata/pull/7952)] Support
runtime dynamic switching between HTTP/1.1 and HTTP/2 for Raft client watch,
with a 2.7.0 version threshold and compatibility fallback
### bugfix:
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index c7140a9767..581d7b7d5b 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -33,6 +33,7 @@
- [[#8020](https://github.com/apache/incubator-seata/pull/8020)] 新增
UnregisterRM 协议，在客户端销毁时通知服务端
- [[#8044](https://github.com/apache/incubator-seata/pull/8044)] 为
UnregisterRM 协议添加 protobuf 序列化支持
- [[#8046](https://github.com/apache/incubator-seata/pull/8046)] 添加了 fastjson2
和 jackson3
+- [[#7952](https://github.com/apache/incubator-seata/pull/7952)]
在Raft客户端支持Watch在HTTP/1.1与HTTP/2之间运行时动态切换
### bugfix:
diff --git a/discovery/seata-discovery-raft/pom.xml
b/discovery/seata-discovery-raft/pom.xml
index a4f3a7a209..fb606cde2d 100644
--- a/discovery/seata-discovery-raft/pom.xml
+++ b/discovery/seata-discovery-raft/pom.xml
@@ -35,6 +35,11 @@
<artifactId>seata-discovery-core</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.seata</groupId>
+ <artifactId>seata-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
@@ -48,4 +53,4 @@
<artifactId>jackson-databind</artifactId>
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git
a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
index aa154aed4d..942f34701b 100644
---
a/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
+++
b/discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
@@ -28,6 +28,7 @@ import
org.apache.seata.common.exception.AuthenticationFailedException;
import org.apache.seata.common.exception.NotSupportYetException;
import org.apache.seata.common.exception.ParseEndpointException;
import org.apache.seata.common.exception.RetryableException;
+import org.apache.seata.common.metadata.ClusterWatchEvent;
import org.apache.seata.common.metadata.Metadata;
import org.apache.seata.common.metadata.MetadataResponse;
import org.apache.seata.common.metadata.Node;
@@ -35,10 +36,12 @@ import org.apache.seata.common.thread.NamedThreadFactory;
import org.apache.seata.common.util.CollectionUtils;
import org.apache.seata.common.util.HttpClientUtil;
import org.apache.seata.common.util.NetUtil;
+import org.apache.seata.common.util.SeataHttpWatch;
import org.apache.seata.common.util.StringUtils;
import org.apache.seata.config.ConfigChangeListener;
import org.apache.seata.config.Configuration;
import org.apache.seata.config.ConfigurationFactory;
+import org.apache.seata.core.protocol.Version;
import org.apache.seata.discovery.registry.RegistryService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,10 +52,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
@@ -64,7 +69,6 @@ import java.util.stream.Stream;
/**
* The type File registry service.
- *
*/
public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeListener> {
@@ -114,6 +118,22 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
private static final AtomicBoolean CLOSED = new AtomicBoolean(false);
+ private static final long DEFAULT_METADATA_MAX_AGE_MS = 30000L;
+
+ private static final long WATCH_TIMEOUT_MS = 30000L;
+
+ private static final long RETRY_DELAY_MS = 1000L;
+
+ private static final int HTTP2_WATCH_READ_TIMEOUT_SECONDS = 30;
+
+ private static final String MIN_HTTP2_VERSION = "2.7.0";
+
+ private static volatile WatchProtocol CURRENT_WATCH_PROTOCOL =
WatchProtocol.HTTP1;
+
+ private static volatile SeataHttpWatch<ClusterWatchEvent> HTTP2_WATCH;
+
+ private static volatile String HTTP2_WATCH_GROUP;
+
/**
* Service node health check
*/
@@ -121,6 +141,18 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
private static final String PREFERRED_NETWORKS;
+ /**
+ * Protocol used by watch mechanism.
+ */
+ private enum WatchProtocol {
+
+ /** HTTP/1.x protocol */
+ HTTP1,
+
+ /** HTTP/2 protocol */
+ HTTP2
+ }
+
static {
TOKEN_EXPIRE_TIME_IN_MILLISECONDS =
CONFIG.getLong(getTokenExpireTimeInMillisecondsKey(), 29 * 60 * 1000L);
USERNAME = CONFIG.getConfig(getRaftUserNameKey());
@@ -146,6 +178,7 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
return instance;
}
+ @SuppressWarnings("AliDeprecation")
@Override
public void register(InetSocketAddress address) throws Exception {}
@@ -170,23 +203,20 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
new LinkedBlockingQueue<>(),
new NamedThreadFactory("refreshMetadata", 1,
true));
REFRESH_METADATA_EXECUTOR.execute(() -> {
- long metadataMaxAgeMs =
CONFIG.getLong(getMetadataMaxAgeMs(), 30000L);
+ long metadataMaxAgeMs =
CONFIG.getLong(getMetadataMaxAgeMs(), DEFAULT_METADATA_MAX_AGE_MS);
long currentTime = System.currentTimeMillis();
while (!CLOSED.get()) {
try {
- // Forced refresh of metadata information
after set age
boolean fetch = System.currentTimeMillis() -
currentTime > metadataMaxAgeMs;
String clusterName =
CURRENT_TRANSACTION_CLUSTER_NAME;
if (!fetch) {
fetch = watch();
}
- // Cluster changes or reaches timeout refresh
time
if (fetch) {
for (String group :
METADATA.groups(clusterName)) {
try {
acquireClusterMetaData(clusterName, group);
} catch (Exception e) {
- // prevents an exception from
being thrown that causes the thread to break
if (e instanceof
RetryableException) {
throw e;
} else {
@@ -203,21 +233,305 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
} catch (RetryableException e) {
LOGGER.error(e.getMessage(), e);
try {
- Thread.sleep(1000);
+ Thread.sleep(RETRY_DELAY_MS);
} catch (InterruptedException ignored) {
}
}
}
+ closeHttp2Watch();
});
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
CLOSED.compareAndSet(false, true);
- REFRESH_METADATA_EXECUTOR.shutdown();
+ closeHttp2Watch();
+ if (REFRESH_METADATA_EXECUTOR != null) {
+ REFRESH_METADATA_EXECUTOR.shutdown();
+ }
}));
}
}
}
}
+ private static boolean watch() throws RetryableException {
+ String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
+ if (StringUtils.isBlank(clusterName)) {
+ return false;
+ }
+
+ WatchProtocol targetProtocol = resolveWatchProtocol(clusterName);
+ switchWatchProtocolIfNecessary(targetProtocol);
+
+ if (targetProtocol == WatchProtocol.HTTP2) {
+ return watchHttp2(clusterName);
+ }
+ return watchHttp1(clusterName);
+ }
+
+ private static void switchWatchProtocolIfNecessary(WatchProtocol
targetProtocol) {
+ if (CURRENT_WATCH_PROTOCOL == targetProtocol) {
+ return;
+ }
+
+ LOGGER.info("Switching raft watch protocol from {} to {}",
CURRENT_WATCH_PROTOCOL, targetProtocol);
+ if (targetProtocol == WatchProtocol.HTTP1) {
+ closeHttp2Watch();
+ }
+ CURRENT_WATCH_PROTOCOL = targetProtocol;
+ }
+
+ private static WatchProtocol resolveWatchProtocol(String clusterName) {
+ if (StringUtils.isBlank(clusterName)) {
+ return WatchProtocol.HTTP1;
+ }
+
+ Set<String> groups = METADATA.groups(clusterName);
+ if (CollectionUtils.isEmpty(groups)) {
+ return WatchProtocol.HTTP1;
+ }
+
+ boolean hasNode = false;
+ for (String group : groups) {
+ List<Node> nodes = METADATA.getNodes(clusterName, group);
+ if (CollectionUtils.isEmpty(nodes)) {
+ continue;
+ }
+ hasNode = true;
+ if (!isClusterHttp2Enabled(clusterName, group)) {
+ return WatchProtocol.HTTP1;
+ }
+ }
+
+ return hasNode ? WatchProtocol.HTTP2 : WatchProtocol.HTTP1;
+ }
+
+ private static boolean watchHttp1(String clusterName) throws
RetryableException {
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE,
ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+ Map<String, String> param = new HashMap<>();
+ Map<String, Long> groupTerms = METADATA.getClusterTerm(clusterName);
+ groupTerms.forEach((k, v) -> param.put(k, String.valueOf(v)));
+ for (String group : groupTerms.keySet()) {
+ String tcAddress = queryHttpAddress(clusterName, group);
+ if (StringUtils.isBlank(tcAddress)) {
+ return false;
+ }
+ if (isTokenExpired()) {
+ refreshToken(tcAddress);
+ }
+ if (StringUtils.isNotBlank(jwtToken)) {
+ header.put(AUTHORIZATION_HEADER, jwtToken);
+ }
+ try (Response response = HttpClientUtil.doPost(
+ "http://"; + tcAddress + "/metadata/v1/watch", param,
header, (int) WATCH_TIMEOUT_MS)) {
+ if (response != null) {
+ int statusCode = response.code();
+ if (statusCode == HttpStatus.SC_UNAUTHORIZED) {
+ if (StringUtils.isNotBlank(USERNAME) &&
StringUtils.isNotBlank(PASSWORD)) {
+ throw new RetryableException("Authentication
failed!");
+ } else {
+ throw new AuthenticationFailedException(
+ "Authentication failed! you should
configure the correct username and password.");
+ }
+ }
+ return statusCode == HttpStatus.SC_OK;
+ }
+ } catch (IOException e) {
+ LOGGER.error("watch cluster node: {}, fail: {}", tcAddress,
e.getMessage());
+ throw new RetryableException(e.getMessage(), e);
+ }
+ break;
+ }
+ return false;
+ }
+
+ private static boolean watchHttp2(String clusterName) throws
RetryableException {
+ Map<String, Long> groupTerms = METADATA.getClusterTerm(clusterName);
+ if (CollectionUtils.isEmpty(groupTerms)) {
+ return false;
+ }
+
+ String group = selectWatchGroup(groupTerms);
+ if (StringUtils.isBlank(group)) {
+ return false;
+ }
+ String tcAddress = queryHttpAddress(clusterName, group);
+ if (StringUtils.isBlank(tcAddress)) {
+ return false;
+ }
+
+ Map<String, String> header = new HashMap<>();
+ header.put(HTTP.CONTENT_TYPE,
ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+
+ Map<String, String> param = new HashMap<>();
+ groupTerms.forEach((k, v) -> param.put(k, String.valueOf(v)));
+
+ if (isTokenExpired()) {
+ refreshToken(tcAddress);
+ }
+ if (StringUtils.isNotBlank(jwtToken)) {
+ header.put(AUTHORIZATION_HEADER, jwtToken);
+ }
+
+ ensureHttp2Watch(group, tcAddress, param, header);
+ SeataHttpWatch<ClusterWatchEvent> watch = HTTP2_WATCH;
+ if (watch == null) {
+ return false;
+ }
+
+ try {
+ SeataHttpWatch.Response<ClusterWatchEvent> response = watch.next();
+ return shouldRefreshMetadata(clusterName, group, response);
+ } catch (RuntimeException e) {
+ if (CLOSED.get()) {
+ closeHttp2Watch();
+ return false;
+ }
+ closeHttp2Watch();
+ throw new RetryableException("HTTP2 watch failed", e);
+ }
+ }
+
+ private static String selectWatchGroup(Map<String, Long> groupTerms) {
+ if (CollectionUtils.isEmpty(groupTerms)) {
+ return null;
+ }
+
+ if (StringUtils.isNotBlank(HTTP2_WATCH_GROUP) &&
groupTerms.containsKey(HTTP2_WATCH_GROUP)) {
+ return HTTP2_WATCH_GROUP;
+ }
+
+ List<String> groups = new ArrayList<>(groupTerms.keySet());
+ Collections.sort(groups);
+ return groups.get(0);
+ }
+
+ private static synchronized void ensureHttp2Watch(
+ String group, String tcAddress, Map<String, String> param,
Map<String, String> header)
+ throws RetryableException {
+
+ if (HTTP2_WATCH != null && StringUtils.equals(group,
HTTP2_WATCH_GROUP)) {
+ return;
+ }
+
+ closeHttp2Watch();
+
+ try {
+ HTTP2_WATCH = HttpClientUtil.watchPost(
+ "http://"; + tcAddress + "/metadata/v1/watch",
+ param,
+ header,
+ ClusterWatchEvent.class,
+ HTTP2_WATCH_READ_TIMEOUT_SECONDS);
+ HTTP2_WATCH_GROUP = group;
+ } catch (IOException e) {
+ closeHttp2Watch();
+ throw new RetryableException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeHttp2Watch();
+ if (e.getMessage() != null && e.getMessage().contains("401")) {
+ tokenTimeStamp = -1;
+ }
+ throw new RetryableException("Failed to create HTTP2 watch", e);
+ }
+ }
+
+ private static boolean shouldRefreshMetadata(
+ String clusterName, String defaultGroup,
SeataHttpWatch.Response<ClusterWatchEvent> response) {
+
+ if (response == null
+ || response.type != SeataHttpWatch.Response.Type.UPDATE
+ || response.object == null
+ || response.object.getMetadata() == null
+ ||
CollectionUtils.isEmpty(response.object.getMetadata().getNodes())) {
+ return false;
+ }
+
+ ClusterWatchEvent event = response.object;
+ MetadataResponse incomingMetadata = event.getMetadata();
+
+ String eventGroup = StringUtils.isNotBlank(event.getGroup()) ?
event.getGroup() : defaultGroup;
+ long localTerm =
METADATA.getClusterTerm(clusterName).getOrDefault(eventGroup, -1L);
+ if (incomingMetadata.getTerm() < localTerm) {
+ return false;
+ }
+ boolean termAdvanced = incomingMetadata.getTerm() > localTerm;
+
+ boolean changed = termAdvanced || hasMetadataChanged(clusterName,
eventGroup, incomingMetadata);
+
+ if (changed) {
+ METADATA.refreshMetadata(clusterName, incomingMetadata);
+ }
+
+ return changed;
+ }
+
+ private static boolean hasMetadataChanged(String clusterName, String
group, MetadataResponse incomingMetadata) {
+ if (incomingMetadata == null) {
+ return false;
+ }
+
+ List<Node> incomingNodes = incomingMetadata.getNodes();
+ List<Node> localNodes = METADATA.getNodes(clusterName, group);
+
+ if (CollectionUtils.isEmpty(localNodes) !=
CollectionUtils.isEmpty(incomingNodes)) {
+ return true;
+ }
+
+ if (CollectionUtils.isEmpty(localNodes)) {
+ return false;
+ }
+
+ if (incomingMetadata.getTerm() >
METADATA.getClusterTerm(clusterName).getOrDefault(group, -1L)) {
+ return true;
+ }
+
+ if (localNodes.size() != incomingNodes.size()) {
+ return true;
+ }
+
+ return
!buildNodeSignatures(localNodes).equals(buildNodeSignatures(incomingNodes));
+ }
+
+ private static Set<String> buildNodeSignatures(List<Node> nodes) {
+ Set<String> signatures = new HashSet<>();
+ for (Node node : nodes) {
+ signatures.add(buildNodeSignature(node));
+ }
+ return signatures;
+ }
+
+ private static String buildNodeSignature(Node node) {
+ if (node == null) {
+ return "";
+ }
+
+ String control = node.getControl() == null
+ ? ""
+ : node.getControl().getHost()
+ + IP_PORT_SPLIT_CHAR
+ + node.getControl().getPort();
+ String transaction = node.getTransaction() == null
+ ? ""
+ : node.getTransaction().getHost()
+ + IP_PORT_SPLIT_CHAR
+ + node.getTransaction().getPort();
+
+ return control + "|" + transaction + "|" + node.getRole() + "|" +
node.getVersion() + "|" + node.getGroup();
+ }
+
+ private static synchronized void closeHttp2Watch() {
+ SeataHttpWatch<ClusterWatchEvent> watch = HTTP2_WATCH;
+ HTTP2_WATCH = null;
+ HTTP2_WATCH_GROUP = null;
+ if (watch != null) {
+ try {
+ watch.close();
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close HTTP2 watch stream", e);
+ }
+ }
+ }
+
private static String queryHttpAddress(String clusterName, String group) {
List<Node> nodeList = METADATA.getNodes(clusterName, group);
List<String> addressList = null;
@@ -232,7 +546,11 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
stream = inetSocketAddresses.stream();
}
} else {
- stream = INIT_ADDRESSES.get(clusterName).stream();
+ List<InetSocketAddress> initAddresses =
INIT_ADDRESSES.get(clusterName);
+ if (CollectionUtils.isEmpty(initAddresses)) {
+ return null;
+ }
+ stream = initAddresses.stream();
}
if (addressList != null) {
return
addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
@@ -256,7 +574,9 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
+ (controlEndpoint != null ?
controlEndpoint.getPort() : inetSocketAddress.getPort());
})
.collect(Collectors.toList());
- return
addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
+ return addressList.isEmpty()
+ ? null
+ :
addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
}
}
@@ -325,7 +645,6 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
private static InetSocketAddress selectEndpoint(String type, Node node) {
if (StringUtils.isBlank(PREFERRED_NETWORKS)) {
- // Use the default method, directly using node.control and
node.transaction
switch (type) {
case "control":
return new InetSocketAddress(
@@ -393,6 +712,7 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
@Override
public void close() {
CLOSED.compareAndSet(false, true);
+ closeHttp2Watch();
}
@Override
@@ -407,44 +727,6 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
return RegistryService.super.aliveLookup(transactionServiceGroup);
}
- private static boolean watch() throws RetryableException {
- Map<String, String> header = new HashMap<>();
- header.put(HTTP.CONTENT_TYPE,
ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
- Map<String, String> param = new HashMap<>();
- String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
- Map<String, Long> groupTerms = METADATA.getClusterTerm(clusterName);
- groupTerms.forEach((k, v) -> param.put(k, String.valueOf(v)));
- for (String group : groupTerms.keySet()) {
- String tcAddress = queryHttpAddress(clusterName, group);
- if (isTokenExpired()) {
- refreshToken(tcAddress);
- }
- if (StringUtils.isNotBlank(jwtToken)) {
- header.put(AUTHORIZATION_HEADER, jwtToken);
- }
- try (Response response =
- HttpClientUtil.doPost("http://"; + tcAddress +
"/metadata/v1/watch", param, header, 30000)) {
- if (response != null) {
- int statusCode = response.code();
- if (statusCode == HttpStatus.SC_UNAUTHORIZED) {
- if (StringUtils.isNotBlank(USERNAME) &&
StringUtils.isNotBlank(PASSWORD)) {
- throw new RetryableException("Authentication
failed!");
- } else {
- throw new AuthenticationFailedException(
- "Authentication failed! you should
configure the correct username and password.");
- }
- }
- return statusCode == HttpStatus.SC_OK;
- }
- } catch (IOException e) {
- LOGGER.error("watch cluster node: {}, fail: {}", tcAddress,
e.getMessage());
- throw new RetryableException(e.getMessage(), e);
- }
- break;
- }
- return false;
- }
-
@Override
public List<InetSocketAddress> refreshAliveLookup(
String transactionServiceGroup, List<InetSocketAddress>
aliveAddress) {
@@ -457,8 +739,6 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
? aliveAddress
: aliveAddress.parallelStream()
.filter(inetSocketAddress -> {
- // Since only follower will turn into
leader, only the follower node needs to be
- // listened to
return inetSocketAddress.getPort() !=
leaderAddress.getPort()
|| !inetSocketAddress
.getAddress()
@@ -518,10 +798,17 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
"Authentication failed! you should configure
the correct username and password.");
}
}
- MetadataResponse metadataResponse;
if (StringUtils.isNotBlank(response)) {
try {
- metadataResponse = OBJECT_MAPPER.readValue(response,
MetadataResponse.class);
+ MetadataResponse metadataResponse =
OBJECT_MAPPER.readValue(response, MetadataResponse.class);
+ if
(CollectionUtils.isEmpty(metadataResponse.getNodes())) {
+ LOGGER.warn(
+ "empty metadata nodes from cluster
endpoint, clusterName={}, group={}, response={}",
+ clusterName,
+ group,
+ response);
+ return;
+ }
METADATA.refreshMetadata(clusterName,
metadataResponse);
} catch (JsonProcessingException e) {
LOGGER.error(e.getMessage(), e);
@@ -534,11 +821,9 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
}
private static void refreshToken(String tcAddress) throws
RetryableException {
- // if username and password is not in config , return
if (StringUtils.isBlank(USERNAME) || StringUtils.isBlank(PASSWORD)) {
return;
}
- // get token and set it in cache
Map<String, String> param = new HashMap<>();
param.put(PRO_USERNAME_KEY, USERNAME);
param.put(PRO_PASSWORD_KEY, PASSWORD);
@@ -554,7 +839,6 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
JsonNode jsonNode = OBJECT_MAPPER.readTree(response);
String codeStatus = jsonNode.get("code").asText();
if (!StringUtils.equals(codeStatus, "200")) {
- // authorized failed,throw exception to kill
process
throw new AuthenticationFailedException(
"Authentication failed! you should
configure the correct username and password.");
}
@@ -564,7 +848,6 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
throw new
AuthenticationFailedException("Authentication failed! Response body is null.");
}
} else {
- // authorized failed,throw exception to kill process
throw new AuthenticationFailedException(
"Authentication failed! you should configure the
correct username and password.");
}
@@ -574,6 +857,30 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
}
}
+ private static boolean supportsHttp2(Node node) {
+ if (node == null) {
+ return false;
+ }
+ String version = node.getVersion();
+ if (StringUtils.isBlank(version)) {
+ return false;
+ }
+ try {
+ return Version.isAboveOrEqualVersion(version, MIN_HTTP2_VERSION);
+ } catch (Exception e) {
+ LOGGER.warn("Invalid version: {}, fallback to HTTP/1.1", version);
+ return false;
+ }
+ }
+
+ private static boolean isClusterHttp2Enabled(String clusterName, String
group) {
+ List<Node> nodes = METADATA.getNodes(clusterName, group);
+ if (CollectionUtils.isEmpty(nodes)) {
+ return false;
+ }
+ return nodes.stream().allMatch(RaftRegistryServiceImpl::supportsHttp2);
+ }
+
@Override
public List<InetSocketAddress> lookup(String key) throws Exception {
String clusterName = getServiceGroup(key);
@@ -597,13 +904,11 @@ public class RaftRegistryServiceImpl implements
RegistryService<ConfigChangeList
return null;
}
INIT_ADDRESSES.put(clusterName, list);
- // init jwt token
try {
refreshToken(queryHttpAddress(clusterName, key));
} catch (Exception e) {
throw new RuntimeException("Init fetch token failed!", e);
}
- // Refresh the metadata by initializing the address
acquireClusterMetaDataByClusterName(clusterName);
startQueryMetadata();
}
diff --git
a/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
b/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
index a85619da05..4473e1f481 100644
---
a/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
+++
b/discovery/seata-discovery-raft/src/test/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImplTest.java
@@ -25,10 +25,14 @@ import okhttp3.ResponseBody;
import org.apache.http.HttpStatus;
import org.apache.seata.common.exception.NotSupportYetException;
import org.apache.seata.common.exception.ParseEndpointException;
+import org.apache.seata.common.exception.RetryableException;
+import org.apache.seata.common.metadata.ClusterRole;
+import org.apache.seata.common.metadata.ClusterWatchEvent;
import org.apache.seata.common.metadata.Metadata;
import org.apache.seata.common.metadata.MetadataResponse;
import org.apache.seata.common.metadata.Node;
import org.apache.seata.common.util.HttpClientUtil;
+import org.apache.seata.common.util.SeataHttpWatch;
import org.apache.seata.config.ConfigurationFactory;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -44,6 +48,7 @@ import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -63,7 +68,10 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class RaftRegistryServiceImplTest {
@@ -91,12 +99,18 @@ class RaftRegistryServiceImplTest {
}
@AfterEach
- public void tearDown() throws NoSuchFieldException, IllegalAccessException
{
+ public void tearDown() throws Exception {
// Reset the CLOSED flag after each test
Field closedField =
RaftRegistryServiceImpl.class.getDeclaredField("CLOSED");
closedField.setAccessible(true);
AtomicBoolean closed = (AtomicBoolean) closedField.get(null);
closed.set(false);
+
+ Method closeHttp2WatchMethod =
RaftRegistryServiceImpl.class.getDeclaredMethod("closeHttp2Watch");
+ closeHttp2WatchMethod.setAccessible(true);
+ closeHttp2WatchMethod.invoke(null);
+
+ setStaticField("HTTP2_WATCH_GROUP", null);
}
/**
@@ -955,87 +969,22 @@ class RaftRegistryServiceImplTest {
}
/**
- * Test watch method with null response
+ * Test watch method with null response.
+ * Note: this also covers the previous "null status line" scenario because
+ * both cases are represented as null okhttp response in current code path.
*/
@Test
public void watchWithNullResponseTest() throws Exception {
- String jsonString =
-
"{\"nodes\":[{\"control\":{\"host\":\"localhost\",\"port\":7091},\"transaction\":{\"host\":\"localhost\",\"port\":8091},\"group\":\"default\",\"role\":\"LEADER\"}],\"storeMode\":\"raft\",\"term\":1}";
-
- Field metadataField =
RaftRegistryServiceImpl.class.getDeclaredField("METADATA");
- metadataField.setAccessible(true);
- Metadata metadata = (Metadata) metadataField.get(null);
-
- ObjectMapper objectMapper = new ObjectMapper();
- MetadataResponse metadataResponse = objectMapper.readValue(jsonString,
MetadataResponse.class);
- metadata.refreshMetadata("default", metadataResponse);
-
- Field clusterNameField =
RaftRegistryServiceImpl.class.getDeclaredField("CURRENT_TRANSACTION_CLUSTER_NAME");
- clusterNameField.setAccessible(true);
- clusterNameField.set(null, "default");
-
- Field initAddressesField =
RaftRegistryServiceImpl.class.getDeclaredField("INIT_ADDRESSES");
- initAddressesField.setAccessible(true);
- Map<String, List<InetSocketAddress>> initAddresses =
- (Map<String, List<InetSocketAddress>>)
initAddressesField.get(null);
- List<InetSocketAddress> addressList = new ArrayList<>();
- addressList.add(new InetSocketAddress("localhost", 7091));
- initAddresses.put("default", addressList);
+ prepareWatchClusterContext(
+ "default",
+ metadataResponse(1L, createNode("localhost", 7091, 8091,
"default", ClusterRole.LEADER, "2.6.9")));
try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
when(HttpClientUtil.doPost(anyString(), anyMap(), anyMap(),
anyInt()))
.thenReturn(null);
-
- Method watchMethod =
RaftRegistryServiceImpl.class.getDeclaredMethod("watch");
- watchMethod.setAccessible(true);
- boolean result = (boolean) watchMethod.invoke(null);
+ boolean result = invokeWatch();
assertFalse(result, "Watch should return false when response is
null");
- } finally {
- initAddresses.remove("default");
- }
- }
-
- /**
- * Test watch method with null status line
- */
- @Test
- public void watchWithNullStatusLineTest() throws Exception {
- String jsonString =
-
"{\"nodes\":[{\"control\":{\"host\":\"localhost\",\"port\":7091},\"transaction\":{\"host\":\"localhost\",\"port\":8091},\"group\":\"default\",\"role\":\"LEADER\"}],\"storeMode\":\"raft\",\"term\":1}";
-
- Field metadataField =
RaftRegistryServiceImpl.class.getDeclaredField("METADATA");
- metadataField.setAccessible(true);
- Metadata metadata = (Metadata) metadataField.get(null);
-
- ObjectMapper objectMapper = new ObjectMapper();
- MetadataResponse metadataResponse = objectMapper.readValue(jsonString,
MetadataResponse.class);
- metadata.refreshMetadata("default", metadataResponse);
-
- Field clusterNameField =
RaftRegistryServiceImpl.class.getDeclaredField("CURRENT_TRANSACTION_CLUSTER_NAME");
- clusterNameField.setAccessible(true);
- clusterNameField.set(null, "default");
-
- Field initAddressesField =
RaftRegistryServiceImpl.class.getDeclaredField("INIT_ADDRESSES");
- initAddressesField.setAccessible(true);
- Map<String, List<InetSocketAddress>> initAddresses =
- (Map<String, List<InetSocketAddress>>)
initAddressesField.get(null);
- List<InetSocketAddress> addressList = new ArrayList<>();
- addressList.add(new InetSocketAddress("localhost", 7091));
- initAddresses.put("default", addressList);
-
- try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
- // Return null to simulate null response scenario
- when(HttpClientUtil.doPost(anyString(), anyMap(), anyMap(),
anyInt()))
- .thenReturn(null);
-
- Method watchMethod =
RaftRegistryServiceImpl.class.getDeclaredMethod("watch");
- watchMethod.setAccessible(true);
- boolean result = (boolean) watchMethod.invoke(null);
-
- assertFalse(result, "Watch should return false when response is
null");
- } finally {
- initAddresses.remove("default");
}
}
@@ -1214,6 +1163,177 @@ class RaftRegistryServiceImplTest {
}
}
+ @Test
+ public void watchHttp2CreateAndConsumeUpdateTest() throws Exception {
+ String clusterName = "watchHttp2CreateCluster";
+ String group = "default";
+ prepareWatchClusterContext(
+ clusterName,
+ metadataResponse(1L, createNode("127.0.0.1", 7091, 8091,
group, ClusterRole.LEADER, "2.7.0")));
+ setTokenNotExpired();
+
+ SeataHttpWatch<ClusterWatchEvent> watch = mockWatch();
+ when(watch.next())
+ .thenReturn(newWatchUpdateResponse(
+ group,
+ metadataResponse(2L, createNode("127.0.0.1", 7091,
8091, group, ClusterRole.LEADER, "2.7.0"))));
+
+ try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
+ mockedStatic
+ .when(() -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()))
+ .thenReturn(watch);
+
+ boolean result = invokeWatchHttp2(clusterName);
+ assertTrue(result, "watchHttp2 should return true when UPDATE term
advances");
+
+ Metadata metadata = (Metadata) getStaticField("METADATA");
+ assertEquals(2L,
metadata.getClusterTerm(clusterName).get(group).longValue());
+ mockedStatic.verify(
+ () -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()),
+ times(1));
+ }
+ }
+
+ @Test
+ public void watchHttp2ReuseForSameGroupTest() throws Exception {
+ String clusterName = "watchHttp2ReuseCluster";
+ String group = "default";
+ prepareWatchClusterContext(
+ clusterName,
+ metadataResponse(1L, createNode("127.0.0.1", 7191, 8191,
group, ClusterRole.LEADER, "2.7.0")));
+ setTokenNotExpired();
+
+ SeataHttpWatch<ClusterWatchEvent> watch = mockWatch();
+ when(watch.next())
+ .thenReturn(newWatchUpdateResponse(
+ group,
+ metadataResponse(2L, createNode("127.0.0.1", 7191,
8191, group, ClusterRole.LEADER, "2.7.0"))))
+ .thenReturn(newWatchUpdateResponse(
+ group,
+ metadataResponse(3L, createNode("127.0.0.1", 7191,
8191, group, ClusterRole.LEADER, "2.7.0"))));
+
+ try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
+ mockedStatic
+ .when(() -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()))
+ .thenReturn(watch);
+
+ assertTrue(invokeWatchHttp2(clusterName));
+ assertTrue(invokeWatchHttp2(clusterName));
+
+ mockedStatic.verify(
+ () -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()),
+ times(1));
+ }
+ }
+
+ @Test
+ public void watchHttp2SwitchGroupStableSelectionTest() throws Exception {
+ String clusterName = "watchHttp2MultiGroupCluster";
+ Node group2Leader = createNode("127.0.0.1", 7092, 8092, "g2",
ClusterRole.LEADER, "2.7.0");
+ Node group1Leader = createNode("127.0.0.1", 7091, 8091, "g1",
ClusterRole.LEADER, "2.7.0");
+ prepareWatchClusterContext(clusterName, metadataResponse(1L,
group2Leader), metadataResponse(1L, group1Leader));
+ setTokenNotExpired();
+ setStaticField("HTTP2_WATCH_GROUP", "missing-group");
+
+ SeataHttpWatch<ClusterWatchEvent> watch = mockWatch();
+ when(watch.next())
+ .thenReturn(newWatchUpdateResponse("g1", metadataResponse(2L,
group1Leader)))
+ .thenReturn(newWatchUpdateResponse("g1", metadataResponse(3L,
group1Leader)));
+
+ try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
+ mockedStatic
+ .when(() -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()))
+ .thenReturn(watch);
+
+ assertTrue(invokeWatchHttp2(clusterName));
+ assertTrue(invokeWatchHttp2(clusterName));
+
+ mockedStatic.verify(
+ () -> HttpClientUtil.watchPost(
+ eq("http://127.0.0.1:7091/metadata/v1/watch";),
+ anyMap(),
+ anyMap(),
+ eq(ClusterWatchEvent.class),
+ anyInt()),
+ times(1));
+ }
+ }
+
+ @Test
+ public void watchHttp2RuntimeExceptionCloseAndRetryableTest() throws
Exception {
+ String clusterName = "watchHttp2ExceptionCluster";
+ String group = "default";
+ prepareWatchClusterContext(
+ clusterName,
+ metadataResponse(1L, createNode("127.0.0.1", 7391, 8391,
group, ClusterRole.LEADER, "2.7.0")));
+ setTokenNotExpired();
+
+ SeataHttpWatch<ClusterWatchEvent> watch = mockWatch();
+ when(watch.next()).thenThrow(new RuntimeException("watch stream
failure"));
+
+ try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
+ mockedStatic
+ .when(() -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()))
+ .thenReturn(watch);
+
+ assertThrows(RetryableException.class, () ->
invokeWatchHttp2(clusterName));
+ verify(watch, times(1)).close();
+ assertNull(getStaticField("HTTP2_WATCH"));
+ assertNull(getStaticField("HTTP2_WATCH_GROUP"));
+ }
+ }
+
+ @Test
+ public void watchHttp2ClosedStateNoRetryNoiseTest() throws Exception {
+ String clusterName = "watchHttp2ClosedCluster";
+ String group = "default";
+ prepareWatchClusterContext(
+ clusterName,
+ metadataResponse(1L, createNode("127.0.0.1", 7491, 8491,
group, ClusterRole.LEADER, "2.7.0")));
+ setTokenNotExpired();
+ setClosed(true);
+
+ SeataHttpWatch<ClusterWatchEvent> watch = mockWatch();
+ when(watch.next()).thenThrow(new RuntimeException("closed during
shutdown"));
+
+ try (MockedStatic<HttpClientUtil> mockedStatic =
Mockito.mockStatic(HttpClientUtil.class)) {
+ mockedStatic
+ .when(() -> HttpClientUtil.watchPost(
+ anyString(), anyMap(), anyMap(),
eq(ClusterWatchEvent.class), anyInt()))
+ .thenReturn(watch);
+
+ boolean result = invokeWatchHttp2(clusterName);
+ assertFalse(result, "watchHttp2 should return false instead of
retrying when client is closed");
+ verify(watch, times(1)).close();
+ } finally {
+ setClosed(false);
+ }
+ }
+
+ @Test
+ public void selectWatchGroupReuseAndSortedFallbackTest() throws Exception {
+ Method selectWatchGroupMethod =
RaftRegistryServiceImpl.class.getDeclaredMethod("selectWatchGroup", Map.class);
+ selectWatchGroupMethod.setAccessible(true);
+
+ Map<String, Long> groupTerms = new HashMap<>();
+ groupTerms.put("g2", 1L);
+ groupTerms.put("g1", 1L);
+
+ setStaticField("HTTP2_WATCH_GROUP", "g2");
+ assertEquals("g2", selectWatchGroupMethod.invoke(null, groupTerms));
+
+ setStaticField("HTTP2_WATCH_GROUP", "unknown");
+ assertEquals("g1", selectWatchGroupMethod.invoke(null, groupTerms));
+
+ assertNull(selectWatchGroupMethod.invoke(null,
Collections.emptyMap()));
+ }
+
/**
* Test lookup with empty raft cluster address
*/
@@ -1287,6 +1407,304 @@ class RaftRegistryServiceImplTest {
}
}
+ @Test
+ public void supportsHttp2VersionThresholdTest() throws Exception {
+ Method supportsHttp2Method =
RaftRegistryServiceImpl.class.getDeclaredMethod("supportsHttp2", Node.class);
+ supportsHttp2Method.setAccessible(true);
+
+ Node nullVersion = new Node();
+ nullVersion.setVersion(null);
+ assertFalse((boolean) supportsHttp2Method.invoke(null, nullVersion));
+
+ Node blankVersion = new Node();
+ blankVersion.setVersion("");
+ assertFalse((boolean) supportsHttp2Method.invoke(null, blankVersion));
+
+ Node invalidVersion = new Node();
+ invalidVersion.setVersion("invalid-version");
+ assertFalse((boolean) supportsHttp2Method.invoke(null,
invalidVersion));
+
+ Node v269 = new Node();
+ v269.setVersion("2.6.9");
+ assertFalse((boolean) supportsHttp2Method.invoke(null, v269));
+
+ Node v270 = new Node();
+ v270.setVersion("2.7.0");
+ assertTrue((boolean) supportsHttp2Method.invoke(null, v270));
+
+ Node v270Snapshot = new Node();
+ v270Snapshot.setVersion("2.7.0-SNAPSHOT");
+ assertTrue((boolean) supportsHttp2Method.invoke(null, v270Snapshot));
+
+ Node v280 = new Node();
+ v280.setVersion("2.8.0");
+ assertTrue((boolean) supportsHttp2Method.invoke(null, v280));
+ }
+
+ @Test
+ public void shouldRefreshMetadataFilterInvalidEventTest() throws Exception
{
+ Method shouldRefreshMetadata =
RaftRegistryServiceImpl.class.getDeclaredMethod(
+ "shouldRefreshMetadata", String.class, String.class,
SeataHttpWatch.Response.class);
+ shouldRefreshMetadata.setAccessible(true);
+
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
"invalidEventCluster", "default", null));
+
+ SeataHttpWatch.Response<ClusterWatchEvent> errorResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.ERROR, null);
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
"invalidEventCluster", "default", errorResponse));
+
+ SeataHttpWatch.Response<ClusterWatchEvent> nullObjectUpdate =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE, null);
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
"invalidEventCluster", "default", nullObjectUpdate));
+
+ ClusterWatchEvent nullMetadataEvent = new ClusterWatchEvent();
+ SeataHttpWatch.Response<ClusterWatchEvent> nullMetadataResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE,
nullMetadataEvent);
+ assertFalse(
+ (boolean) shouldRefreshMetadata.invoke(null,
"invalidEventCluster", "default", nullMetadataResponse));
+
+ MetadataResponse emptyNodesMetadata = new MetadataResponse();
+ emptyNodesMetadata.setNodes(Collections.emptyList());
+ emptyNodesMetadata.setStoreMode("raft");
+ emptyNodesMetadata.setTerm(1L);
+ ClusterWatchEvent emptyNodesEvent = new ClusterWatchEvent();
+ emptyNodesEvent.setGroup("default");
+ emptyNodesEvent.setMetadata(emptyNodesMetadata);
+ SeataHttpWatch.Response<ClusterWatchEvent> emptyNodesResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE, emptyNodesEvent);
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
"invalidEventCluster", "default", emptyNodesResponse));
+ }
+
+ @Test
+ public void shouldRefreshMetadataOnTermOrNodeChangeTest() throws Exception
{
+ Field metadataField =
RaftRegistryServiceImpl.class.getDeclaredField("METADATA");
+ metadataField.setAccessible(true);
+ Metadata metadata = (Metadata) metadataField.get(null);
+
+ Method shouldRefreshMetadata =
RaftRegistryServiceImpl.class.getDeclaredMethod(
+ "shouldRefreshMetadata", String.class, String.class,
SeataHttpWatch.Response.class);
+ shouldRefreshMetadata.setAccessible(true);
+
+ String termAdvancedCluster = "termAdvancedCluster";
+ metadata.refreshMetadata(
+ termAdvancedCluster,
+ metadataResponse(1L, createNode("127.0.0.1", 7091, 8091,
"default", ClusterRole.LEADER, "2.7.0")));
+ ClusterWatchEvent termAdvancedEvent = new ClusterWatchEvent();
+ termAdvancedEvent.setGroup("default");
+ termAdvancedEvent.setMetadata(
+ metadataResponse(2L, createNode("127.0.0.1", 7091, 8091,
"default", ClusterRole.LEADER, "2.7.0")));
+ SeataHttpWatch.Response<ClusterWatchEvent> termAdvancedResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE,
termAdvancedEvent);
+ assertTrue((boolean) shouldRefreshMetadata.invoke(null,
termAdvancedCluster, "default", termAdvancedResponse));
+
+ String unchangedCluster = "unchangedCluster";
+ metadata.refreshMetadata(
+ unchangedCluster,
+ metadataResponse(3L, createNode("127.0.0.1", 7092, 8092,
"default", ClusterRole.LEADER, "2.7.0")));
+ ClusterWatchEvent unchangedEvent = new ClusterWatchEvent();
+ unchangedEvent.setGroup("default");
+ unchangedEvent.setMetadata(
+ metadataResponse(3L, createNode("127.0.0.1", 7092, 8092,
"default", ClusterRole.LEADER, "2.7.0")));
+ SeataHttpWatch.Response<ClusterWatchEvent> unchangedResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE, unchangedEvent);
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
unchangedCluster, "default", unchangedResponse));
+
+ String nodeChangedCluster = "nodeChangedCluster";
+ metadata.refreshMetadata(
+ nodeChangedCluster,
+ metadataResponse(5L, createNode("127.0.0.1", 7093, 8093,
"default", ClusterRole.LEADER, "2.7.0")));
+ ClusterWatchEvent nodeChangedEvent = new ClusterWatchEvent();
+ nodeChangedEvent.setGroup("default");
+ nodeChangedEvent.setMetadata(
+ metadataResponse(5L, createNode("127.0.0.1", 7193, 8193,
"default", ClusterRole.LEADER, "2.7.0")));
+ SeataHttpWatch.Response<ClusterWatchEvent> nodeChangedResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE,
nodeChangedEvent);
+ assertTrue((boolean) shouldRefreshMetadata.invoke(null,
nodeChangedCluster, "default", nodeChangedResponse));
+
+ String staleTermCluster = "staleTermCluster";
+ Node currentLeader = createNode("127.0.0.1", 7094, 8094, "default",
ClusterRole.LEADER, "2.7.0");
+ metadata.refreshMetadata(staleTermCluster, metadataResponse(10L,
currentLeader));
+ ClusterWatchEvent staleTermEvent = new ClusterWatchEvent();
+ staleTermEvent.setGroup("default");
+ staleTermEvent.setMetadata(
+ metadataResponse(9L, createNode("127.0.0.1", 7194, 8194,
"default", ClusterRole.FOLLOWER, "2.7.0")));
+ SeataHttpWatch.Response<ClusterWatchEvent> staleTermResponse =
+ new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE, staleTermEvent);
+
+ assertFalse((boolean) shouldRefreshMetadata.invoke(null,
staleTermCluster, "default", staleTermResponse));
+ assertEquals(
+ 10L,
metadata.getClusterTerm(staleTermCluster).get("default").longValue());
+ assertEquals(
+ currentLeader.getTransaction().getPort(),
+ metadata.getNodes(staleTermCluster, "default")
+ .get(0)
+ .getTransaction()
+ .getPort());
+ }
+
+ @Test
+ public void hasMetadataChangedOrderInsensitiveTest() throws Exception {
+ Field metadataField =
RaftRegistryServiceImpl.class.getDeclaredField("METADATA");
+ metadataField.setAccessible(true);
+ Metadata metadata = (Metadata) metadataField.get(null);
+
+ Method hasMetadataChanged =
RaftRegistryServiceImpl.class.getDeclaredMethod(
+ "hasMetadataChanged", String.class, String.class,
MetadataResponse.class);
+ hasMetadataChanged.setAccessible(true);
+
+ String clusterName = "orderInsensitiveCluster";
+ Node first = createNode("127.0.0.1", 7291, 8291, "default",
ClusterRole.LEADER, "2.7.0");
+ Node second = createNode("127.0.0.1", 7292, 8292, "default",
ClusterRole.FOLLOWER, "2.7.0");
+ metadata.refreshMetadata(clusterName, metadataResponse(10L, first,
second));
+
+ MetadataResponse reversedOrder = metadataResponse(10L, second, first);
+ assertFalse((boolean) hasMetadataChanged.invoke(null, clusterName,
"default", reversedOrder));
+
+ Node roleChanged = createNode("127.0.0.1", 7292, 8292, "default",
ClusterRole.LEADER, "2.7.0");
+ MetadataResponse changedNodeSignature = metadataResponse(10L, first,
roleChanged);
+ assertTrue((boolean) hasMetadataChanged.invoke(null, clusterName,
"default", changedNodeSignature));
+ }
+
+ @Test
+ public void hasMetadataChangedEdgeCasesTest() throws Exception {
+ Metadata metadata = (Metadata) getStaticField("METADATA");
+
+ Method hasMetadataChanged =
RaftRegistryServiceImpl.class.getDeclaredMethod(
+ "hasMetadataChanged", String.class, String.class,
MetadataResponse.class);
+ hasMetadataChanged.setAccessible(true);
+
+ assertFalse((boolean) hasMetadataChanged.invoke(null,
"nullIncomingCluster", "default", null));
+
+ Node first = createNode("127.0.0.1", 7391, 8391, "default",
ClusterRole.LEADER, "2.7.0");
+ assertTrue(
+ (boolean) hasMetadataChanged.invoke(null, "emptyLocalCluster",
"default", metadataResponse(1L, first)));
+
+ assertFalse((boolean) hasMetadataChanged.invoke(null,
"emptyBothCluster", "default", metadataResponse(1L)));
+
+ String termChangedCluster = "termChangedCluster";
+ metadata.refreshMetadata(termChangedCluster, metadataResponse(1L,
first));
+ assertTrue(
+ (boolean) hasMetadataChanged.invoke(null, termChangedCluster,
"default", metadataResponse(2L, first)));
+
+ String nodeSizeChangedCluster = "nodeSizeChangedCluster";
+ metadata.refreshMetadata(nodeSizeChangedCluster, metadataResponse(3L,
first));
+ Node second = createNode("127.0.0.1", 7392, 8392, "default",
ClusterRole.FOLLOWER, "2.7.0");
+ assertTrue((boolean) hasMetadataChanged.invoke(
+ null, nodeSizeChangedCluster, "default", metadataResponse(3L,
first, second)));
+ }
+
+ @Test
+ public void buildNodeSignatureEdgeCasesTest() throws Exception {
+ Method buildNodeSignature =
RaftRegistryServiceImpl.class.getDeclaredMethod("buildNodeSignature",
Node.class);
+ buildNodeSignature.setAccessible(true);
+
+ assertEquals("", buildNodeSignature.invoke(null, new Object[] {null}));
+
+ Node node = new Node();
+ node.setGroup("default");
+ node.setRole(ClusterRole.LEADER);
+ node.setVersion("2.7.0");
+ assertEquals("||LEADER|2.7.0|default", buildNodeSignature.invoke(null,
node));
+ }
+
+ private static void prepareWatchClusterContext(String clusterName,
MetadataResponse... metadataResponses)
+ throws Exception {
+ Metadata metadata = (Metadata) getStaticField("METADATA");
+ for (MetadataResponse metadataResponse : metadataResponses) {
+ metadata.refreshMetadata(clusterName, metadataResponse);
+ }
+ setStaticField("CURRENT_TRANSACTION_CLUSTER_NAME", clusterName);
+ setStaticField("CURRENT_TRANSACTION_SERVICE_GROUP", "tx");
+
+ Map<String, List<InetSocketAddress>> aliveNodes =
+ (Map<String, List<InetSocketAddress>>)
getStaticField("ALIVE_NODES");
+ aliveNodes.remove("tx");
+ }
+
+ private static boolean invokeWatch() throws Exception {
+ Method watchMethod =
RaftRegistryServiceImpl.class.getDeclaredMethod("watch");
+ watchMethod.setAccessible(true);
+ try {
+ return (boolean) watchMethod.invoke(null);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ throw new RuntimeException(cause);
+ }
+ }
+
+ private static boolean invokeWatchHttp2(String clusterName) throws
Exception {
+ Method watchHttp2Method =
RaftRegistryServiceImpl.class.getDeclaredMethod("watchHttp2", String.class);
+ watchHttp2Method.setAccessible(true);
+ try {
+ return (boolean) watchHttp2Method.invoke(null, clusterName);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ throw new RuntimeException(cause);
+ }
+ }
+
+ private static SeataHttpWatch.Response<ClusterWatchEvent>
newWatchUpdateResponse(
+ String group, MetadataResponse metadataResponse) {
+ ClusterWatchEvent event = new ClusterWatchEvent();
+ event.setGroup(group);
+ event.setMetadata(metadataResponse);
+ return new
SeataHttpWatch.Response<>(SeataHttpWatch.Response.Type.UPDATE, event);
+ }
+
+ private static SeataHttpWatch<ClusterWatchEvent> mockWatch() {
+ return mock(SeataHttpWatch.class);
+ }
+
+ private static void setTokenNotExpired() throws Exception {
+ Field tokenTimeStamp =
RaftRegistryServiceImpl.class.getDeclaredField("tokenTimeStamp");
+ tokenTimeStamp.setAccessible(true);
+ tokenTimeStamp.setLong(RaftRegistryServiceImpl.class,
System.currentTimeMillis());
+ }
+
+ private static void setClosed(boolean value) throws Exception {
+ Field closedField =
RaftRegistryServiceImpl.class.getDeclaredField("CLOSED");
+ closedField.setAccessible(true);
+ AtomicBoolean closed = (AtomicBoolean) closedField.get(null);
+ closed.set(value);
+ }
+
+ private static void setStaticField(String fieldName, Object value) throws
Exception {
+ Field field =
RaftRegistryServiceImpl.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(null, value);
+ }
+
+ private static Object getStaticField(String fieldName) throws Exception {
+ Field field =
RaftRegistryServiceImpl.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field.get(null);
+ }
+
+ private static Node createNode(
+ String host, int controlPort, int transactionPort, String group,
ClusterRole role, String version) {
+ Node node = new Node();
+ node.setControl(new Node.Endpoint(host, controlPort));
+ node.setTransaction(new Node.Endpoint(host, transactionPort));
+ node.setGroup(group);
+ node.setRole(role);
+ node.setVersion(version);
+ return node;
+ }
+
+ private static MetadataResponse metadataResponse(long term, Node... nodes)
{
+ MetadataResponse metadataResponse = new MetadataResponse();
+ metadataResponse.setNodes(Arrays.asList(nodes));
+ metadataResponse.setStoreMode("raft");
+ metadataResponse.setTerm(term);
+ return metadataResponse;
+ }
+
/**
* Test startQueryMetadata creates thread pool
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
