[fix](cloud) fix routine load job stuck if commit transaction failed (#40539)

At the before commit stage, a write lock will be added. If the commit
transaction fails, the thread will return directly and the write lock
will no longer be released which cause job stuck.

Author: sollhui

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

@@ -505,7 +505,17 @@ private void commitTransaction(long dbId, List<Table> tableList, long transactio
         }
 
         final CommitTxnRequest commitTxnRequest = builder.build();
-        commitTxn(commitTxnRequest, transactionId, is2PC, dbId, tableList);
+        try {
+            commitTxn(commitTxnRequest, transactionId, is2PC, dbId, tableList);
+        } catch (UserException e) {
+            // For routine load, it is necessary to release the write lock when commit transaction fails,
+            // otherwise it will cause the lock added in beforeCommitted to not be released.
+            if (txnCommitAttachment != null && txnCommitAttachment instanceof RLTaskTxnCommitAttachment) {
+                RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment = (RLTaskTxnCommitAttachment) txnCommitAttachment;
+                Env.getCurrentEnv().getRoutineLoadManager().getJob(rlTaskTxnCommitAttachment.getJobId()).writeUnlock();
+            }
+            throw e;
+        }
     }
 
     private void commitTxn(CommitTxnRequest commitTxnRequest, long transactionId, boolean is2PC, long dbId,
@@ -1037,6 +1047,12 @@ public void abortTransaction(Long dbId, Long transactionId, String reason,
             Preconditions.checkNotNull(abortTxnResponse.getStatus());
         } catch (RpcException e) {
             LOG.warn("abortTxn failed, transactionId:{}, Exception", transactionId, e);
+            // For routine load, it is necessary to release the write lock when abort transaction fails,
+            // otherwise it will cause the lock added in beforeAborted to not be released.
+            if (txnCommitAttachment != null && txnCommitAttachment instanceof RLTaskTxnCommitAttachment) {
+                RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment = (RLTaskTxnCommitAttachment) txnCommitAttachment;
+                Env.getCurrentEnv().getRoutineLoadManager().getJob(rlTaskTxnCommitAttachment.getJobId()).writeUnlock();
+            }
             throw new UserException("abortTxn failed, errMsg:" + e.getMessage());
         }
 

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -485,7 +485,7 @@ protected void writeLock() {
         lock.writeLock().lock();
     }
 
-    protected void writeUnlock() {
+    public void writeUnlock() {
         lock.writeLock().unlock();
     }