From 9a1a57d63f1a8dfa90be970c0c461d1d8cd59137 Mon Sep 17 00:00:00 2001
From: Y Ethan Guo <ethan.guoyihua@gmail.com>
Date: Mon, 6 Mar 2023 17:26:17 -0800
Subject: [PATCH] [HUDI-5882] Close table metadata and file system view
 instances in Metadata Table Validator

---
 .../HoodieMetadataTableValidator.java         | 70 +++++++++++--------
 1 file changed, 40 insertions(+), 30 deletions(-)

diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index f56fcb638db4..dce560b720e8 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -443,41 +443,45 @@ public boolean doMetadataTableValidation() {
       return true;
     }
 
-    HoodieMetadataValidationContext metadataTableBasedContext =
-        new HoodieMetadataValidationContext(engineContext, cfg, metaClient, true);
-    HoodieMetadataValidationContext fsBasedContext =
-        new HoodieMetadataValidationContext(engineContext, cfg, metaClient, false);
+    try (HoodieMetadataValidationContext metadataTableBasedContext =
+             new HoodieMetadataValidationContext(engineContext, cfg, metaClient, true);
+         HoodieMetadataValidationContext fsBasedContext =
+             new HoodieMetadataValidationContext(engineContext, cfg, metaClient, false)) {
+      Set<String> finalBaseFilesForCleaning = baseFilesForCleaning;
+      List<Pair<Boolean, String>> result = engineContext.parallelize(allPartitions, allPartitions.size()).map(partitionPath -> {
+        try {
+          validateFilesInPartition(metadataTableBasedContext, fsBasedContext, partitionPath, finalBaseFilesForCleaning);
+          LOG.info(String.format("Metadata table validation succeeded for partition %s (partition %s)", partitionPath, taskLabels));
+          return Pair.of(true, "");
+        } catch (HoodieValidationException e) {
+          LOG.error(
+              String.format("Metadata table validation failed for partition %s due to HoodieValidationException (partition %s)",
+                  partitionPath, taskLabels), e);
+          if (!cfg.ignoreFailed) {
+            throw e;
+          }
+          return Pair.of(false, e.getMessage() + " for partition: " + partitionPath);
+        }
+      }).collectAsList();
 
-    Set<String> finalBaseFilesForCleaning = baseFilesForCleaning;
-    List<Pair<Boolean, String>> result = engineContext.parallelize(allPartitions, allPartitions.size()).map(partitionPath -> {
-      try {
-        validateFilesInPartition(metadataTableBasedContext, fsBasedContext, partitionPath, finalBaseFilesForCleaning);
-        LOG.info(String.format("Metadata table validation succeeded for partition %s (partition %s)", partitionPath, taskLabels));
-        return Pair.of(true, "");
-      } catch (HoodieValidationException e) {
-        LOG.error(
-            String.format("Metadata table validation failed for partition %s due to HoodieValidationException (partition %s)",
-                partitionPath, taskLabels), e);
-        if (!cfg.ignoreFailed) {
-          throw e;
+      for (Pair<Boolean, String> res : result) {
+        finalResult &= res.getKey();
+        if (res.getKey().equals(false)) {
+          LOG.error("Metadata Validation failed for table: " + cfg.basePath + " with error: " + res.getValue());
         }
-        return Pair.of(false, e.getMessage() + " for partition: " + partitionPath);
       }
-    }).collectAsList();
 
-    for (Pair<Boolean, String> res : result) {
-      finalResult &= res.getKey();
-      if (res.getKey().equals(false)) {
-        LOG.error("Metadata Validation failed for table: " + cfg.basePath + " with error: " + res.getValue());
+      if (finalResult) {
+        LOG.info(String.format("Metadata table validation succeeded (%s).", taskLabels));
+        return true;
+      } else {
+        LOG.warn(String.format("Metadata table validation failed (%s).", taskLabels));
+        return false;
       }
-    }
-
-    if (finalResult) {
-      LOG.info(String.format("Metadata table validation succeeded (%s).", taskLabels));
+    } catch (Exception e) {
+      LOG.warn("Error closing HoodieMetadataValidationContext, "
+          + "ignoring the error as the validation is successful.", e);
       return true;
-    } else {
-      LOG.warn(String.format("Metadata table validation failed (%s).", taskLabels));
-      return false;
     }
   }
 
@@ -1018,7 +1022,7 @@ public int compare(HoodieColumnRangeMetadata<Comparable> o1, HoodieColumnRangeMe
    * the same information regardless of whether metadata table is enabled, which is
    * verified in the {@link HoodieMetadataTableValidator}.
    */
-  private static class HoodieMetadataValidationContext implements Serializable {
+  private static class HoodieMetadataValidationContext implements AutoCloseable, Serializable {
 
     private static final Logger LOG = LogManager.getLogger(HoodieMetadataValidationContext.class);
 
@@ -1145,5 +1149,11 @@ private Option<BloomFilterData> readBloomFilterFromFile(String partitionPath, St
           .setBloomFilter(ByteBuffer.wrap(bloomFilter.serializeToString().getBytes()))
           .build());
     }
+
+    @Override
+    public void close() throws Exception {
+      tableMetadata.close();
+      fileSystemView.close();
+    }
   }
 }