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

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.flink.sink.dynamic;
 
+import java.io.IOException;
 import java.io.Serializable;
 import java.util.Map;
 import javax.annotation.Nullable;
@@ -31,8 +32,11 @@
 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
@@ -48,6 +52,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;
@@ -120,9 +126,18 @@ 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 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);
+      }
     }
   }
 

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

@@ -25,13 +25,17 @@
 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;
 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;
@@ -121,4 +125,80 @@ void testCacheSize() {
     cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
     assertThat(cache.maximumSize()).isEqualTo(1000);
   }
+
+  @Test
+  void testClosesCatalogAfterSchemaLookup() {
+    Table table = CATALOG_EXTENSION.catalog().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(
+        "table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId());
+
+    assertThat(catalogLoader.closeCount()).isEqualTo(1);
+  }
+
+  private static class CloseCountingCatalogLoader implements CatalogLoader {
+    private final CatalogLoader delegate;
+    private int closeCount = 0;
+
+    private CloseCountingCatalogLoader(CatalogLoader delegate) {
+      this.delegate = delegate;
+    }
+
+    private int closeCount() {
+      return closeCount;
+    }
+
+    @Override
+    public Catalog loadCatalog() {
+      return new CloseCountingCatalog(delegate.loadCatalog());
+    }
+
+    @Override
+    @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+    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<TableIdentifier> 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;
+      }
+    }
+  }
 }