add

Signed-off-by: gengjun-git <gengjun@starrocks.com>

Author: gengjun-git

fe/fe-core/src/main/java/com/starrocks/transaction/DatabaseTransactionMgr.java

@@ -501,7 +501,13 @@ public VisibleStateWaiter commitPreparedTransaction(long transactionId) throws S
             Span unprotectedCommitSpan = TraceManager.startSpan("unprotectedCommitPreparedTransaction", txnSpan);
 
             // transaction state transform
-            boolean txnOperated = unprotectedCommitPreparedTransaction(copiedState, db);
+            boolean txnOperated;
+            writeLock();
+            try {
+                txnOperated = unprotectedCommitPreparedTransaction(copiedState, db);
+            } finally {
+                writeUnlock();
+            }
             if (!txnOperated) {
                 return null;
             }
@@ -1115,7 +1121,7 @@ public boolean canTxnFinished(TransactionState txn, Set<Long> errReplicas, Set<L
         return true;
     }
 
-    public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, long lockTimeoutMs)
+    public TransactionState finishTransaction(long transactionId, Set<Long> errorReplicaIds, long lockTimeoutMs)
             throws StarRocksException {
         TransactionState transactionState = getTransactionState(transactionId);
         // add all commit errors and publish errors to a single set
@@ -1145,7 +1151,7 @@ public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, lon
                     }
                 });
 
-                return;
+                return copiedState;
             } finally {
                 transactionState.writeUnlock();
             }
@@ -1224,8 +1230,10 @@ public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, lon
                                             "wait for publishing partition %d version %d. self version: %d. table %d",
                                             physicalPartitionId, physicalPartition.getVisibleVersion() + 1,
                                             partitionCommitInfo.getVersion(), tableId);
-                            copiedState.setErrorMsg(errMsg);
-                            return;
+                            // set errMsg to transactionState instead of copiedState,
+                            // because copiedState will not be upserted in this case.
+                            transactionState.setErrorMsg(errMsg);
+                            return transactionState;
                         }
 
                         if (table.isCloudNativeTableOrMaterializedView()) {
@@ -1309,7 +1317,9 @@ public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, lon
                                             tablet.getId(), healthReplicaNum, quorumReplicaNum, tableId,
                                             physicalPartitionId,
                                             physicalPartition.getVisibleVersion() + 1);
-                                    copiedState.setErrorMsg(errMsg);
+                                    // set errMsg to transactionState instead of copiedState,
+                                    // because copiedState will not be upserted in this case.
+                                    transactionState.setErrorMsg(errMsg);
                                     hasError = true;
                                 }
 
@@ -1344,7 +1354,7 @@ public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, lon
                                     "version not equal to partition commit version or commit version - 1 if it's not a " +
                                     "upgrade stage, its a fatal error. ",
                             copiedState);
-                    return;
+                    return transactionState;
                 }
                 copiedState.setErrorReplicas(errorReplicaIds);
                 copiedState.setFinishTime(System.currentTimeMillis());
@@ -1397,6 +1407,7 @@ public void finishTransaction(long transactionId, Set<Long> errorReplicaIds, lon
         GlobalStateMgr.getCurrentState().getLocalMetastore().handleMVRepair(copiedState);
         LOG.info("finish transaction {} successfully", copiedState);
         updateTransactionMetrics(copiedState);
+        return copiedState;
     }
 
     protected boolean unprotectedCommitPreparedTransaction(TransactionState transactionState, Database db) {
@@ -1405,7 +1416,7 @@ protected boolean unprotectedCommitPreparedTransaction(TransactionState transact
             return false;
         }
         // commit timestamps needs to be strictly monotonically increasing
-        long commitTs = Math.max(System.currentTimeMillis(), maxCommitTs + 1);
+        long commitTs = reserveCommitTs();
         transactionState.setCommitTime(commitTs);
         // update transaction state version
         transactionState.setTransactionStatus(TransactionStatus.COMMITTED);
@@ -1518,6 +1529,13 @@ protected boolean unprotectedCommitPreparedTransaction(TransactionState transact
         return true;
     }
 
+    private long reserveCommitTs() {
+        Preconditions.checkState(transactionLock.isWriteLockedByCurrentThread());
+        long commitTs = Math.max(System.currentTimeMillis(), maxCommitTs + 1);
+        maxCommitTs = commitTs;
+        return commitTs;
+    }
+
     // for add/update/delete TransactionState
     protected void unprotectUpsertTransactionState(TransactionState transactionState) {
         // it's OK if getCommitTime() returns -1
@@ -2052,7 +2070,7 @@ GlobalStateMgr getGlobalStateMgr() {
         return globalStateMgr;
     }
 
-    public void finishTransactionNew(TransactionState transactionState, Set<Long> publishErrorReplicas)
+    public TransactionState finishTransactionNew(TransactionState transactionState, Set<Long> publishErrorReplicas)
             throws StarRocksException {
         Database db = globalStateMgr.getLocalMetastore().getDb(transactionState.getDbId());
         if (db == null) {
@@ -2072,7 +2090,7 @@ public void finishTransactionNew(TransactionState transactionState, Set<Long> pu
                     }
                 });
 
-                return;
+                return copiedState;
             } finally {
                 transactionState.writeUnlock();
             }
@@ -2133,6 +2151,7 @@ public void finishTransactionNew(TransactionState transactionState, Set<Long> pu
         GlobalStateMgr.getCurrentState().getLocalMetastore().handleMVRepair(copiedState);
         LOG.info("finish transaction {} successfully", copiedState);
         updateTransactionMetrics(copiedState);
+        return copiedState;
     }
 
     // only for test
@@ -2194,7 +2213,7 @@ private boolean isTxnStateBatchConsistent(Database db, TransactionStateBatch sta
     }
 
 
-    public void finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> errorReplicaIds) {
+    public TransactionStateBatch finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> errorReplicaIds) {
         Database db = globalStateMgr.getLocalMetastore().getDb(stateBatch.getDbId());
         if (db == null) {
             stateBatch.writeLock();
@@ -2211,7 +2230,7 @@ public void finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> e
                         writeUnlock();
                     }
                 });
-                return;
+                return copiedStateBatch;
             } finally {
                 stateBatch.writeUnlock();
             }
@@ -2231,7 +2250,7 @@ public void finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> e
                 copiedStateBatch = new TransactionStateBatch(stateBatch);
                 // check whether version is consistent
                 if (!isTxnStateBatchConsistent(db, copiedStateBatch)) {
-                    return;
+                    return stateBatch;
                 }
 
                 copiedStateBatch.setTransactionVisibleInfo();
@@ -2264,6 +2283,7 @@ public void finishTransactionBatch(TransactionStateBatch stateBatch, Set<Long> e
         for (TransactionState transactionState : copiedStateBatch.getTransactionStates()) {
             updateTransactionMetrics(transactionState);
         }
+        return copiedStateBatch;
     }
 
     private void updateTransactionMetrics(TransactionState txnState) {

fe/fe-core/src/main/java/com/starrocks/transaction/GlobalTransactionMgr.java

@@ -620,27 +620,30 @@ public boolean existCommittedTxns(Long dbId, Long tableId, Long partitionId) {
      * @param errorReplicaIds
      * @return
      */
-    public void finishTransaction(long dbId, long transactionId, Set<Long> errorReplicaIds) throws StarRocksException {
+    public TransactionState finishTransaction(long dbId, long transactionId, Set<Long> errorReplicaIds)
+            throws StarRocksException {
         DatabaseTransactionMgr dbTransactionMgr = getDatabaseTransactionMgr(dbId);
-        dbTransactionMgr.finishTransaction(transactionId, errorReplicaIds, 0L);
+        return dbTransactionMgr.finishTransaction(transactionId, errorReplicaIds, 0L);
     }
 
-    public void finishTransaction(long dbId, long transactionId, Set<Long> errorReplicaIds, long lockTimeoutMs)
+    public TransactionState finishTransaction(long dbId, long transactionId, Set<Long> errorReplicaIds,
+                                              long lockTimeoutMs)
             throws StarRocksException {
         DatabaseTransactionMgr dbTransactionMgr = getDatabaseTransactionMgr(dbId);
-        dbTransactionMgr.finishTransaction(transactionId, errorReplicaIds, lockTimeoutMs);
+        return dbTransactionMgr.finishTransaction(transactionId, errorReplicaIds, lockTimeoutMs);
     }
 
-    public void finishTransactionBatch(long dbId, TransactionStateBatch stateBatch, Set<Long> errorReplicaIds)
+    public TransactionStateBatch finishTransactionBatch(long dbId, TransactionStateBatch stateBatch,
+                                                        Set<Long> errorReplicaIds)
             throws StarRocksException {
         DatabaseTransactionMgr dbTransactionMgr = getDatabaseTransactionMgr(dbId);
-        dbTransactionMgr.finishTransactionBatch(stateBatch, errorReplicaIds);
+        return dbTransactionMgr.finishTransactionBatch(stateBatch, errorReplicaIds);
     }
 
-    public void finishTransactionNew(TransactionState txnState, Set<Long> publishErrorReplicas) throws
+    public TransactionState finishTransactionNew(TransactionState txnState, Set<Long> publishErrorReplicas) throws
             StarRocksException {
         DatabaseTransactionMgr dbTransactionMgr = getDatabaseTransactionMgr(txnState.getDbId());
-        dbTransactionMgr.finishTransactionNew(txnState, publishErrorReplicas);
+        return dbTransactionMgr.finishTransactionNew(txnState, publishErrorReplicas);
     }
 
     public boolean canTxnFinished(TransactionState txn, Set<Long> errReplicas,

fe/fe-core/src/main/java/com/starrocks/transaction/PublishVersionDaemon.java

@@ -387,7 +387,7 @@ private void tryFinishTransaction(TransactionState transactionState) throws Star
             try {
                 // Attempt to finish the transaction with a lock timeout. If it fails, it will be retried in the next cycle.
                 // This approach prevents blocking subsequent transactions due to the current one.
-                globalTransactionMgr.finishTransaction(transactionState.getDbId(),
+                transactionState = globalTransactionMgr.finishTransaction(transactionState.getDbId(),
                         transactionState.getTransactionId(), publishErrorReplicaIds,
                         Config.finish_transaction_default_lock_timeout_ms);
             } catch (StarRocksException exception) {
@@ -421,7 +421,7 @@ private void publishVersionNew(GlobalTransactionMgr globalTransactionMgr, List<T
             }
             try {
                 if (transactionState.checkCanFinish()) {
-                    globalTransactionMgr.finishTransactionNew(transactionState, publishErrorReplicas);
+                    transactionState = globalTransactionMgr.finishTransactionNew(transactionState, publishErrorReplicas);
                 }
                 if (transactionState.getTransactionStatus() != TransactionStatus.VISIBLE) {
                     transactionState.updateSendTaskTime();
@@ -847,9 +847,10 @@ private CompletableFuture<Void> publishLakeTransactionBatchAsync(TransactionStat
                     // as soon as the publish task succeeds. Set readyToFinishTime here so that
                     // publishCanFinishLatencyMs and publishAckLatencyMs are recorded correctly.
                     states.forEach(TransactionState::setReadyToFinishTimeIfUnset);
-                    globalTransactionMgr.finishTransactionBatch(dbId, txnStateBatch, null);
+                    TransactionStateBatch latestStateBatch =
+                            globalTransactionMgr.finishTransactionBatch(dbId, txnStateBatch, null);
                     // here create the job to drop txnLog, for the visibleVersion has been updated
-                    submitDeleteTxnLogJob(txnStateBatch);
+                    submitDeleteTxnLogJob(latestStateBatch);
                 } catch (StarRocksException e) {
                     throw new RuntimeException(e);
                 }