Tsukilc commented on code in PR #7952:
URL: https://github.com/apache/incubator-seata/pull/7952#discussion_r2901893705
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -203,21 +234,277 @@ 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) {
Review Comment:
Switch when the target protocol differs from the current protocol, enabling
rolling updates.
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -203,21 +234,277 @@ 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 = groupTerms.keySet().iterator().next();
+ 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);
+
+ try {
+ SeataHttpWatch.Response<ClusterWatchEvent> response =
HTTP2_WATCH.next();
+ return shouldRefreshMetadata(clusterName, group, response);
+ } catch (RuntimeException e) {
+ closeHttp2Watch();
+ throw new RetryableException("HTTP2 watch failed", e);
+ }
+ }
+
+ 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) {
Review Comment:
Used to determine whether there are changes in the node metadata.to ensure
don't refresh if there are no changes.
--
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]
