Skip to content

Flink: Close catalog opened by the dynamic sink serializer cache #17437

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
vishnuprakaz wants to merge 7 commits into
base: main
Choose a base branch
from
+52 −1
Open

Flink: Close catalog opened by the dynamic sink serializer cache #17437

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d4b2b6b
Flink: Close catalog opened by the dynamic sink serializer cache
vishnuprakaz Jul 30, 2026
6d539f2
Rewrite serializer cache close test without Mockito
vishnuprakaz Aug 4, 2026
68de894
Rework close test on the extension's catalog loader
vishnuprakaz Aug 5, 2026
e3ab69a
Use the extension's managed catalog in test setup
vishnuprakaz Aug 5, 2026
2ea25d5
Manage catalog lifecycle via TableLoader in serializer cache
vishnuprakaz Aug 6, 2026
ff5fa12
Load the catalog once and reuse it in the serializer cache
vishnuprakaz Aug 6, 2026
0a43689
Shorten catalog field comment
vishnuprakaz Aug 6, 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
26 changes: 23 additions & 3 deletions .../v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
*/
package org.apache.iceberg.flink.sink.dynamic;

import java.io.Closeable;
import java.io.IOException;
import java.io.Serializable;
import java.util.Map;
import javax.annotation.Nullable;
Expand All @@ -28,11 +30,14 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* A Cache which holds Flink's {@link RowDataSerializer} for a given table name and schema. This
Expand All @@ -48,6 +53,8 @@
@Internal
class TableSerializerCache implements Serializable {

private static final Logger LOG = LoggerFactory.getLogger(TableSerializerCache.class);

private final CatalogLoader catalogLoader;
private final int maximumSize;
private transient Map<String, SerializerInfo> serializers;
Expand Down Expand Up @@ -120,9 +127,22 @@ private class SerializerInfo {
}

private void update() {
Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
schemas = table.schemas();
specs = table.specs();
// The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
// reading the table metadata, to avoid leaking one per cache miss.
Catalog catalog = catalogLoader.loadCatalog();
try {
Table table = catalog.loadTable(TableIdentifier.parse(tableName));
schemas = table.schemas();
specs = table.specs();
} finally {
if (catalog instanceof Closeable) {
try {
((Closeable) catalog).close();
} catch (IOException e) {
LOG.warn("Failed to close catalog {}", catalog.name(), e);
}
}
Comment thread
mxm marked this conversation as resolved.
Outdated

@mxm mxm Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Would it be feasible to let the CatalogLoader manage the lifecycle of the catalog and reuse it when necessary? It is a bit odd to check for Closable when the Catalog interface doesn't contain this interface.

@vishnuprakaz vishnuprakaz Aug 5, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

On this I looked into having the loader manage and reuse the catalog. the problem I ran into is that the serializer has no teardown hook, so as far as I can tell nothing at this layer could ever close a held catalog. Does that match your thinking that the reuse part belongs in the TaskManager level cache?

While checking, I noticed loadCatalog() has six other call sites in the sink that don't cloae the catalog either. The operator-level ones do have close() hooks, would it be worth a similar interim fix for those, or better to leave them for the TaskManager level cache work?

For the Closeable check here.... would it make sense to switch update() to TableLoader.fromCatalog() in a try with resources, like FlinkSink and IcebergSink do for one shot loads?
but the lifecycle (and the Closeable check) would live inside TableLoader instead of this class.

}
}
}

Expand Down
48 changes: 48 additions & 0 deletions ...1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,16 +25,20 @@
import static org.apache.iceberg.types.Types.StringType;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.IOException;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.HadoopCatalogExtension;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;

Expand Down Expand Up @@ -121,4 +125,48 @@ void testCacheSize() {
cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
assertThat(cache.maximumSize()).isEqualTo(1000);
}

@Test
void testClosesCatalogAfterSchemaLookup() {
CloseCountingInMemoryCatalog catalog = new CloseCountingInMemoryCatalog();
catalog.initialize("tracking", Map.of());
catalog.createNamespace(Namespace.of("db"));
Table table = catalog.createTable(TableIdentifier.of("db", "table"), schema1);
cache = new TableSerializerCache(new SingleCatalogLoader(catalog), 10);

// schema/spec ids are unknown, so this misses the cache and loads a catalog to resolve them
cache.serializerWithSchemaAndSpec(
"db.table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId());

assertThat(catalog.closeCount).isEqualTo(1);
}

private static class CloseCountingInMemoryCatalog extends InMemoryCatalog {
private int closeCount = 0;

@Override
public void close() throws IOException {
super.close();
closeCount += 1;
}
}

private static class SingleCatalogLoader implements CatalogLoader {

@mxm mxm Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Can we replace this with HadoopCatalogExtension?

private final Catalog catalog;

private SingleCatalogLoader(Catalog catalog) {
this.catalog = catalog;
}

@Override
public Catalog loadCatalog() {
return catalog;
}

@Override
@SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
public CatalogLoader clone() {
return this;
}
}
}
Loading