[fix](cloud-mow) Add retry when calculating delete bitmap timeout whe…

…n loading data (apache#40562)

Add retry when calculating delete bitmap timeout on broker load , like
stream load doing.

be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp

@@ -95,6 +95,7 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
     }
     // wait for all finished
     token->wait();
+    DBUG_EXECUTE_IF("CloudEngineCalcDeleteBitmapTask.execute.enable_wait", { sleep(3); });
 
     LOG(INFO) << "finish to calculate delete bitmap on transaction."
               << "transaction_id=" << transaction_id << ", cost(us): " << watch.get_elapse_time_us()

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2984,7 +2984,7 @@ public static int metaServiceRpcRetryTimes() {
     public static String security_checker_class_name = "";
 
     @ConfField(mutable = true)
-    public static int mow_insert_into_commit_retry_times = 10;
+    public static int mow_calculate_delete_bitmap_retry_times = 10;
 
     @ConfField(mutable = true, description = {"指定S3 Load endpoint白名单, 举例: s3_load_endpoint_white_list=a,b,c",
             "the white list for the s3 load endpoint, if it is empty, no white list will be set,"

fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java

@@ -930,7 +930,28 @@ private void debugCalcDeleteBitmapRandomTimeout() throws UserException {
     public boolean commitAndPublishTransaction(DatabaseIf db, List<Table> tableList, long transactionId,
                                                List<TabletCommitInfo> tabletCommitInfos, long timeoutMillis)
             throws UserException {
-        return commitAndPublishTransaction(db, tableList, transactionId, tabletCommitInfos, timeoutMillis, null);
+        int retryTimes = 0;
+        boolean res = false;
+        while (true) {
+            try {
+                res = commitAndPublishTransaction(db, tableList, transactionId, tabletCommitInfos, timeoutMillis, null);
+                break;
+            } catch (UserException e) {
+                LOG.warn("failed to commit txn, txnId={},retryTimes={},exception={}",
+                        transactionId, retryTimes, e);
+                // only mow table will catch DELETE_BITMAP_LOCK_ERR and need to retry
+                if (e.getErrorCode() == InternalErrorCode.DELETE_BITMAP_LOCK_ERR) {
+                    retryTimes++;
+                    if (retryTimes >= Config.mow_calculate_delete_bitmap_retry_times) {
+                        // should throw exception after running out of retry times
+                        throw e;
+                    }
+                } else {
+                    throw e;
+                }
+            }
+        }
+        return res;
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/common/util/MetaLockUtils.java

@@ -127,8 +127,14 @@ public static void writeUnlockTables(List<? extends TableIf> tableList) {
     }
 
     public static void commitLockTables(List<Table> tableList) {
-        for (Table table : tableList) {
-            table.commitLock();
+        for (int i = 0; i < tableList.size(); i++) {
+            try {
+                tableList.get(i).commitLock();
+            } catch (Exception e) {
+                for (int j = i - 1; j >= 0; j--) {
+                    tableList.get(i).commitUnlock();
+                }
+            }
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java

@@ -29,6 +29,7 @@
 import org.apache.doris.common.DataQualityException;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.DuplicatedRequestException;
+import org.apache.doris.common.InternalErrorCode;
 import org.apache.doris.common.LabelAlreadyUsedException;
 import org.apache.doris.common.MetaNotFoundException;
 import org.apache.doris.common.QuotaExceedException;
@@ -335,42 +336,59 @@ private void onLoadingTaskFinished(BrokerLoadingTaskAttachment attachment) {
         }
         Database db = null;
         List<Table> tableList = null;
-        try {
-            db = getDb();
-            tableList = db.getTablesOnIdOrderOrThrowException(Lists.newArrayList(fileGroupAggInfo.getAllTableIds()));
-            if (Config.isCloudMode()) {
-                MetaLockUtils.commitLockTables(tableList);
-            } else {
-                MetaLockUtils.writeLockTablesOrMetaException(tableList);
+        int retryTimes = 0;
+        while (true) {
+            try {
+                db = getDb();
+                tableList = db.getTablesOnIdOrderOrThrowException(
+                        Lists.newArrayList(fileGroupAggInfo.getAllTableIds()));
+                if (Config.isCloudMode()) {
+                    MetaLockUtils.commitLockTables(tableList);
+                } else {
+                    MetaLockUtils.writeLockTablesOrMetaException(tableList);
+                }
+            } catch (MetaNotFoundException e) {
+                LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
+                        .add("database_id", dbId)
+                        .add("error_msg", "db has been deleted when job is loading")
+                        .build(), e);
+                cancelJobWithoutCheck(new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, e.getMessage()), true, true);
+                return;
             }
-        } catch (MetaNotFoundException e) {
-            LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
-                    .add("database_id", dbId)
-                    .add("error_msg", "db has been deleted when job is loading")
-                    .build(), e);
-            cancelJobWithoutCheck(new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, e.getMessage()), true, true);
-            return;
-        }
-        try {
-            LOG.info(new LogBuilder(LogKey.LOAD_JOB, id)
-                    .add("txn_id", transactionId)
-                    .add("msg", "Load job try to commit txn")
-                    .build());
-            Env.getCurrentGlobalTransactionMgr().commitTransaction(
-                    dbId, tableList, transactionId, commitInfos, getLoadJobFinalOperation());
-            afterLoadingTaskCommitTransaction(tableList);
-            afterCommit();
-        } catch (UserException e) {
-            LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
-                    .add("database_id", dbId)
-                    .add("error_msg", "Failed to commit txn with error:" + e.getMessage())
-                    .build(), e);
-            cancelJobWithoutCheck(new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, e.getMessage()), true, true);
-        } finally {
-            if (Config.isCloudMode()) {
-                MetaLockUtils.commitUnlockTables(tableList);
-            } else {
-                MetaLockUtils.writeUnlockTables(tableList);
+            try {
+                LOG.info(new LogBuilder(LogKey.LOAD_JOB, id)
+                        .add("txn_id", transactionId)
+                        .add("msg", "Load job try to commit txn")
+                        .build());
+                Env.getCurrentGlobalTransactionMgr().commitTransaction(
+                        dbId, tableList, transactionId, commitInfos, getLoadJobFinalOperation());
+                afterLoadingTaskCommitTransaction(tableList);
+                afterCommit();
+                return;
+            } catch (UserException e) {
+                LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
+                        .add("database_id", dbId)
+                        .add("retry_times", retryTimes)
+                        .add("error_msg", "Failed to commit txn with error:" + e.getMessage())
+                        .build(), e);
+                if (e.getErrorCode() == InternalErrorCode.DELETE_BITMAP_LOCK_ERR) {
+                    retryTimes++;
+                    if (retryTimes >= Config.mow_calculate_delete_bitmap_retry_times) {
+                        LOG.warn("cancelJob {} because up to max retry time,exception {}", id, e);
+                        cancelJobWithoutCheck(new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, e.getMessage()), true,
+                                true);
+                        return;
+                    }
+                } else {
+                    cancelJobWithoutCheck(new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, e.getMessage()), true, true);
+                    return;
+                }
+            } finally {
+                if (Config.isCloudMode()) {
+                    MetaLockUtils.commitUnlockTables(tableList);
+                } else {
+                    MetaLockUtils.writeUnlockTables(tableList);
+                }
             }
         }
     }

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/SparkLoadJob.java

@@ -45,9 +45,11 @@
 import org.apache.doris.catalog.Tablet;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DataQualityException;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.DuplicatedRequestException;
+import org.apache.doris.common.InternalErrorCode;
 import org.apache.doris.common.LabelAlreadyUsedException;
 import org.apache.doris.common.LoadException;
 import org.apache.doris.common.MetaNotFoundException;
@@ -656,21 +658,50 @@ public void updateLoadingStatus() throws UserException {
     }
 
     private void tryCommitJob() throws UserException {
-        LOG.info(new LogBuilder(LogKey.LOAD_JOB, id).add("txn_id", transactionId)
-                .add("msg", "Load job try to commit txn").build());
-        Database db = getDb();
-        List<Table> tableList = db.getTablesOnIdOrderOrThrowException(
-                Lists.newArrayList(tableToLoadPartitions.keySet()));
-        MetaLockUtils.writeLockTablesOrMetaException(tableList);
-        try {
-            Env.getCurrentGlobalTransactionMgr().commitTransaction(
-                    dbId, tableList, transactionId, commitInfos,
-                    new LoadJobFinalOperation(id, loadingStatus, progress, loadStartTimestamp,
-                                              finishTimestamp, state, failMsg));
-        } catch (TabletQuorumFailedException e) {
-            // retry in next loop
-        } finally {
-            MetaLockUtils.writeUnlockTables(tableList);
+        int retryTimes = 0;
+        while (true) {
+            Database db = getDb();
+            List<Table> tableList = db.getTablesOnIdOrderOrThrowException(
+                    Lists.newArrayList(tableToLoadPartitions.keySet()));
+            if (Config.isCloudMode()) {
+                MetaLockUtils.commitLockTables(tableList);
+            } else {
+                MetaLockUtils.writeLockTablesOrMetaException(tableList);
+            }
+            try {
+                LOG.info(new LogBuilder(LogKey.LOAD_JOB, id).add("txn_id", transactionId)
+                        .add("msg", "Load job try to commit txn").build());
+                Env.getCurrentGlobalTransactionMgr().commitTransaction(
+                        dbId, tableList, transactionId, commitInfos,
+                        new LoadJobFinalOperation(id, loadingStatus, progress, loadStartTimestamp,
+                                finishTimestamp, state, failMsg));
+                return;
+            } catch (TabletQuorumFailedException e) {
+                // retry in next loop
+                return;
+            } catch (UserException e) {
+                LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
+                        .add("txn_id", transactionId)
+                        .add("database_id", dbId)
+                        .add("retry_times", retryTimes)
+                        .add("error_msg", "Failed to commit txn with error:" + e.getMessage())
+                        .build(), e);
+                if (e.getErrorCode() == InternalErrorCode.DELETE_BITMAP_LOCK_ERR) {
+                    retryTimes++;
+                    if (retryTimes >= Config.mow_calculate_delete_bitmap_retry_times) {
+                        LOG.warn("cancelJob {} because up to max retry time, exception {}", id, e);
+                        throw e;
+                    }
+                } else {
+                    throw e;
+                }
+            } finally {
+                if (Config.isCloudMode()) {
+                    MetaLockUtils.commitUnlockTables(tableList);
+                } else {
+                    MetaLockUtils.writeUnlockTables(tableList);
+                }
+            }
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java

@@ -24,7 +24,6 @@
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeConstants;
-import org.apache.doris.common.InternalErrorCode;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.nereids.NereidsPlanner;
@@ -193,25 +192,7 @@ public void executeSingleInsert(StmtExecutor executor, long jobId) throws Except
             executor.updateProfile(false);
             execImpl(executor, jobId);
             checkStrictModeAndFilterRatio();
-            int retryTimes = 0;
-            while (true) {
-                try {
-                    onComplete();
-                    break;
-                } catch (UserException e) {
-                    LOG.warn("failed to commit txn, txnId={}, jobId={}, retryTimes={}",
-                            getTxnId(), jobId, retryTimes, e);
-                    if (e.getErrorCode() == InternalErrorCode.DELETE_BITMAP_LOCK_ERR) {
-                        retryTimes++;
-                        if (retryTimes >= Config.mow_insert_into_commit_retry_times) {
-                            // should throw exception after running out of retry times
-                            throw e;
-                        }
-                    } else {
-                        throw e;
-                    }
-                }
-            }
+            onComplete();
         } catch (Throwable t) {
             onFail(t);
             // retry insert into from select when meet E-230 in cloud

regression-test/data/fault_injection_p0/cloud/test_cloud_mow_broker_load_with_retry.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+19
+
+-- !select --
+19
+

regression-test/data/fault_injection_p0/cloud/test_cloud_mow_insert_with_retry.out

@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	1	1
+2	2	2
+
+-- !sql --
+1	1	1
+
+-- !sql --
+1	1	1
+2	2	2
+
+-- !sql --
+1	1	1
+