You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by ji...@apache.org on 2024/10/11 07:55:53 UTC
(amoro) branch 0.7.x updated: [AMORO-3110] Avoid optimization status blocking in running (#3113)
This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch 0.7.x
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/0.7.x by this push:
new f1667cbc3 [AMORO-3110] Avoid optimization status blocking in running (#3113)
f1667cbc3 is described below
commit f1667cbc3b4972b5bc35deaaf51cd3bd554ba63a
Author: rfyu <39...@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
(cherry picked from commit 9877c206e5a44b1b274fb9084765758b10fa333f)
Signed-off-by: zhoujinsong <zh...@163.com>
---
.../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/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index fdab892c8..e34a7cebe 100644
--- a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++ b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -477,8 +477,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/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
index 07e688793..cfb932dab 100644
--- a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/persistence/PersistentBase.java
+++ b/amoro-ams/amoro-ams-server/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
@@ -55,6 +59,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/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index 6e8e0c1d0..490a8fbaa 100644
--- a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++ b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -364,7 +364,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));