From d4b2b6bfdcdf8173e841a90351e358b73af03655 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Thu, 30 Jul 2026 22:50:10 +0530 Subject: [PATCH 1/7] Flink: Close catalog opened by the dynamic sink serializer cache --- .../sink/dynamic/TableSerializerCache.java | 26 +++++++++++++++--- .../dynamic/TestTableSerializerCache.java | 27 +++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java index 84d0ed9be5d0..35723f0e036f 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java @@ -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; @@ -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 @@ -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 serializers; @@ -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); + } + } + } } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java index 1cf2c8bae001..d7df87c5186c 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java @@ -24,7 +24,13 @@ import static org.apache.iceberg.types.Types.NestedField.required; import static org.apache.iceberg.types.Types.StringType; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; +import java.io.Closeable; import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.table.runtime.typeutils.RowDataSerializer; @@ -121,4 +127,25 @@ void testCacheSize() { cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000); assertThat(cache.maximumSize()).isEqualTo(1000); } + + @Test + void testClosesCatalogAfterSchemaLookup() throws Exception { + Table table = + CATALOG_EXTENSION + .catalogLoader() + .loadCatalog() + .createTable(TableIdentifier.of("table"), schema1); + + Catalog catalog = mock(Catalog.class, withSettings().extraInterfaces(Closeable.class)); + when(catalog.loadTable(any(TableIdentifier.class))).thenReturn(table); + CatalogLoader catalogLoader = mock(CatalogLoader.class); + when(catalogLoader.loadCatalog()).thenReturn(catalog); + cache = new TableSerializerCache(catalogLoader, 10); + + // schema/spec ids are unknown, so this misses the cache and loads a catalog to resolve them + cache.serializerWithSchemaAndSpec( + "table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId()); + + verify((Closeable) catalog).close(); + } } From 6d539f2573d91faed599182625ecf48e844419f5 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Tue, 4 Aug 2026 16:04:32 +0530 Subject: [PATCH 2/7] Rewrite serializer cache close test without Mockito --- .../dynamic/TestTableSerializerCache.java | 61 +++++++++++++------ 1 file changed, 41 insertions(+), 20 deletions(-) diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java index d7df87c5186c..e1f98922f30e 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java @@ -24,13 +24,9 @@ import static org.apache.iceberg.types.Types.NestedField.required; import static org.apache.iceberg.types.Types.StringType; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.mockito.Mockito.withSettings; -import java.io.Closeable; +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; @@ -38,9 +34,11 @@ 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; @@ -129,23 +127,46 @@ void testCacheSize() { } @Test - void testClosesCatalogAfterSchemaLookup() throws Exception { - Table table = - CATALOG_EXTENSION - .catalogLoader() - .loadCatalog() - .createTable(TableIdentifier.of("table"), schema1); - - Catalog catalog = mock(Catalog.class, withSettings().extraInterfaces(Closeable.class)); - when(catalog.loadTable(any(TableIdentifier.class))).thenReturn(table); - CatalogLoader catalogLoader = mock(CatalogLoader.class); - when(catalogLoader.loadCatalog()).thenReturn(catalog); - cache = new TableSerializerCache(catalogLoader, 10); + 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( - "table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId()); + "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 { + private final Catalog catalog; + + private SingleCatalogLoader(Catalog catalog) { + this.catalog = catalog; + } + + @Override + public Catalog loadCatalog() { + return catalog; + } - verify((Closeable) catalog).close(); + @Override + @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"}) + public CatalogLoader clone() { + return this; + } } } From 68de8943bec4fd75908fed2dd6eae9e1b340ae07 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Wed, 5 Aug 2026 12:10:59 +0530 Subject: [PATCH 3/7] Rework close test on the extension's catalog loader --- .../dynamic/TestTableSerializerCache.java | 78 ++++++++++++++----- 1 file changed, 57 insertions(+), 21 deletions(-) diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java index e1f98922f30e..98bb82b37b23 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java @@ -25,8 +25,9 @@ import static org.apache.iceberg.types.Types.StringType; import static org.assertj.core.api.Assertions.assertThat; +import java.io.Closeable; import java.io.IOException; -import java.util.Map; +import java.util.List; import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.table.runtime.typeutils.RowDataSerializer; @@ -38,7 +39,6 @@ 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; @@ -128,39 +128,38 @@ void testCacheSize() { @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); + Table table = + CATALOG_EXTENSION + .catalogLoader() + .loadCatalog() + .createTable(TableIdentifier.of("table"), schema1); + + CloseCountingCatalogLoader catalogLoader = + new CloseCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader()); + cache = new TableSerializerCache(catalogLoader, 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()); + "table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId()); - assertThat(catalog.closeCount).isEqualTo(1); + assertThat(catalogLoader.closeCount()).isEqualTo(1); } - private static class CloseCountingInMemoryCatalog extends InMemoryCatalog { + private static class CloseCountingCatalogLoader implements CatalogLoader { + private final CatalogLoader delegate; private int closeCount = 0; - @Override - public void close() throws IOException { - super.close(); - closeCount += 1; + private CloseCountingCatalogLoader(CatalogLoader delegate) { + this.delegate = delegate; } - } - private static class SingleCatalogLoader implements CatalogLoader { - private final Catalog catalog; - - private SingleCatalogLoader(Catalog catalog) { - this.catalog = catalog; + private int closeCount() { + return closeCount; } @Override public Catalog loadCatalog() { - return catalog; + return new CloseCountingCatalog(delegate.loadCatalog()); } @Override @@ -168,5 +167,42 @@ public Catalog loadCatalog() { public CatalogLoader clone() { return this; } + + private class CloseCountingCatalog implements Catalog, Closeable { + private final Catalog delegate; + + private CloseCountingCatalog(Catalog delegate) { + this.delegate = delegate; + } + + @Override + public List listTables(Namespace namespace) { + return delegate.listTables(namespace); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + return delegate.dropTable(identifier, purge); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + delegate.renameTable(from, to); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + return delegate.loadTable(identifier); + } + + @Override + public void close() throws IOException { + if (delegate instanceof Closeable) { + ((Closeable) delegate).close(); + } + + closeCount += 1; + } + } } } From e3ab69a2930949b48cb741e9f49e75f015b519fd Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Wed, 5 Aug 2026 12:33:37 +0530 Subject: [PATCH 4/7] Use the extension's managed catalog in test setup --- .../flink/sink/dynamic/TestTableSerializerCache.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java index 98bb82b37b23..1c64b6813a5d 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java @@ -128,11 +128,7 @@ void testCacheSize() { @Test void testClosesCatalogAfterSchemaLookup() { - Table table = - CATALOG_EXTENSION - .catalogLoader() - .loadCatalog() - .createTable(TableIdentifier.of("table"), schema1); + Table table = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1); CloseCountingCatalogLoader catalogLoader = new CloseCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader()); From 2ea25d51adc9afd71147e0c79d9111d1c5dbf876 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Thu, 6 Aug 2026 12:26:14 +0530 Subject: [PATCH 5/7] Manage catalog lifecycle via TableLoader in serializer cache --- .../sink/dynamic/TableSerializerCache.java | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java index 35723f0e036f..d9ea5ad55c68 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java @@ -18,7 +18,6 @@ */ package org.apache.iceberg.flink.sink.dynamic; -import java.io.Closeable; import java.io.IOException; import java.io.Serializable; import java.util.Map; @@ -30,10 +29,10 @@ 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.flink.TableLoader; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.slf4j.Logger; @@ -127,21 +126,17 @@ private class SerializerInfo { } private void update() { - // 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)); + // The serializer has no teardown hook, so a catalog cannot be held for reuse; load and + // close one per cache miss. + try (TableLoader tableLoader = + TableLoader.fromCatalog(catalogLoader, TableIdentifier.parse(tableName))) { + tableLoader.open(); + Table table = tableLoader.loadTable(); 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); - } - } + } catch (IOException e) { + // only close() throws IOException here; a failed close should not fail the lookup + LOG.warn("Failed to close catalog for table {}", tableName, e); } } } From ff5fa12312a549d3b21ed8b03ddb2bf14345cc97 Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Thu, 6 Aug 2026 17:46:52 +0530 Subject: [PATCH 6/7] Load the catalog once and reuse it in the serializer cache --- .../sink/dynamic/TableSerializerCache.java | 28 +++----- .../dynamic/TestTableSerializerCache.java | 67 +++++-------------- 2 files changed, 26 insertions(+), 69 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java index d9ea5ad55c68..1ea58d785067 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java @@ -18,7 +18,6 @@ */ package org.apache.iceberg.flink.sink.dynamic; -import java.io.IOException; import java.io.Serializable; import java.util.Map; import javax.annotation.Nullable; @@ -29,14 +28,12 @@ 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.flink.TableLoader; 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 @@ -52,9 +49,11 @@ @Internal class TableSerializerCache implements Serializable { - private static final Logger LOG = LoggerFactory.getLogger(TableSerializerCache.class); - private final CatalogLoader catalogLoader; + + // Loaded once and reused for all lookups; not closed here, cleanup is deferred to the + // planned TaskManager-level catalog cache. Transient: loaded on first use per instance. + private transient Catalog catalog; private final int maximumSize; private transient Map serializers; @@ -126,18 +125,13 @@ private class SerializerInfo { } private void update() { - // The serializer has no teardown hook, so a catalog cannot be held for reuse; load and - // close one per cache miss. - try (TableLoader tableLoader = - TableLoader.fromCatalog(catalogLoader, TableIdentifier.parse(tableName))) { - tableLoader.open(); - Table table = tableLoader.loadTable(); - schemas = table.schemas(); - specs = table.specs(); - } catch (IOException e) { - // only close() throws IOException here; a failed close should not fail the lookup - LOG.warn("Failed to close catalog for table {}", tableName, e); + if (catalog == null) { + catalog = catalogLoader.loadCatalog(); } + + Table table = catalog.loadTable(TableIdentifier.parse(tableName)); + schemas = table.schemas(); + specs = table.specs(); } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java index 1c64b6813a5d..bbfb39618fc3 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java @@ -25,9 +25,6 @@ import static org.apache.iceberg.types.Types.StringType; import static org.assertj.core.api.Assertions.assertThat; -import java.io.Closeable; -import java.io.IOException; -import java.util.List; import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.table.runtime.typeutils.RowDataSerializer; @@ -35,7 +32,6 @@ 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; @@ -127,35 +123,39 @@ void testCacheSize() { } @Test - void testClosesCatalogAfterSchemaLookup() { + void testReusesCatalogAcrossLookups() { Table table = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1); + Table table2 = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table2"), schema2); - CloseCountingCatalogLoader catalogLoader = - new CloseCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader()); + LoadCountingCatalogLoader catalogLoader = + new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader()); cache = new TableSerializerCache(catalogLoader, 10); - // schema/spec ids are unknown, so this misses the cache and loads a catalog to resolve them + // schema/spec ids are unknown, so both lookups miss the cache and need the catalog cache.serializerWithSchemaAndSpec( "table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId()); + cache.serializerWithSchemaAndSpec( + "table2", table2.schema().schemaId(), PartitionSpec.unpartitioned().specId()); - assertThat(catalogLoader.closeCount()).isEqualTo(1); + assertThat(catalogLoader.loadCount()).isEqualTo(1); } - private static class CloseCountingCatalogLoader implements CatalogLoader { + private static class LoadCountingCatalogLoader implements CatalogLoader { private final CatalogLoader delegate; - private int closeCount = 0; + private int loadCount = 0; - private CloseCountingCatalogLoader(CatalogLoader delegate) { + private LoadCountingCatalogLoader(CatalogLoader delegate) { this.delegate = delegate; } - private int closeCount() { - return closeCount; + private int loadCount() { + return loadCount; } @Override public Catalog loadCatalog() { - return new CloseCountingCatalog(delegate.loadCatalog()); + loadCount += 1; + return delegate.loadCatalog(); } @Override @@ -163,42 +163,5 @@ public Catalog loadCatalog() { public CatalogLoader clone() { return this; } - - private class CloseCountingCatalog implements Catalog, Closeable { - private final Catalog delegate; - - private CloseCountingCatalog(Catalog delegate) { - this.delegate = delegate; - } - - @Override - public List listTables(Namespace namespace) { - return delegate.listTables(namespace); - } - - @Override - public boolean dropTable(TableIdentifier identifier, boolean purge) { - return delegate.dropTable(identifier, purge); - } - - @Override - public void renameTable(TableIdentifier from, TableIdentifier to) { - delegate.renameTable(from, to); - } - - @Override - public Table loadTable(TableIdentifier identifier) { - return delegate.loadTable(identifier); - } - - @Override - public void close() throws IOException { - if (delegate instanceof Closeable) { - ((Closeable) delegate).close(); - } - - closeCount += 1; - } - } } } From 0a436891bb59dc14bd86842c5af86b6daeb4c56f Mon Sep 17 00:00:00 2001 From: vishnuprakaz Date: Thu, 6 Aug 2026 17:49:06 +0530 Subject: [PATCH 7/7] Shorten catalog field comment --- .../iceberg/flink/sink/dynamic/TableSerializerCache.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java index 1ea58d785067..2d0beb5ddd63 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java @@ -51,8 +51,7 @@ class TableSerializerCache implements Serializable { private final CatalogLoader catalogLoader; - // Loaded once and reused for all lookups; not closed here, cleanup is deferred to the - // planned TaskManager-level catalog cache. Transient: loaded on first use per instance. + // Intentionally not closed; the catalog is reused for the serializer's lifetime. private transient Catalog catalog; private final int maximumSize; private transient Map serializers;