Copilot commented on code in PR #7952:
URL: https://github.com/apache/incubator-seata/pull/7952#discussion_r2921764190
##########
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);
Review Comment:
`watchHttp2` calls `HTTP2_WATCH.next()` directly on the volatile field.
Since `closeHttp2Watch()` can be called concurrently (e.g., via `close()` or
the shutdown hook) and sets `HTTP2_WATCH` to null, this can race and throw a
`NullPointerException`. Capture `HTTP2_WATCH` into a local variable (or
synchronize around read/close) before calling `next()`, and handle the case
where it becomes null during shutdown.
##########
discovery/seata-discovery-raft/pom.xml:
##########
@@ -43,9 +48,14 @@
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>mockwebserver</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
Review Comment:
The `mockwebserver` test dependency appears unused in this module (no
references under `src/test/java`). If it's not needed for these tests, it
should be removed to avoid pulling in extra artifacts and keeping the module's
dependency set minimal.
```suggestion
```
##########
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;
+ }
Review Comment:
The newly introduced HTTP/2 watch logic isn’t directly exercised by unit
tests in this module (e.g., no test that stubs `HttpClientUtil.watchPost(...)`
and asserts watch creation/retry/close behavior). The added tests cover
`supportsHttp2` and `shouldRefreshMetadata`, but
`watchHttp2`/`ensureHttp2Watch` behavior itself remains unverified.
##########
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();
Review Comment:
`watchHttp2` picks an arbitrary group via
`groupTerms.keySet().iterator().next()`. With a `ConcurrentHashMap`, iteration
order is not guaranteed, so when multiple groups exist this can flip between
groups across iterations and cause unnecessary watch teardown/recreation
(because `ensureHttp2Watch` caches by `HTTP2_WATCH_GROUP`). Consider choosing a
deterministic group (e.g., sort keys and pick the first, or prefer the
configured/default group) or key the watch cache by address+groups instead of a
single group name.
```suggestion
List<String> groups = new ArrayList<>(groupTerms.keySet());
Collections.sort(groups);
String group = groups.get(0);
```
##########
discovery/seata-discovery-raft/src/main/java/org/apache/seata/discovery/registry/raft/RaftRegistryServiceImpl.java:
##########
@@ -64,8 +69,8 @@
/**
* The type File registry service.
- *
*/
+@SuppressWarnings("ALL")
Review Comment:
`@SuppressWarnings("ALL")` on the whole class suppresses all compiler
warnings (including ones that could indicate real problems) and makes future
maintenance harder. Prefer removing it and, if needed, suppress only the
specific warning(s) at the narrowest scope (e.g., on the single unchecked cast
in `selectExternalEndpoint`).
```suggestion
```
--
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]
