(amoro) branch master updated: [AMORO-3110] Avoid optimization status blocking in running (#3113)

[jinsongzhou]

This is an automated email from the ASF dual-hosted git repository.

jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git


The following commit(s) were added to refs/heads/master by this push:
     new 9877c206e [AMORO-3110] Avoid optimization status blocking in running 
(#3113)
9877c206e is described below

commit 9877c206e5a44b1b274fb9084765758b10fa333f
Author: rfyu <39233058+r...@users.noreply.github.com>
AuthorDate: Sat Sep 14 16:42:00 2024 +0800

    [AMORO-3110] Avoid optimization status blocking in running (#3113)
    
    * [AMORO-3110] Avoid optimization status blocking in running
    
    * format
    
    * Ignore the exception when it occurs and do not perform a rollback
---
 .../apache/amoro/server/optimizing/OptimizingQueue.java  |  2 --
 .../apache/amoro/server/persistence/PersistentBase.java  | 16 ++++++++++++++++
 .../java/org/apache/amoro/server/table/TableRuntime.java |  2 +-
 3 files changed, 17 insertions(+), 3 deletions(-)

diff --git 
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
 
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index 6b3d92fcf..47a3f401e 100644
--- 
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++ 
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -466,8 +466,6 @@ public class OptimizingQueue extends PersistentBase {
             persistProcessCompleted(false);
           }
         }
-      } catch (Exception e) {
-        LOG.error("accept result error:", e);
       } finally {
         lock.unlock();
       }
diff --git 
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
 
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
index 8c9d00389..4a2d6fb26 100644
--- 
a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
+++ 
b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
@@ -23,6 +23,8 @@ import org.apache.amoro.server.exception.PersistenceException;
 import 
org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.ibatis.session.TransactionIsolationLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.function.Consumer;
@@ -31,6 +33,8 @@ import java.util.function.Supplier;
 
 public abstract class PersistentBase {
 
+  private static final Logger LOG = 
LoggerFactory.getLogger(PersistentBase.class);
+
   protected PersistentBase() {}
 
   @VisibleForTesting
@@ -69,6 +73,18 @@ public abstract class PersistentBase {
     }
   }
 
+  protected final <T> void doAsIgnoreError(Class<T> mapperClz, Consumer<T> 
consumer) {
+    try (NestedSqlSession session = beginSession()) {
+      try {
+        T mapper = getMapper(session, mapperClz);
+        consumer.accept(mapper);
+        session.commit();
+      } catch (Throwable t) {
+        LOG.error("Ignore error in doAsIgnoreError", t);
+      }
+    }
+  }
+
   protected final void doAsTransaction(Runnable... operations) {
     try (NestedSqlSession session = beginSession()) {
       try {
diff --git 
a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java 
b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index 73855d3a0..e70adfa80 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -413,7 +413,7 @@ public class TableRuntime extends StatedPersistentBase {
   }
 
   public void addTaskQuota(TaskRuntime.TaskQuota taskQuota) {
-    doAs(OptimizingMapper.class, mapper -> mapper.insertTaskQuota(taskQuota));
+    doAsIgnoreError(OptimizingMapper.class, mapper -> 
mapper.insertTaskQuota(taskQuota));
     taskQuotas.add(taskQuota);
     long validTime = System.currentTimeMillis() - 
AmoroServiceConstants.QUOTA_LOOK_BACK_TIME;
     this.taskQuotas.removeIf(task -> task.checkExpired(validTime));