[ARCTIC-1737] Fix the tasks of ScheduledExecutorService that halts …

…when an exception or error is thrown (apache#1739) add try-catch for scheduleWithFixedDelay
shendanfengg · Jul 26, 2023 · 7665174 · 7665174
1 parent 9f005a0
commit 7665174
Show file tree

Hide file tree

Showing 5 changed files with 78 additions and 55 deletions.
diff --git a/...rver/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java b/...rver/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
@@ -28,6 +28,8 @@
 import com.netease.arctic.ams.server.service.ServiceContainer;
 import com.netease.arctic.optimizer.Optimizer;
 import com.netease.arctic.optimizer.factory.OptimizerFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -111,23 +113,28 @@ public OptimizerFactory findOptimizerFactory(String type) throws NoSuchObjectExc
 
   @Override
   public void close() throws IOException {
-    
+
   }
 
   public static class OptimizerMonitor {
+    private static final Logger LOG = LoggerFactory.getLogger(OptimizerMonitor.class);
 
     private static final long OPTIMIZER_JOB_TIMEOUT = 10 * 60 * 1000;
 
     public void monitorStatus() {
-      long currentTime = System.currentTimeMillis();
-      List<com.netease.arctic.ams.server.model.Optimizer> optimizers =
-          ServiceContainer.getOptimizerService().getOptimizers();
-      optimizers.forEach(optimizer -> {
-        if ((currentTime - optimizer.getUpdateTime().getTime()) > OPTIMIZER_JOB_TIMEOUT) {
-          ServiceContainer.getOptimizerService()
-              .updateOptimizerStatus(optimizer.getJobId(), TableTaskStatus.FAILED);
-        }
-      });
+      try {
+        long currentTime = System.currentTimeMillis();
+        List<com.netease.arctic.ams.server.model.Optimizer> optimizers =
+            ServiceContainer.getOptimizerService().getOptimizers();
+        optimizers.forEach(optimizer -> {
+          if ((currentTime - optimizer.getUpdateTime().getTime()) > OPTIMIZER_JOB_TIMEOUT) {
+            ServiceContainer.getOptimizerService()
+                .updateOptimizerStatus(optimizer.getJobId(), TableTaskStatus.FAILED);
+          }
+        });
+      } catch (Throwable t) {
+        LOG.error("monitor optimizer status failed", t);
+      }
     }
   }
 }

diff --git a/...ver/src/main/java/com/netease/arctic/ams/server/service/impl/OrphanFilesCleanService.java b/...ver/src/main/java/com/netease/arctic/ams/server/service/impl/OrphanFilesCleanService.java
@@ -75,18 +75,22 @@ public class OrphanFilesCleanService implements IOrphanFilesCleanService {
 
   @Override
   public synchronized void checkOrphanFilesCleanTasks() {
-    LOG.info("Schedule Orphan Cleaner");
-    if (cleanTasks == null) {
-      cleanTasks = new ScheduledTasks<>(ThreadPool.Type.ORPHAN);
-    }
+    try {
+      LOG.info("Schedule Orphan Cleaner");
+      if (cleanTasks == null) {
+        cleanTasks = new ScheduledTasks<>(ThreadPool.Type.ORPHAN);
+      }
 
-    Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
-    cleanTasks.checkRunningTask(tableIds,
-        () -> 0L,
-        () -> CHECK_INTERVAL,
-        TableOrphanFileClean::new,
-        true);
-    LOG.info("Schedule Orphan Cleaner finished with {} tasks", tableIds.size());
+      Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
+      cleanTasks.checkRunningTask(tableIds,
+          () -> 0L,
+          () -> CHECK_INTERVAL,
+          TableOrphanFileClean::new,
+          true);
+      LOG.info("Schedule Orphan Cleaner finished with {} tasks", tableIds.size());
+    } catch (Throwable t) {
+      LOG.error("unexpected error when checkOrphanFilesCleanTasks", t);
+    }
   }
 
   @Override

diff --git a/...rver/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java b/...rver/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java
@@ -65,19 +65,23 @@ public class SupportHiveSyncService implements ISupportHiveSyncService {
 
   @Override
   public void checkHiveSyncTasks() {
-    LOG.info("Schedule Support Hive Sync");
-    if (syncTasks == null) {
-      syncTasks = new ScheduledTasks<>(ThreadPool.Type.HIVE_SYNC);
+    try {
+      LOG.info("Schedule Support Hive Sync");
+      if (syncTasks == null) {
+        syncTasks = new ScheduledTasks<>(ThreadPool.Type.HIVE_SYNC);
+      }
+      List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
+      Set<TableIdentifier> ids =
+          tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
+      syncTasks.checkRunningTask(ids,
+          () -> 0L,
+          () -> SYNC_INTERVAL,
+          SupportHiveSyncService.SupportHiveSyncTask::new,
+          false);
+      LOG.info("Schedule Support Hive Sync finished with {} valid ids", ids.size());
+    } catch (Throwable t) {
+      LOG.error("Schedule Support Hive Sync failed", t);
     }
-    List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
-    Set<TableIdentifier> ids =
-        tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
-    syncTasks.checkRunningTask(ids,
-        () -> 0L,
-        () -> SYNC_INTERVAL,
-        SupportHiveSyncService.SupportHiveSyncTask::new,
-        false);
-    LOG.info("Schedule Support Hive Sync finished with {} valid ids", ids.size());
   }
 
   @Override

diff --git a/...s-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java b/...s-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
@@ -75,18 +75,22 @@ public class TableExpireService implements ITableExpireService {
 
   @Override
   public synchronized void checkTableExpireTasks() {
-    LOG.info("Schedule Expired Cleaner");
-    if (cleanTasks == null) {
-      cleanTasks = new ScheduledTasks<>(ThreadPool.Type.EXPIRE);
-    }
+    try {
+      LOG.info("Schedule Expired Cleaner");
+      if (cleanTasks == null) {
+        cleanTasks = new ScheduledTasks<>(ThreadPool.Type.EXPIRE);
+      }
 
-    Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
-    cleanTasks.checkRunningTask(tableIds,
-        () -> 0L,
-        () -> EXPIRE_INTERVAL,
-        TableExpireTask::new,
-        false);
-    LOG.info("Schedule Expired Cleaner finished with {} valid ids", tableIds.size());
+      Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
+      cleanTasks.checkRunningTask(tableIds,
+          () -> 0L,
+          () -> EXPIRE_INTERVAL,
+          TableExpireTask::new,
+          false);
+      LOG.info("Schedule Expired Cleaner finished with {} valid ids", tableIds.size());
+    } catch (Throwable t) {
+      LOG.error("unexpected error when checkTableExpireTasks", t);
+    }
   }
 
   @Override

diff --git a/...ms-server/src/main/java/com/netease/arctic/ams/server/service/impl/TrashCleanService.java b/...ms-server/src/main/java/com/netease/arctic/ams/server/service/impl/TrashCleanService.java
@@ -50,18 +50,22 @@ public class TrashCleanService implements Closeable {
   private ScheduledTasks<TableIdentifier, TableTrashCleanTask> cleanTasks;
 
   public synchronized void checkTrashCleanTasks() {
-    LOG.info("Schedule Trash Cleaner");
-    if (cleanTasks == null) {
-      cleanTasks = new ScheduledTasks<>(ThreadPool.Type.TRASH_CLEAN);
-    }
+    try {
+      LOG.info("Schedule Trash Cleaner");
+      if (cleanTasks == null) {
+        cleanTasks = new ScheduledTasks<>(ThreadPool.Type.TRASH_CLEAN);
+      }
 
-    Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
-    cleanTasks.checkRunningTask(tableIds,
-        () -> RandomUtils.nextLong(0, CHECK_INTERVAL),
-        () -> CHECK_INTERVAL,
-        TableTrashCleanTask::new,
-        true);
-    LOG.info("Schedule Trash Cleaner finished with {} tasks", tableIds.size());
+      Set<TableIdentifier> tableIds = CatalogUtil.loadTablesFromCatalog();
+      cleanTasks.checkRunningTask(tableIds,
+          () -> RandomUtils.nextLong(0, CHECK_INTERVAL),
+          () -> CHECK_INTERVAL,
+          TableTrashCleanTask::new,
+          true);
+      LOG.info("Schedule Trash Cleaner finished with {} tasks", tableIds.size());
+    } catch (Throwable t) {
+      LOG.error("Schedule Trash Cleaner unexpected error", t);
+    }
   }
 
   @Override