From 311582001f92a5d7c9940377ee1233385b6240cd Mon Sep 17 00:00:00 2001
From: meegoo <hujie-dlut@qq.com>
Date: Mon, 16 May 2022 21:43:20 +0800
Subject: [PATCH] [Enhancement] Optimize quorum publish, wait some time util
 all backends finish

---
 .../src/main/java/com/starrocks/common/Config.java   |  6 ++++++
 .../transaction/DatabaseTransactionMgr.java          | 12 +++++++++++-
 2 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/fe/fe-core/src/main/java/com/starrocks/common/Config.java b/fe/fe-core/src/main/java/com/starrocks/common/Config.java
index c9e52ece6ef91..bda337d1d76c2 100644
--- a/fe/fe-core/src/main/java/com/starrocks/common/Config.java
+++ b/fe/fe-core/src/main/java/com/starrocks/common/Config.java
@@ -1376,4 +1376,10 @@ public class Config extends ConfigBase {
      */
     @ConfField(mutable = true)
     public static long min_routine_load_lag_for_metrics = 10000;
+
+    /**
+     * after wait quorom_publish_wait_time_ms, will do quorum publish
+     */
+    @ConfField(mutable = true)
+    public static int quorom_publish_wait_time_ms = 500;
 }
diff --git a/fe/fe-core/src/main/java/com/starrocks/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/com/starrocks/transaction/DatabaseTransactionMgr.java
index bcb1caae93708..294592b3e873f 100644
--- a/fe/fe-core/src/main/java/com/starrocks/transaction/DatabaseTransactionMgr.java
+++ b/fe/fe-core/src/main/java/com/starrocks/transaction/DatabaseTransactionMgr.java
@@ -644,7 +644,7 @@ public boolean canTxnFinished(TransactionState txn, Set<Long> errReplicas, Set<L
             return true;
         }
         db.readLock();
-
+        long currentTs = System.currentTimeMillis();
         try {
             // check each table involved in transaction
             for (TableCommitInfo tableCommitInfo : txn.getIdToTableCommitInfos().values()) {
@@ -670,6 +670,7 @@ public boolean canTxnFinished(TransactionState txn, Set<Long> errReplicas, Set<L
 
                     List<MaterializedIndex> allIndices = txn.getPartitionLoadedTblIndexes(tableId, partition);
                     int quorumNum = partitionInfo.getQuorumNum(partitionId);
+                    int replicaNum = partitionInfo.getReplicationNum(partitionId);
                     for (MaterializedIndex index : allIndices) {
                         for (Tablet tablet : index.getTablets()) {
                             int successHealthyReplicaNum = 0;
@@ -705,6 +706,15 @@ public boolean canTxnFinished(TransactionState txn, Set<Long> errReplicas, Set<L
                             if (successHealthyReplicaNum < quorumNum) {
                                 return false;
                             }
+                            // quorum publish will make table unstable
+                            // so that we wait quorom_publish_wait_time_ms util all backend publish finish
+                            // before quorum publish
+                            if (successHealthyReplicaNum != replicaNum
+                                    && !unfinishedBackends.isEmpty()
+                                    && currentTs
+                                            - txn.getCommitTime() < Config.quorom_publish_wait_time_ms) {
+                                return false;
+                            }
                         }
                     }
                 }