From 8dd0444ef96db4028bd28d5e4d8653c290764152 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Sat, 11 Dec 2021 16:19:10 +0800
Subject: [PATCH] [HUDI-2984] Implement #close for AbstractTableFileSystemView
 (#4285)

---
 .../view/AbstractTableFileSystemView.java     | 26 ++++++++++++++-----
 .../table/view/HoodieTableFileSystemView.java |  4 +--
 .../view/RemoteHoodieTableFileSystemView.java |  3 ++-
 .../partitioner/profile/WriteProfile.java     |  2 +-
 4 files changed, 25 insertions(+), 10 deletions(-)

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index 1a2ce01d8..d946cff95 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -243,6 +243,16 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
         + replacedFileGroups.size() + " replaced file groups");
   }
 
+  @Override
+  public void close() {
+    try {
+      writeLock.lock();
+      clear();
+    } finally {
+      writeLock.unlock();
+    }
+  }
+
   /**
    * Clears the partition Map and reset view states.
    */
@@ -250,12 +260,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
   public final void reset() {
     try {
       writeLock.lock();
-
-      addedPartitions.clear();
-      resetViewState();
-
-      bootstrapIndex = null;
-
+      clear();
       // Initialize with new Hoodie timeline.
       init(metaClient, getTimeline());
     } finally {
@@ -263,6 +268,15 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
     }
   }
 
+  /**
+   * Clear the resource.
+   */
+  private void clear() {
+    addedPartitions.clear();
+    resetViewState();
+    bootstrapIndex = null;
+  }
+
   /**
    * Allows all view metadata in file system view storage to be reset by subclasses.
    */
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java
index 0bbbf3e9e..b8f7f313a 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java
@@ -349,12 +349,12 @@ public class HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystem
 
   @Override
   public void close() {
-    closed = true;
-    super.reset();
+    super.close();
     partitionToFileGroupsMap = null;
     fgIdToPendingCompaction = null;
     fgIdToBootstrapBaseFile = null;
     fgIdToReplaceInstants = null;
+    closed = true;
   }
 
   @Override
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java
index 4c2153010..c3a3847fd 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java
@@ -126,7 +126,7 @@ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView,
   private final int serverPort;
   private final String basePath;
   private final HoodieTableMetaClient metaClient;
-  private final HoodieTimeline timeline;
+  private HoodieTimeline timeline;
   private final ObjectMapper mapper;
   private final int timeoutSecs;
 
@@ -450,6 +450,7 @@ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView,
 
   @Override
   public void reset() {
+    timeline = metaClient.reloadActiveTimeline().filterCompletedAndCompactionInstants();
     refresh();
   }
 
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
index f6840e5ef..84fcd03f0 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
@@ -249,7 +249,7 @@ public class WriteProfile {
       return;
     }
     this.metaClient.reloadActiveTimeline();
-    this.fsView = getFileSystemView();
+    this.fsView.sync();
     recordProfile();
     cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstants());
     this.smallFilesMap.clear();