xianjingfeng commented on code in PR #1897:
URL:
https://github.com/apache/incubator-uniffle/pull/1897#discussion_r1694429232
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -179,47 +185,105 @@ public void nodesCheckTest() {
nodesCheck();
}
- private void updateExcludeNodes(String path) {
- int originalExcludeNodesNumber = excludeNodes.size();
+ private synchronized void updateExcludeNodes(String path) {
+ int originalExcludeNodesNumber = excludedNodes.size();
try {
Path hadoopPath = new Path(path);
FileStatus fileStatus = hadoopFileSystem.getFileStatus(hadoopPath);
if (fileStatus != null && fileStatus.isFile()) {
long latestModificationTime = fileStatus.getModificationTime();
if (excludeLastModify.get() != latestModificationTime) {
- parseExcludeNodesFile(hadoopFileSystem.open(hadoopPath));
+ excludedNodes =
parseExcludedNodesFile(hadoopFileSystem.open(hadoopPath), true);
+ LOG.info(
+ "Updated exclude nodes and {} nodes were marked as exclude
nodes",
+ excludedNodes.size());
+ // update exclude nodes and last modify time
excludeLastModify.set(latestModificationTime);
}
} else {
- excludeNodes = Sets.newConcurrentHashSet();
+ excludedNodes = Sets.newConcurrentHashSet();
}
} catch (FileNotFoundException fileNotFoundException) {
- excludeNodes = Sets.newConcurrentHashSet();
+ excludedNodes = Sets.newConcurrentHashSet();
} catch (Exception e) {
LOG.warn("Error when updating exclude nodes, the exclude nodes file
path: {}.", path, e);
}
- int newlyExcludeNodesNumber = excludeNodes.size();
+ int newlyExcludeNodesNumber = excludedNodes.size();
if (newlyExcludeNodesNumber != originalExcludeNodesNumber) {
- LOG.info("Exclude nodes number: {}, nodes list: {}",
newlyExcludeNodesNumber, excludeNodes);
+ LOG.info("Exclude nodes number: {}, nodes list: {}",
newlyExcludeNodesNumber, excludedNodes);
}
- CoordinatorMetrics.gaugeExcludeServerNum.set(excludeNodes.size());
+ CoordinatorMetrics.gaugeExcludeServerNum.set(excludedNodes.size());
}
- private void parseExcludeNodesFile(DataInputStream fsDataInputStream) throws
IOException {
+ private Set<String> parseExcludedNodesFile(
+ DataInputStream fsDataInputStream, boolean notNeedComment) throws
IOException {
Set<String> nodes = Sets.newConcurrentHashSet();
try (BufferedReader br =
new BufferedReader(new InputStreamReader(fsDataInputStream,
StandardCharsets.UTF_8))) {
String line;
while ((line = br.readLine()) != null) {
- if (!StringUtils.isEmpty(line) && !line.trim().startsWith("#")) {
+ if (notNeedComment) {
+ if (!StringUtils.isEmpty(line) && !line.trim().startsWith("#")) {
+ nodes.add(line.trim());
+ }
+ } else {
nodes.add(line.trim());
}
}
}
- // update exclude nodes and last modify time
- excludeNodes = nodes;
- LOG.info(
- "Updated exclude nodes and {} nodes were marked as exclude nodes",
excludeNodes.size());
+ return nodes;
+ }
+
+ private synchronized boolean putInExcludedNodesFile(List<String>
excludedNodes)
Review Comment:
We can define a method to write to hdfs, the code is as follows.
```
private synchronized writeExcludeNodesToFile(List<String> excludedNodes) {
// 1.write all nodes in excludedNodes to a tmp file.
// 2.rename the tmp file to exclude-file
}
```
And then the method `putInExcludedNodesFile` would be as follows.
```
private synchronized boolean putInExcludedNodesFile(List<String>
excludedNodes) {
//1. invoke `parseExcludedNodesFile`
//2. merge the result of `parseExcludedNodesFile` and excludedNodes
//3. invoke 'writeExcludeNodesToFile'
}
```
--
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]
