YongGoose commented on code in PR #7586:
URL: https://github.com/apache/incubator-seata/pull/7586#discussion_r2278983328
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -221,46 +245,61 @@ protected static void startQueryMetadata() {
}
}
- private static String queryHttpAddress(String clusterName, String group) {
+ // 그러면 여기에서 string을 반환하는 것이 아니라 node를 반환해야 된다. (node를 반환해서 node의 버전에 따라 결정)
+ // 또는 2.0이 가능한 node를 먼저 선택하게 하는 거는 어떨까?
+ private static Node selectNodeForHttpAddress(String clusterName, String
group) {
List<Node> nodeList = METADATA.getNodes(clusterName, group);
- List<String> addressList = null;
- Stream<InetSocketAddress> stream = null;
+
if (CollectionUtils.isNotEmpty(nodeList)) {
List<InetSocketAddress> inetSocketAddresses =
ALIVE_NODES.get(CURRENT_TRANSACTION_SERVICE_GROUP);
+
if (CollectionUtils.isEmpty(inetSocketAddresses)) {
- addressList = nodeList.stream()
- .map(RaftRegistryServiceImpl::selectControlEndpointStr)
- .collect(Collectors.toList());
- } else {
- stream = inetSocketAddresses.stream();
+ return
nodeList.get(ThreadLocalRandom.current().nextInt(nodeList.size()));
}
- } else {
- stream = INIT_ADDRESSES.get(clusterName).stream();
- }
- if (addressList != null) {
- return
addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
- } else {
+
Map<String, Node> map = new HashMap<>();
- if (CollectionUtils.isNotEmpty(nodeList)) {
- for (Node node : nodeList) {
- InetSocketAddress inetSocketAddress =
selectTransactionEndpoint(node);
- map.put(inetSocketAddress.getHostString() +
IP_PORT_SPLIT_CHAR + inetSocketAddress.getPort(), node);
- }
+ for (Node node : nodeList) {
+ InetSocketAddress inetSocketAddress =
selectTransactionEndpoint(node);
+ map.put(inetSocketAddress.getHostString() + IP_PORT_SPLIT_CHAR
+ inetSocketAddress.getPort(), node);
}
- addressList = stream.map(inetSocketAddress -> {
- String host = NetUtil.toStringHost(inetSocketAddress);
- Node node = map.get(host + IP_PORT_SPLIT_CHAR +
inetSocketAddress.getPort());
- InetSocketAddress controlEndpoint = null;
- if (node != null) {
- controlEndpoint = selectControlEndpoint(node);
- }
- return host
- + IP_PORT_SPLIT_CHAR
- + (controlEndpoint != null ?
controlEndpoint.getPort() : inetSocketAddress.getPort());
- })
+
+ List<Node> aliveNodes = inetSocketAddresses.stream()
+ .map(addr -> map.get(NetUtil.toStringHost(addr) +
IP_PORT_SPLIT_CHAR + addr.getPort()))
+ .filter(Objects::nonNull)
.collect(Collectors.toList());
- return
addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
+
+ if (!aliveNodes.isEmpty()) {
+ return
aliveNodes.get(ThreadLocalRandom.current().nextInt(aliveNodes.size()));
+ }
+ } else {
+ List<InetSocketAddress> initAddresses =
INIT_ADDRESSES.get(clusterName);
+ if (CollectionUtils.isNotEmpty(initAddresses)) {
+
+ return null;
+ }
+ }
+ return null;
+ }
+
+ private static String queryHttpAddress(String clusterName, Node
selectedNode) {
Review Comment:
This method extracts the HTTP address based on the selected node and the
cluster name.
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -221,46 +245,61 @@ protected static void startQueryMetadata() {
}
}
- private static String queryHttpAddress(String clusterName, String group) {
+ // 그러면 여기에서 string을 반환하는 것이 아니라 node를 반환해야 된다. (node를 반환해서 node의 버전에 따라 결정)
+ // 또는 2.0이 가능한 node를 먼저 선택하게 하는 거는 어떨까?
+ private static Node selectNodeForHttpAddress(String clusterName, String
group) {
Review Comment:
This method is designed to select the node.
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -418,33 +457,48 @@ private static boolean watch() throws RetryableException {
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);
+ Node selectedNode = selectNodeForHttpAddress(clusterName, group);
+ String tcAddress = queryHttpAddress(clusterName, selectedNode);
+
if (isTokenExpired()) {
- refreshToken(tcAddress);
+ refreshToken(clusterName, selectedNode);
}
if (StringUtils.isNotBlank(jwtToken)) {
header.put(AUTHORIZATION_HEADER, jwtToken);
}
- try (CloseableHttpResponse response =
- HttpClientUtil.doPost("http://"; + tcAddress +
"/metadata/v1/watch", param, header, 30000)) {
- if (response != null) {
- StatusLine statusLine = response.getStatusLine();
- if (statusLine != null && statusLine.getStatusCode() ==
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 executeWatchRequest(selectedNode, tcAddress, param, header);
+ }
+ return false;
+ }
+
+ private static boolean executeWatchRequest(
+ Node selectedNode, String tcAddress, Map<String, String> param,
Map<String, String> header)
+ throws RetryableException {
+ if (selectedNode.isHttp2Supported()) {
+ Http2ClientUtil.doPost("http://"; + tcAddress +
"/metadata/v1/watch", param, header, CALLBACK);
+ return true; // TODO : Handle the response properly
+ }
Review Comment:
So that HTTP/2 is used based on the node’s version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
