Flink: Close catalog opened by the dynamic sink serializer cache

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java

@@ -28,6 +28,7 @@
 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;
@@ -49,6 +50,9 @@
 class TableSerializerCache implements Serializable {
 
   private final CatalogLoader catalogLoader;
+
+  // Intentionally not closed; the catalog is reused for the serializer's lifetime.
+  private transient Catalog catalog;
   private final int maximumSize;
   private transient Map<String, SerializerInfo> serializers;
 
@@ -120,7 +124,11 @@ private class SerializerInfo {
     }
 
     private void update() {
-      Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
+      if (catalog == null) {
+        catalog = catalogLoader.loadCatalog();
+      }
+
+      Table table = catalog.loadTable(TableIdentifier.parse(tableName));
       schemas = table.schemas();
       specs = table.specs();
     }

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java

@@ -121,4 +121,47 @@ void testCacheSize() {
     cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
     assertThat(cache.maximumSize()).isEqualTo(1000);
   }
+
+  @Test
+  void testReusesCatalogAcrossLookups() {
+    Table table = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table"), schema1);
+    Table table2 = CATALOG_EXTENSION.catalog().createTable(TableIdentifier.of("table2"), schema2);
+
+    LoadCountingCatalogLoader catalogLoader =
+        new LoadCountingCatalogLoader(CATALOG_EXTENSION.catalogLoader());
+    cache = new TableSerializerCache(catalogLoader, 10);
+
+    // 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.loadCount()).isEqualTo(1);
+  }
+
+  private static class LoadCountingCatalogLoader implements CatalogLoader {
+    private final CatalogLoader delegate;
+    private int loadCount = 0;
+
+    private LoadCountingCatalogLoader(CatalogLoader delegate) {
+      this.delegate = delegate;
+    }
+
+    private int loadCount() {
+      return loadCount;
+    }
+
+    @Override
+    public Catalog loadCatalog() {
+      loadCount += 1;
+      return delegate.loadCatalog();
+    }
+
+    @Override
+    @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+    public CatalogLoader clone() {
+      return this;
+    }
+  }
 }