Skip to content

Core: Close FileIO on cache eviction to prevent thread leaks #15910

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

Open
utafrali wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Core: Close FileIO on cache eviction to prevent thread leaks #15910

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
adc2033
Core: Close FileIO on cache eviction to prevent thread leaks
utafrali Apr 8, 2026
038f0ed
Address review feedback
utafrali Apr 9, 2026
e16a5d9
Address review feedback
utafrali Apr 10, 2026
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
9 changes: 8 additions & 1 deletion core/src/main/java/org/apache/iceberg/CachingCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,9 +95,16 @@ class MetadataTableInvalidatingRemovalListener
@Override
public void onRemoval(TableIdentifier tableIdentifier, Table table, RemovalCause cause) {
LOG.debug("Evicted {} from the table cache ({})", tableIdentifier, cause);
if (RemovalCause.EXPIRED.equals(cause)) {
if (cause.wasEvicted()) {
if (!MetadataTableUtils.hasMetadataTableName(tableIdentifier)) {
tableCache.invalidateAll(metadataTableIdentifiers(tableIdentifier));
if (table != null) {
try {
table.io().close();
Copy link
Copy Markdown
Contributor

@anoopj anoopj Apr 10, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

This may not be safe for all catalogs. e.g. HadoopCatalog creates a common fileIO at the catalog level and not table level. The same fileIO reference is shared with the table object. So the close() here would break tables in the catalog that are still actively used.

Can this be just handled as part of the catalog's close() instead?

} catch (Exception e) {
LOG.warn("Failed to close FileIO for evicted table {}", tableIdentifier, e);
}
}
}
}
}
Expand Down
27 changes: 27 additions & 0 deletions core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.github.benmanes.caffeine.cache.Cache;
import java.io.IOException;
Expand All @@ -43,6 +46,7 @@
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.FakeTicker;
Expand Down Expand Up @@ -436,6 +440,29 @@ public void testRegisterTableWithOverwriteInvalidatesCache() throws Exception {
assertThat(catalog.cache().asMap()).doesNotContainKey(targetIdent);
}

@Test
public void testFileIOClosedOnCacheEviction() throws IOException {
FileIO mockIO = mock(FileIO.class);
Table mockTable = mock(Table.class);
TableIdentifier tableIdent = TableIdentifier.of("db", "tbl");

Catalog mockCatalog = mock(Catalog.class);
when(mockCatalog.loadTable(tableIdent)).thenReturn(mockTable);
when(mockTable.io()).thenReturn(mockIO);

TestableCachingCatalog catalog =
TestableCachingCatalog.wrap(mockCatalog, EXPIRATION_TTL, ticker);

catalog.loadTable(tableIdent);
assertThat(catalog.cache().asMap()).containsKey(tableIdent);

ticker.advance(EXPIRATION_TTL.plus(Duration.ofSeconds(1)));
catalog.cache().cleanUp();

assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent);
verify(mockIO).close();
}

public static TableIdentifier[] metadataTables(TableIdentifier tableIdent) {
return Arrays.stream(MetadataTableType.values())
.map(type -> TableIdentifier.parse(tableIdent + "." + type.name().toLowerCase(Locale.ROOT)))
Expand Down
Loading