[HUDI-2984] Implement #close for AbstractTableFileSystemView

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -243,26 +243,40 @@ private void resetFileGroupsReplaced(HoodieTimeline timeline) {
         + replacedFileGroups.size() + " replaced file groups");
   }
 
+  @Override
+  public void close() {
+    try {
+      writeLock.lock();
+      clear();
+    } finally {
+      writeLock.unlock();
+    }
+  }
+
   /**
    * Clears the partition Map and reset view states.
    */
   @Override
   public final void reset() {
     try {
       writeLock.lock();
-
-      addedPartitions.clear();
-      resetViewState();
-
-      bootstrapIndex = null;
-
+      clear();
       // Initialize with new Hoodie timeline.
       init(metaClient, getTimeline());
     } finally {
       writeLock.unlock();
     }
   }
 
+  /**
+   * 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.
    */

hudi-common/src/main/java/org/apache/hudi/common/table/view/HoodieTableFileSystemView.java

@@ -349,12 +349,12 @@ protected Option<HoodieInstant> getReplaceInstant(final HoodieFileGroupId fileGr
 
   @Override
   public void close() {
-    closed = true;
-    super.reset();
+    super.close();
     partitionToFileGroupsMap = null;
     fgIdToPendingCompaction = null;
     fgIdToBootstrapBaseFile = null;
     fgIdToReplaceInstants = null;
+    closed = true;
   }
 
   @Override

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 void close() {
 
   @Override
   public void reset() {
+    timeline = metaClient.reloadActiveTimeline().filterCompletedAndCompactionInstants();
     refresh();
   }
 

hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java

@@ -249,7 +249,7 @@ public synchronized void reload(long checkpointId) {
       return;
     }
     this.metaClient.reloadActiveTimeline();
-    this.fsView = getFileSystemView();
+    this.fsView.sync();
     recordProfile();
     cleanMetadataCache(this.metaClient.getCommitsTimeline().filterCompletedInstants().getInstants());
     this.smallFilesMap.clear();