Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

optimize: optimize the retry logic in the acquireMetadata method #6098

Merged
merged 3 commits into from
Dec 1, 2023
Merged

optimize: optimize the retry logic in the acquireMetadata method #6098

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fc518d1
optimize the retry logic in the acquireMetadata method
ggbocoder Nov 30, 2023
1ad9171
add signature
ggbocoder Nov 30, 2023
98f5538
Update discovery/seata-discovery-raft/src/main/java/io/seata/discover…
ggbocoder Nov 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changes/en-us/2.x.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#6071](https://github.com/seata/seata/pull/6071)] add git infos to jars
- [[#6042](https://github.com/seata/seata/pull/6042)] add secure authentication to interfaces in ClusterController
- [[#6091](https://github.com/seata/seata/pull/6091)] Optimizing the method of obtaining the tc address during raft authentication
- [[#6098](https://github.com/seata/seata/pull/6098)] optimize the retry logic in the acquireMetadata method


### security:
Expand Down
2 changes: 1 addition & 1 deletion changes/zh-cn/2.x.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
- [[#6071](https://github.com/seata/seata/pull/6071)] 添加git信息到JAR包中
- [[#6042](https://github.com/seata/seata/pull/6042)] 增加raft模式鉴权机制
- [[#6091](https://github.com/seata/seata/pull/6091)] 优化raft鉴权时获取tc地址的方式

- [[#6098](https://github.com/seata/seata/pull/6098)] 优化acquireMetadata方法的重试逻辑

### security:
- [[#6069](https://github.com/seata/seata/pull/6069)] 升级Guava依赖版本，修复安全漏洞
Expand Down
90 changes: 45 additions & 45 deletions ...iscovery-raft/src/main/java/io/seata/discovery/registry/raft/RaftRegistryServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,30 +169,38 @@ protected static void startQueryMetadata() {
long metadataMaxAgeMs = CONFIG.getLong(ConfigurationKeys.CLIENT_METADATA_MAX_AGE_MS, 30000L);
long currentTime = System.currentTimeMillis();
while (!CLOSED.get()) {
// Forced refresh of metadata information after set age
boolean fetch = System.currentTimeMillis() - currentTime > metadataMaxAgeMs;
String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
if (!fetch) {
fetch = watch();
}
// Cluster changes or reaches timeout refresh time
if (fetch) {
AtomicBoolean success = new AtomicBoolean(true);
METADATA.groups(clusterName).parallelStream().forEach(group -> {
try {
acquireClusterMetaData(clusterName, group);
} catch (Exception e) {
success.set(false);
// prevents an exception from being thrown that causes the thread to break
LOGGER.error("failed to get the leader address,error: {}", e.getMessage());
try {
// Forced refresh of metadata information after set age
boolean fetch = System.currentTimeMillis() - currentTime > metadataMaxAgeMs;
String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
if (!fetch) {
fetch = watch();
}
// Cluster changes or reaches timeout refresh time
if (fetch) {
for (String group : METADATA.groups(clusterName)) {
try {
acquireClusterMetaData(clusterName, group);
} catch (Exception e) {
// prevents an exception from being thrown that causes the thread to break
if (e instanceof RetryableException) {
throw e;
} else {
LOGGER.error("failed to get the leader address,error: {}", e.getMessage());
}
}
}
});
if (success.get()) {
currentTime = System.currentTimeMillis();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("refresh seata cluster metadata time: {}", currentTime);
}
}
} catch (RetryableException e) {
LOGGER.error(e.getMessage(), e);
try {
Thread.sleep(1000);
} catch (InterruptedException ignored) {
}
}
}
});
Expand Down Expand Up @@ -290,7 +298,7 @@ public List<InetSocketAddress> aliveLookup(String transactionServiceGroup) {
return RegistryService.super.aliveLookup(transactionServiceGroup);
}

private static boolean watch() {
private static boolean watch() throws RetryableException {
Map<String, String> header = new HashMap<>();
header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
Map<String, String> param = new HashMap<>();
Expand All @@ -299,36 +307,28 @@ private static boolean watch() {
groupTerms.forEach((k, v) -> param.put(k, String.valueOf(v)));
for (String group : groupTerms.keySet()) {
String tcAddress = queryHttpAddress(clusterName, group);
try {
if (isTokenExpired()) {
refreshToken(tcAddress);
}
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.");
}
if (isTokenExpired()) {
refreshToken(tcAddress);
}
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 statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
}
} catch (IOException e) {
throw new RetryableException(e.getMessage(), e);
return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
}
} catch (RetryableException e) {
} catch (IOException e) {
LOGGER.error("watch cluster node: {}, fail: {}", tcAddress, e.getMessage());
try {
Thread.sleep(1000);
} catch (InterruptedException ignored) {
}
continue;
throw new RetryableException(e.getMessage(), e);
}
break;
}
Expand Down
Loading