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

HDFS-17749. [ARR] Fix wrong results due to the wrong usage of asyncComplete(null). #7466

Open
wants to merge 2 commits into
base: trunk
Choose a base branch
from
+89 −48
Open

HDFS-17749. [ARR] Fix wrong results due to the wrong usage of asyncComplete(null). #7466

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c639f4f
HDFS-17749. [ARR] Fix wrong results due to the wrong usage of asyncCo…
hfutatzhanghb Mar 5, 2025
da5fef7
reset unnecessary modification.
hfutatzhanghb Mar 5, 2025
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
4 changes: 2 additions & 2 deletions ...f/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2178,12 +2178,12 @@ protected HdfsFileStatus getFileInfoAll(final List<RemoteLocation> locations,
protected HdfsFileStatus getFileInfoAll(final List<RemoteLocation> locations,
final RemoteMethod method, long timeOutMs) throws IOException {

// Get the file info from everybody
// Get the file info from everybody.
Map<RemoteLocation, HdfsFileStatus> results =
rpcClient.invokeConcurrent(locations, method, false, false, timeOutMs,
HdfsFileStatus.class);
int children = 0;
// We return the first file
// We return the first file.
HdfsFileStatus dirStatus = null;
for (RemoteLocation loc : locations) {
HdfsFileStatus fileStatus = results.get(loc);
Expand Down
133 changes: 87 additions & 46 deletions ...java/org/apache/hadoop/hdfs/server/federation/router/async/RouterAsyncClientProtocol.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,7 +348,6 @@ public boolean mkdirs(String src, FsPermission masked, boolean createParent)
return rpcClient.invokeAll(locations, method);
}

asyncComplete(false);
if (locations.size() > 1) {
// Check if this directory already exists
asyncTry(() -> {
Expand All @@ -361,32 +360,41 @@ public boolean mkdirs(String src, FsPermission masked, boolean createParent)
return false;
});
});
asyncCatch((ret, ex) -> {
asyncCatch((ret, ioe) -> {
// Can't query if this file exists or not.
LOG.error("Error getting file info for {} while proxying mkdirs: {}",
src, ex.getMessage());
src, ioe.getMessage());
return false;
}, IOException.class);
}

final RemoteLocation firstLocation = locations.get(0);
asyncApply((AsyncApplyFunction<Boolean, Boolean>) success -> {
if (success) {
asyncComplete(true);
return;
}
asyncApply((AsyncApplyFunction<Boolean, Boolean>)ret -> {
if (!ret) {
final RemoteLocation firstLocation = locations.get(0);
asyncTry(() -> {
rpcClient.invokeSingle(firstLocation, method, Boolean.class);
});
asyncCatch((AsyncCatchFunction<Object, IOException>) (o, ioe) -> {
final List<RemoteLocation> newLocations = checkFaultTolerantRetry(
method, src, ioe, firstLocation, locations);
rpcClient.invokeSequential(
newLocations, method, Boolean.class, Boolean.TRUE);
}, IOException.class);
} else {
asyncComplete(ret);
}
});
} else {
final RemoteLocation firstLocation = locations.get(0);
asyncTry(() -> {
rpcClient.invokeSingle(firstLocation, method, Boolean.class);
});

asyncCatch((AsyncCatchFunction<Object, IOException>) (o, ioe) -> {
final List<RemoteLocation> newLocations = checkFaultTolerantRetry(
method, src, ioe, firstLocation, locations);
rpcClient.invokeSequential(
newLocations, method, Boolean.class, Boolean.TRUE);
}, IOException.class);
});

}
return asyncReturn(Boolean.class);
}

Expand Down Expand Up @@ -480,6 +488,7 @@ public DirectoryListing getListing(
return null;
});
});
boolean finalNamenodeListingExists = namenodeListingExists;
asyncApply(o -> {
// Update the remaining count to include left mount points
if (nnListing.size() > 0) {
Expand All @@ -492,10 +501,12 @@ public DirectoryListing getListing(
}
}
}
return null;
return finalNamenodeListingExists;
});
} else {
asyncComplete(namenodeListingExists);
}
asyncComplete(namenodeListingExists);

asyncApply((ApplyFunction<Boolean, Object>) exists -> {
if (!exists && nnListing.size() == 0 && children == null) {
// NN returns a null object if the directory cannot be found and has no
Expand Down Expand Up @@ -541,7 +552,6 @@ protected List<RemoteResult<RemoteLocation, DirectoryListing>> getListingInt(
return asyncReturn(List.class);
}


@Override
public HdfsFileStatus getFileInfo(String src) throws IOException {
rpcServer.checkOperation(NameNode.OperationCategory.READ);
Expand All @@ -564,7 +574,7 @@ public HdfsFileStatus getFileInfo(String src) throws IOException {
|| e instanceof RouterResolveException) {
noLocationException[0] = e;
}
throw e;
return null;
}, IOException.class);

asyncApply((AsyncApplyFunction<HdfsFileStatus, Object>) ret -> {
Expand All @@ -582,15 +592,18 @@ public HdfsFileStatus getFileInfo(String src) throws IOException {
// The src is a mount point, but there are no files or directories
getMountPointStatus(src, 0, 0, false);
} else {
if (noLocationException[0] != null) {
throw noLocationException[0];
}
asyncComplete(null);
return;
}
asyncApply((ApplyFunction<HdfsFileStatus, HdfsFileStatus>) result -> {
// Can't find mount point for path and the path didn't contain any sub monit points,
// throw the NoLocationException to client.
if (result == null && noLocationException[0] != null) {
throw noLocationException[0];
}

return result;
});
} else {
Expand Down Expand Up @@ -639,7 +652,14 @@ public HdfsFileStatus getMountPointStatus(
final int[] childrenNums = new int[]{childrenNum};
final EnumSet<HdfsFileStatus.Flags>[] flags =
new EnumSet[]{EnumSet.noneOf(HdfsFileStatus.Flags.class)};
asyncComplete(null);
long inodeId = 0;
HdfsFileStatus.Builder builder = new HdfsFileStatus.Builder();
if (setPath) {
Path path = new Path(name);
String nameStr = path.getName();
builder.path(DFSUtil.string2Bytes(nameStr));
}

if (getSubclusterResolver() instanceof MountTableResolver) {
asyncTry(() -> {
String mName = name.startsWith("/") ? name : "/" + name;
Expand All @@ -664,13 +684,45 @@ public HdfsFileStatus getMountPointStatus(
.getFlags(fInfo.isEncrypted(), fInfo.isErasureCoded(),
fInfo.isSnapshotEnabled(), fInfo.hasAcl());
}
return fInfo;
return builder.isdir(true)
.mtime(modTime)
.atime(accessTime)
.perm(permission[0])
.owner(owner[0])
.group(group[0])
.symlink(new byte[0])
.fileId(inodeId)
.children(childrenNums[0])
.flags(flags[0])
.build();
});
} else {
asyncComplete(builder.isdir(true)
.mtime(modTime)
.atime(accessTime)
.perm(permission[0])
.owner(owner[0])
.group(group[0])
.symlink(new byte[0])
.fileId(inodeId)
.children(childrenNums[0])
.flags(flags[0])
.build());
}
});
asyncCatch((CatchFunction<HdfsFileStatus, IOException>) (status, e) -> {
LOG.error("Cannot get mount point: {}", e.getMessage());
return status;
return builder.isdir(true)
.mtime(modTime)
.atime(accessTime)
.perm(permission[0])
.owner(owner[0])
.group(group[0])
.symlink(new byte[0])
.fileId(inodeId)
.children(childrenNums[0])
.flags(flags[0])
.build();
}, IOException.class);
} else {
try {
Expand All @@ -684,44 +736,33 @@ public HdfsFileStatus getMountPointStatus(
} else {
LOG.debug(msg);
}
} finally {
asyncComplete(builder.isdir(true)
.mtime(modTime)
.atime(accessTime)
.perm(permission[0])
.owner(owner[0])
.group(group[0])
.symlink(new byte[0])
.fileId(inodeId)
.children(childrenNums[0])
.flags(flags[0])
.build());
}
}
long inodeId = 0;
HdfsFileStatus.Builder builder = new HdfsFileStatus.Builder();
asyncApply((ApplyFunction<HdfsFileStatus, HdfsFileStatus>) status -> {
if (setPath) {
Path path = new Path(name);
String nameStr = path.getName();
builder.path(DFSUtil.string2Bytes(nameStr));
}

return builder.isdir(true)
.mtime(modTime)
.atime(accessTime)
.perm(permission[0])
.owner(owner[0])
.group(group[0])
.symlink(new byte[0])
.fileId(inodeId)
.children(childrenNums[0])
.flags(flags[0])
.build();
});
return asyncReturn(HdfsFileStatus.class);
}

@Override
protected HdfsFileStatus getFileInfoAll(final List<RemoteLocation> locations,
final RemoteMethod method, long timeOutMs) throws IOException {

asyncComplete(null);
// Get the file info from everybody
// Get the file info from everybody.
rpcClient.invokeConcurrent(locations, method, false, false, timeOutMs,
HdfsFileStatus.class);
asyncApply(res -> {
Map<RemoteLocation, HdfsFileStatus> results = (Map<RemoteLocation, HdfsFileStatus>) res;
int children = 0;
// We return the first file
// We return the first file.
HdfsFileStatus dirStatus = null;
for (RemoteLocation loc : locations) {
HdfsFileStatus fileStatus = results.get(loc);
Expand Down