Copilot commented on code in PR #7952:
URL: https://github.com/apache/incubator-seata/pull/7952#discussion_r2944587541
##########
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;
Review Comment:
`HTTP2_WATCH_READ_TIMEOUT_SECONDS` is set to 300s and passed into
`HttpClientUtil.watchPost(...)`. In `startQueryMetadata`, the watch call runs
on the same loop that enforces `metadataMaxAgeMs` refresh, so a long
`watch.next()` block can prevent the periodic refresh from running for up to 5
minutes. Consider tying the HTTP/2 watch read timeout to `WATCH_TIMEOUT_MS` /
`metadataMaxAgeMs`, or decoupling the periodic refresh timer from the blocking
watch read so metadata refresh deadlines are still honored.
##########
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>
Review Comment:
This module adds a dependency on `seata-core` solely to use
`org.apache.seata.core.protocol.Version` for version comparison. That increases
coupling and pulls in Netty-related classes transitively, which is heavy for
the discovery module and makes layering harder to maintain. Consider moving the
version comparison helper to `seata-common` (or duplicating a minimal
comparator locally) so `seata-discovery-raft` doesn’t need to depend on
`seata-core`.
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -574,6 +854,30 @@ private static void refreshToken(String tcAddress) throws
RetryableException {
}
}
+ 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;
+ }
Review Comment:
`supportsHttp2` wraps `Version.isAboveOrEqualVersion(...)` in a try/catch,
but `Version.isAboveOrEqualVersion` already catches conversion errors
internally and returns `false` (logging at ERROR). As written, this catch block
will effectively never run for invalid versions, and the WARN log here is
misleading/unreachable. Consider removing the try/catch, or switching to a
version utility that throws so the warning path is meaningful (and ideally
avoid double-logging on invalid versions).
##########
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)) {
+ return true;
+ }
Review Comment:
`hasMetadataChanged` treats any term mismatch as a change (`incomingTerm !=
localTerm`), and `shouldRefreshMetadata` will then refresh metadata even when
the incoming term is *lower* than the local term. That can regress
`clusterTerm` and leader selection if an out-of-order/stale watch event is
received. Consider ignoring events with `incomingTerm < localTerm` (and only
treating term mismatch as change when `incomingTerm > localTerm`) to keep terms
monotonic.
--
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]
