From 07174b27f8eceb5fe1e8ed6bb8c4799307a1c177 Mon Sep 17 00:00:00 2001
From: javsanbel2 <javiersb1997@gmail.com>
Date: Wed, 27 Nov 2024 12:19:12 +0100
Subject: [PATCH] change expired metadata handler

---
 .../handler/ExpiredMetadataHandler.java       | 40 +++++++++----------
 1 file changed, 19 insertions(+), 21 deletions(-)

diff --git a/beekeeper-metadata-cleanup/src/main/java/com/expediagroup/beekeeper/metadata/cleanup/handler/ExpiredMetadataHandler.java b/beekeeper-metadata-cleanup/src/main/java/com/expediagroup/beekeeper/metadata/cleanup/handler/ExpiredMetadataHandler.java
index 9a0f628e..6d0a1df9 100644
--- a/beekeeper-metadata-cleanup/src/main/java/com/expediagroup/beekeeper/metadata/cleanup/handler/ExpiredMetadataHandler.java
+++ b/beekeeper-metadata-cleanup/src/main/java/com/expediagroup/beekeeper/metadata/cleanup/handler/ExpiredMetadataHandler.java
@@ -32,7 +32,7 @@
 import com.expediagroup.beekeeper.cleanup.metadata.CleanerClientFactory;
 import com.expediagroup.beekeeper.cleanup.metadata.MetadataCleaner;
 import com.expediagroup.beekeeper.cleanup.path.PathCleaner;
-import com.expediagroup.beekeeper.core.error.BeekeeperException;
+import com.expediagroup.beekeeper.core.error.BeekeeperIcebergException;
 import com.expediagroup.beekeeper.core.model.HousekeepingMetadata;
 import com.expediagroup.beekeeper.core.model.HousekeepingStatus;
 import com.expediagroup.beekeeper.core.repository.HousekeepingMetadataRepository;
@@ -78,6 +78,11 @@ public void cleanupMetadata(HousekeepingMetadata housekeepingMetadata, LocalDate
       if (deleted && !dryRunEnabled) {
         updateAttemptsAndStatus(housekeepingMetadata, DELETED);
       }
+    } catch (BeekeeperIcebergException e) {
+      updateAttemptsAndStatus(housekeepingMetadata, SKIPPED);
+      log
+          .warn("Table \"{}.{}\" is skipped because is iceberg or could not be identified ",
+              housekeepingMetadata.getDatabaseName(), housekeepingMetadata.getTableName(), e);
     } catch (Exception e) {
       updateAttemptsAndStatus(housekeepingMetadata, FAILED);
       log
@@ -114,21 +119,18 @@ private boolean cleanUpTable(CleanerClient client, HousekeepingMetadata housekee
     String tableName = housekeepingMetadata.getTableName();
     log.info("Cleaning up metadata for \"{}.{}\"", databaseName, tableName);
     if (metadataCleaner.tableExists(client, databaseName, tableName)) {
-      try {
-        metadataCleaner.dropTable(housekeepingMetadata, client);
-        pathCleaner.cleanupPath(housekeepingMetadata);
-      } catch (BeekeeperException e) {
-        log.warn("Skipping cleanup for Iceberg table \"{}.{}\": {}", databaseName, tableName, e.getMessage());
-        updateStatus(housekeepingMetadata, SKIPPED, dryRunEnabled);
-        return false;
-      }
+      metadataCleaner.dropTable(housekeepingMetadata, client);
+      pathCleaner.cleanupPath(housekeepingMetadata);
     } else {
       log.info("Cannot drop table \"{}.{}\". Table does not exist.", databaseName, tableName);
     }
     return true;
   }
 
-  private boolean cleanupPartition(CleanerClient client, HousekeepingMetadata housekeepingMetadata, boolean dryRunEnabled) {
+  private boolean cleanupPartition(
+      CleanerClient client,
+      HousekeepingMetadata housekeepingMetadata,
+      boolean dryRunEnabled) {
     if (!S3PathValidator.validPartitionPath(housekeepingMetadata.getPath())) {
       log.warn("Will not clean up partition path \"{}\" because it is not valid.", housekeepingMetadata.getPath());
       updateStatus(housekeepingMetadata, SKIPPED, dryRunEnabled);
@@ -136,20 +138,16 @@ private boolean cleanupPartition(CleanerClient client, HousekeepingMetadata hous
     }
     String databaseName = housekeepingMetadata.getDatabaseName();
     String tableName = housekeepingMetadata.getTableName();
-    log.info("Cleaning up metadata for partition \"{}\" in table \"{}.{}\"", housekeepingMetadata.getPartitionName(), databaseName, tableName);
+    log.info("Cleaning up metadata for \"{}.{}\"", databaseName, tableName);
     if (metadataCleaner.tableExists(client, databaseName, tableName)) {
-      try {
-        boolean partitionDeleted = metadataCleaner.dropPartition(housekeepingMetadata, client);
-        if (partitionDeleted) {
-          pathCleaner.cleanupPath(housekeepingMetadata);
-        }
-      } catch (BeekeeperException e) {
-        log.warn("Skipping cleanup for Iceberg partition \"{}\" in table \"{}.{}\": {}", housekeepingMetadata.getPartitionName(), databaseName, tableName, e.getMessage());
-        updateStatus(housekeepingMetadata, SKIPPED, dryRunEnabled);
-        return false;
+      boolean partitionDeleted = metadataCleaner.dropPartition(housekeepingMetadata, client);
+      if (partitionDeleted) {
+        pathCleaner.cleanupPath(housekeepingMetadata);
       }
     } else {
-      log.info("Cannot drop partition \"{}\" from table \"{}.{}\". Table does not exist.", housekeepingMetadata.getPartitionName(), databaseName, tableName);
+      log
+          .info("Cannot drop partition \"{}\" from table \"{}.{}\". Table does not exist.",
+              housekeepingMetadata.getPartitionName(), databaseName, tableName);
     }
     return true;
   }