[Enhancement] Add transaction error message to loads internal table (backport StarRocks#61364) (StarRocks#62852)

Signed-off-by: meegoo <[email protected]>
Co-authored-by: meegoo <[email protected]>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/common/Config.java

@@ -1216,7 +1216,7 @@ public class Config extends ConfigBase {
      * Online optimize table allows to optimize a table without blocking write operations.
      */
     @ConfField(mutable = true)
-    public static boolean enable_online_optimize_table = true;
+    public static boolean enable_online_optimize_table = false;
 
     /**
      * If set to true, FE will check backend available capacity by storage medium when create table

fe/fe-core/src/main/java/com/starrocks/load/LoadConstants.java

@@ -29,6 +29,7 @@ public class LoadConstants {
     public static final String RUNTIME_DETAILS_PLAN_TIME_MS = "plan_time_ms";
     public static final String RUNTIME_DETAILS_RECEIVE_DATA_TIME_MS = "receive_data_time_ms";
     public static final String RUNTIME_DETAILS_BEGIN_TXN_TIME_MS = "begin_txn_time_ms";
+    public static final String RUNTIME_DETAILS_TXN_ERROR_MSG = "txn_error_msg";
 
     public static final String PROPERTIES_TIMEOUT = "timeout";
     public static final String PROPERTIES_MAX_FILTER_RATIO = "max_filter_ratio";

fe/fe-core/src/main/java/com/starrocks/load/loadv2/LoadJob.java

@@ -927,6 +927,16 @@ public String toRuntimeDetails() {
         }
         runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_LOAD_ID, Joiner.on(", ").join(loadIds));
         runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_TXN_ID, transactionId);
+        TransactionState txnState = GlobalStateMgr.getCurrentState()
+                .getGlobalTransactionMgr().getTransactionState(dbId, transactionId);
+        if (txnState != null) {
+            txnState.writeLock();
+            try {
+                runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_TXN_ERROR_MSG, txnState.getErrMsg());
+            } finally {
+                txnState.writeUnlock();
+            }
+        }
         runtimeDetails.putAll(loadingStatus.getLoadStatistic().toRuntimeDetails());
         Gson gson = new Gson();
         return gson.toJson(runtimeDetails);

fe/fe-core/src/main/java/com/starrocks/load/streamload/StreamLoadTask.java

@@ -1534,6 +1534,15 @@ public String toRuntimeDetails() {
         runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_BEGIN_TXN_TIME_MS, beginTxnTimeMs);
         runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_RECEIVE_DATA_TIME_MS, receiveDataTimeMs);
         runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_PLAN_TIME_MS, planTimeMs);
+        TransactionState txnState = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().getTransactionState(dbId, txnId);
+        if (txnState != null) {
+            txnState.writeLock();
+            try {
+                runtimeDetails.put(LoadConstants.RUNTIME_DETAILS_TXN_ERROR_MSG, txnState.getErrMsg());
+            } finally {
+                txnState.writeUnlock();
+            }
+        }
         Gson gson = new Gson();
         return gson.toJson(runtimeDetails);
     }

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

@@ -613,6 +613,15 @@ public boolean publishPartitionBatch(Database db, long tableId, PartitionPublish
                 stateBatch.putBeTablets(partitionId, nodeToTablets);
             }
         } catch (Exception e) {
+            for (int i = 0; i < transactionStates.size(); i++) {
+                TransactionState txnState = transactionStates.get(i);
+                txnState.writeLock();
+                try {
+                    txnState.setErrorMsg("Fail to publish partition " + partitionId + " error " + e.getMessage());
+                } finally {
+                    txnState.writeUnlock();
+                }
+            }
             LOG.error("Fail to publish partition {} of txnIds {}:", partitionId,
                     txnInfos.stream().map(i -> i.txnId).collect(Collectors.toList()), e);
             return false;
@@ -892,6 +901,13 @@ private boolean publishPartition(@NotNull Database db, @NotNull TableCommitInfo
             }
             return true;
         } catch (Throwable e) {
+            txnState.writeLock();
+            try {
+                txnState.setErrorMsg("Fail to publish partition " + partitionCommitInfo.getPhysicalPartitionId()
+                        + " error " + e.getMessage());
+            } finally {
+                txnState.writeUnlock();
+            }
             // prevent excessive logging
             if (partitionCommitInfo.getVersionTime() < 0 &&
                     Math.abs(partitionCommitInfo.getVersionTime()) + 10000 < System.currentTimeMillis()) {

fe/fe-core/src/test/java/com/starrocks/alter/OptimizeJobV2BuilderTest.java

@@ -15,6 +15,7 @@
 package com.starrocks.alter;
 
 import com.starrocks.catalog.OlapTable;
+import com.starrocks.common.Config;
 import com.starrocks.common.StarRocksException;
 import com.starrocks.sql.ast.KeysDesc;
 import com.starrocks.sql.ast.OptimizeClause;
@@ -86,6 +87,7 @@ public void testBuildMergePartitionsJob() throws StarRocksException {
 
     @Test
     public void testBuildWithoutOptimizeClause() throws StarRocksException {
+        Config.enable_online_optimize_table = true;
         // Create a mock OlapTable
         OlapTable table = Mockito.mock(OlapTable.class);
         Mockito.when(table.getId()).thenReturn(123L);
@@ -108,4 +110,4 @@ public void testBuildWithoutOptimizeClause() throws StarRocksException {
         Assertions.assertEquals(123L, onlineOptimizeJob.getTableId());
         Assertions.assertEquals("myTable", onlineOptimizeJob.getTableName());
     }
-}
+}