[ARCTIC-1245][AMS] If the Optimize_task is in the commit process, the…

… MySQL connection is disconnected, let it retry next time (#1249) * fix close_wait problem in limit query * Retry when commit fails due to MySQL disconnection --------- Co-authored-by: Chao He <hechao@rd.netease.com>
apache · Mar 22, 2023 · afbd1c0 · afbd1c0
1 parent 26c9d41
commit afbd1c0
Showing 1 changed file with 19 additions and 4 deletions.
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
@@ -123,6 +123,10 @@ public class TableOptimizeItem extends IJDBCService {
   private volatile double quotaCache;
   private volatile String groupNameCache;
 
+  private BasicOptimizeCommit optimizeCommit;
+  private long commitTime;
+  private static final long INIT_COMMIT_TIME = 0L;
+
   /**
    * -1: not initialized
    * 0: not committed
@@ -209,7 +213,7 @@ public void tryTriggerCommit() {
       if (waitCommit.get()) {
         return;
       }
-      if (!allTasksPrepared()) {
+      if (!allTasksPrepared() && !isRetryCommit()) {
         return;
       }
       boolean success = ServiceContainer.getOptimizeService().triggerOptimizeCommit(this);
@@ -863,7 +867,7 @@ private void optimizeTasksCommitted(BasicOptimizeCommit optimizeCommit,
 
       sqlSession.commit(true);
     }
-
+    this.commitTime = INIT_COMMIT_TIME;
     updateTableOptimizeStatus();
   }
 
@@ -1028,11 +1032,14 @@ public void commitOptimizeTasks() throws Exception {
     }
 
     try {
+      // If it is a retry task, do optimizeTasksCommitted()
+      if (isRetryCommit()) {
+        optimizeTasksCommitted(optimizeCommit, commitTime);
+      }
       Map<String, List<OptimizeTaskItem>> tasksToCommit = getOptimizeTasksToCommit();
       long taskCount = tasksToCommit.values().stream().mapToLong(Collection::size).sum();
       if (MapUtils.isNotEmpty(tasksToCommit)) {
         LOG.info("{} get {} tasks of {} partitions to commit", tableIdentifier, taskCount, tasksToCommit.size());
-        BasicOptimizeCommit optimizeCommit;
         if (com.netease.arctic.utils.TableTypeUtil.isIcebergTableFormat(getArcticTable())) {
           optimizeCommit = new IcebergOptimizeCommit(getArcticTable(true), tasksToCommit);
         } else if (TableTypeUtil.isHive(getArcticTable())) {
@@ -1044,7 +1051,7 @@ public void commitOptimizeTasks() throws Exception {
 
         boolean committed = optimizeCommit.commit(tableOptimizeRuntime.getCurrentSnapshotId());
         if (committed) {
-          long commitTime = System.currentTimeMillis();
+          commitTime = System.currentTimeMillis();
           optimizeTasksCommitted(optimizeCommit, commitTime);
         } else {
           LOG.warn("{} commit failed, clear optimize tasks", tableIdentifier);
@@ -1058,6 +1065,14 @@ public void commitOptimizeTasks() throws Exception {
     }
   }
 
+  /**
+   * Check if the task currently being submitted needs to be retried
+   * @return boolean
+   */
+  private boolean isRetryCommit() {
+    return commitTime != INIT_COMMIT_TIME;
+  }
+
   /**
    * Get all optimize tasks.
    *