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

HBASE-23679 FileSystem objects leak when cleaned up in cleanupBulkLoad #1029

Closed
wants to merge 2 commits into from
+20 −17
Closed

HBASE-23679 FileSystem objects leak when cleaned up in cleanupBulkLoad #1029

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
63bf5f7
HBASE-23679 FileSystem objects leak when cleaned up in cleanupBulkLoad
joshelser Jan 13, 2020
6eda3a3
Fix logging
joshelser Jan 13, 2020
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
37 changes: 20 additions & 17 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,26 +153,15 @@ public String prepareBulkLoad(final HRegion region, final PrepareBulkLoadRequest

public void cleanupBulkLoad(final HRegion region, final CleanupBulkLoadRequest request)
throws IOException {
try {
region.getCoprocessorHost().preCleanupBulkLoad(getActiveUser());
region.getCoprocessorHost().preCleanupBulkLoad(getActiveUser());

Path path = new Path(request.getBulkToken());
if (!fs.delete(path, true)) {
if (fs.exists(path)) {
throw new IOException("Failed to clean up " + path);
}
}
LOG.info("Cleaned up " + path + " successfully.");
} finally {
UserGroupInformation ugi = getActiveUser().getUGI();
try {
if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
FileSystem.closeAllForUGI(ugi);
}
} catch (IOException e) {
LOG.error("Failed to close FileSystem for: " + ugi, e);
Path path = new Path(request.getBulkToken());
if (!fs.delete(path, true)) {
if (fs.exists(path)) {
throw new IOException("Failed to clean up " + path);
}
}
LOG.trace("Cleaned up {} successfully.", path);
}

private Consumer<HRegion> fsCreatedListener;
Expand Down Expand Up @@ -281,6 +270,13 @@ public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
public Map<byte[], List<Path>> run() {
FileSystem fs = null;
try {
/*
* This is creating and caching a new FileSystem instance. Other code called
* "beneath" this method will rely on this FileSystem instance being in the
* cache. This is important as those methods make _no_ attempt to close this
* FileSystem instance. It is critical that here, in SecureBulkLoadManager,
* we are tracking the lifecycle and closing the FS when safe to do so.
*/
fs = FileSystem.get(conf);
for(Pair<byte[], String> el: familyPaths) {
Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
Expand All @@ -305,6 +301,13 @@ public Map<byte[], List<Path>> run() {
});
} finally {
decrementUgiReference(ugi);
try {
if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
FileSystem.closeAllForUGI(ugi);
}
} catch (IOException e) {
LOG.error("Failed to close FileSystem for: {}", ugi, e);
}
if (region.getCoprocessorHost() != null) {
region.getCoprocessorHost().postBulkLoadHFile(familyPaths, map);
}
Expand Down