funky-eyes commented on code in PR #7952:
URL: https://github.com/apache/incubator-seata/pull/7952#discussion_r2937811019
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -203,21 +233,302 @@ protected static void startQueryMetadata() {
} 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, 0L);
+ 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, 0L)) {
Review Comment:
```suggestion
if (incomingMetadata.getTerm() >
METADATA.getClusterTerm(clusterName).getOrDefault(group, -1L)) {
```
--
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]
