Skip to content

[HUDI-1358] Fix leaks in DiskBasedMap and LazyFileIterable #2249

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

Merged
vinothchandar merged 1 commit into from
Nov 23, 2020
Merged

[HUDI-1358] Fix leaks in DiskBasedMap and LazyFileIterable #2249

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 2 additions & 1 deletion hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,7 +273,8 @@ public WriteStatus close() {
insertRecordsWritten++;
}
}
keyToNewRecords.clear();

((ExternalSpillableMap) keyToNewRecords).close();
Copy link
Contributor Author

@bvaradar bvaradar Nov 12, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

merge handle seems to be the only place where deleting on jvm shutdown would be a problem. In other cases, they are fairly long lived.

writtenRecordKeys.clear();

if (fileWriter != null) {
Expand Down
64 changes: 37 additions & 27 deletions hudi-common/src/main/java/org/apache/hudi/common/util/collection/DiskBasedMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,8 @@ public final class DiskBasedMap<T extends Serializable, R extends Serializable>
private ThreadLocal<BufferedRandomAccessFile> randomAccessFile = new ThreadLocal<>();
private Queue<BufferedRandomAccessFile> openedAccessFiles = new ConcurrentLinkedQueue<>();

private transient Thread shutdownThread = null;

public DiskBasedMap(String baseFilePath) throws IOException {
this.valueMetadataMap = new ConcurrentHashMap<>();
this.writeOnlyFile = new File(baseFilePath, UUID.randomUUID().toString());
Expand Down Expand Up @@ -126,33 +128,8 @@ private void initFile(File writeOnlyFile) throws IOException {
* (typically 4 KB) to disk.
*/
private void addShutDownHook() {
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
try {
if (writeOnlyFileHandle != null) {
writeOnlyFileHandle.flush();
fileOutputStream.getChannel().force(false);
writeOnlyFileHandle.close();
}

while (!openedAccessFiles.isEmpty()) {
BufferedRandomAccessFile file = openedAccessFiles.poll();
if (null != file) {
try {
file.close();
} catch (IOException ioe) {
// skip exception
}
}
}
writeOnlyFile.delete();
} catch (Exception e) {
// delete the file for any sort of exception
writeOnlyFile.delete();
}
}
});
shutdownThread = new Thread(this::cleanup);
Runtime.getRuntime().addShutdownHook(shutdownThread);
}

private void flushToDisk() {
Expand Down Expand Up @@ -267,6 +244,39 @@ public void clear() {
// reducing concurrency). Instead, just clear the pointer map. The file will be removed on exit.
}

public void close() {
cleanup();
if (shutdownThread != null) {
Runtime.getRuntime().removeShutdownHook(shutdownThread);
}
}

private void cleanup() {
valueMetadataMap.clear();
try {
if (writeOnlyFileHandle != null) {
writeOnlyFileHandle.flush();
fileOutputStream.getChannel().force(false);
writeOnlyFileHandle.close();
}

while (!openedAccessFiles.isEmpty()) {
BufferedRandomAccessFile file = openedAccessFiles.poll();
if (null != file) {
try {
file.close();
} catch (IOException ioe) {
// skip exception
}
}
}
writeOnlyFile.delete();
} catch (Exception e) {
// delete the file for any sort of exception
writeOnlyFile.delete();
}
}

@Override
public Set<T> keySet() {
return valueMetadataMap.keySet();
Expand Down
4 changes: 4 additions & 0 deletions hudi-common/src/main/java/org/apache/hudi/common/util/collection/ExternalSpillableMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,10 @@ public void clear() {
currentInMemoryMapSize = 0L;
}

public void close() {
getDiskBasedMap().close();
}

@Override
public Set<T> keySet() {
Set<T> keySet = new HashSet<T>();
Expand Down
15 changes: 9 additions & 6 deletions hudi-common/src/main/java/org/apache/hudi/common/util/collection/LazyFileIterable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ public class LazyFileIterable<T, R> implements Iterable<R> {
// Stores the key and corresponding value's latest metadata spilled to disk
private final Map<T, DiskBasedMap.ValueMetadata> inMemoryMetadataOfSpilledData;

private transient Thread shutdownThread = null;

public LazyFileIterable(String filePath, Map<T, DiskBasedMap.ValueMetadata> map) {
this.filePath = filePath;
this.inMemoryMetadataOfSpilledData = map;
Expand Down Expand Up @@ -103,6 +105,11 @@ public void forEachRemaining(Consumer<? super R> action) {
}

private void close() {
closeHandle();
Runtime.getRuntime().removeShutdownHook(shutdownThread);
}

private void closeHandle() {
if (readOnlyFileHandle != null) {
try {
readOnlyFileHandle.close();
Expand All @@ -114,12 +121,8 @@ private void close() {
}

private void addShutdownHook() {
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
close();
}
});
shutdownThread = new Thread(this::closeHandle);
Runtime.getRuntime().addShutdownHook(shutdownThread);
}
}
}