open table loader if needed

apache · Sep 13, 2023 · b95e86d · b95e86d
1 parent d5f884d
commit b95e86d
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 0 deletions.
diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/TableLoader.java
@@ -38,6 +38,8 @@ public interface TableLoader extends Closeable, Serializable, Cloneable {
 
   void open();
 
+  boolean isOpen();
+
   Table loadTable();
 
   /** Clone a TableLoader */
@@ -75,6 +77,11 @@ public void open() {
       tables = new HadoopTables(hadoopConf.get());
     }
 
+    @Override
+    public boolean isOpen() {
+      return tables != null;
+    }
+
     @Override
     public Table loadTable() {
       FlinkEnvironmentContext.init();
@@ -115,6 +122,11 @@ public void open() {
       catalog = catalogLoader.loadCatalog();
     }
 
+    @Override
+    public boolean isOpen() {
+      return catalog != null;
+    }
+
     @Override
     public Table loadTable() {
       FlinkEnvironmentContext.init();
@@ -126,6 +138,7 @@ public void close() throws IOException {
       if (catalog instanceof Closeable) {
         ((Closeable) catalog).close();
       }
+      catalog = null;
     }
 
     @Override

diff --git a/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/ReloadingTableSupplier.java b/flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/ReloadingTableSupplier.java
@@ -65,6 +65,9 @@ public Table get() {
     }
 
     if (System.currentTimeMillis() > nextReloadTimeMs) {
+      if (!tableLoader.isOpen()) {
+        tableLoader.open();
+      }
       table = tableLoader.loadTable();
       nextReloadTimeMs = calcNextReloadTimeMs(System.currentTimeMillis());
       LOG.info("Table {} reloaded, next load time is at {}", table.name(), nextReloadTimeMs);