This is an automated email from the ASF dual-hosted git repository.
mymeiyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5ea9bdcf3dd [fix](fe) Keep rollup jobs waiting on conflict txn abort
failure (#67661)
5ea9bdcf3dd is described below
commit 5ea9bdcf3dd25cffe09237b0f25b9d50b5afbb88
Author: meiyi <[email protected]>
AuthorDate: Fri Sep 11 17:20:32 2026 +0800
[fix](fe) Keep rollup jobs waiting on conflict txn abort failure (#67661)
Related PR: #65196
Problem Summary:
A rollup job in WAITING_TXN checks and aborts failed conflict
transactions before proceeding. If a transaction is already committed,
becomes visible, or is cleaned up before the abort completes,
abortTransaction() can throw UserException. RollupJobV2 propagates this
as AlterCancelException, causing the entire rollup job to be cancelled.
Restarting FE with committed but not yet visible transactions reproduces
this failure (the txn_insert_restart_fe_with_schema_change regression
case can produce this bug).
Apply the retry behavior introduced for schema change in #65196 to
rollup jobs: log the abort failure, keep the job in WAITING_TXN, and
re-query transaction state in the next scheduler round. Also exclude
committed and final transactions from the shared failed-transaction
selector to avoid unnecessary abort attempts.
---
.../java/org/apache/doris/alter/RollupJobV2.java | 10 ++++--
.../doris/transaction/GlobalTransactionMgr.java | 4 +++
.../org/apache/doris/alter/RollupJobV2Test.java | 37 ++++++++++++++++++++++
.../transaction/GlobalTransactionMgrTest.java | 19 +++++++++++
4 files changed, 68 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java
b/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java
index 7c5523373b4..060e24e3be4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java
@@ -708,8 +708,14 @@ public class RollupJobV2 extends AlterJobV2 implements
GsonPostProcessable {
if (Config.enable_abort_txn_by_checking_conflict_txn) {
List<TransactionState> failedTxns =
GlobalTransactionMgr.checkFailedTxns(unFinishedTxns);
for (TransactionState txn : failedTxns) {
- Env.getCurrentGlobalTransactionMgr()
- .abortTransaction(txn.getDbId(),
txn.getTransactionId(), "Cancel by schema change");
+ try {
+ Env.getCurrentGlobalTransactionMgr()
+ .abortTransaction(txn.getDbId(),
txn.getTransactionId(), "Cancel by schema change");
+ } catch (UserException e) {
+ LOG.warn("failed to abort previous load txn {}, wait next
round. rollup job: {}",
+ txn.getTransactionId(), jobId, e);
+ return false;
+ }
}
}
return unFinishedTxns.isEmpty();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
index 621c8b9b220..ae6a1abcd36 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
@@ -515,6 +515,10 @@ public class GlobalTransactionMgr implements
GlobalTransactionMgrIface {
public static List<TransactionState>
checkFailedTxns(List<TransactionState> conflictTxns) {
List<TransactionState> failedTxns = new ArrayList<>();
for (TransactionState txn : conflictTxns) {
+ TransactionStatus status = txn.getTransactionStatus();
+ if (status == TransactionStatus.COMMITTED ||
status.isFinalStatus()) {
+ continue;
+ }
if (checkFailedTxnsByCoordinator(txn)) {
failedTxns.add(txn);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
index c525ea2d4a4..b796516ab67 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/alter/RollupJobV2Test.java
@@ -55,7 +55,13 @@ import org.apache.doris.task.AgentTaskQueue;
import org.apache.doris.thrift.TStorageFormat;
import org.apache.doris.thrift.TTaskType;
import org.apache.doris.transaction.FakeTransactionIDGenerator;
+import org.apache.doris.transaction.GlobalTransactionMgr;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TransactionState;
+import org.apache.doris.transaction.TransactionState.LoadJobSourceType;
+import org.apache.doris.transaction.TransactionState.TxnCoordinator;
+import org.apache.doris.transaction.TransactionState.TxnSourceType;
+import org.apache.doris.transaction.TransactionStatus;
import com.google.common.collect.Lists;
import org.junit.jupiter.api.AfterEach;
@@ -144,6 +150,37 @@ public class RollupJobV2Test {
}
}
+ @Test
+ public void testCommitWhileAbortingPreviousLoad() throws Exception {
+ long txnId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1),
"commit_during_rollup_abort",
+ new TxnCoordinator(TxnSourceType.FE, 0, "missing", 0),
LoadJobSourceType.FRONTEND, 60);
+ TransactionState txn =
masterTransMgr.getTransactionState(CatalogTestUtil.testDbId1, txnId);
+ Database db =
masterEnv.getInternalCatalog().getDbOrDdlException(CatalogTestUtil.testDbId1);
+ OlapTable table = (OlapTable)
db.getTableOrDdlException(CatalogTestUtil.testTableId1);
+ MaterializedViewHandler handler =
masterEnv.getMaterializedViewHandler();
+ handler.process(Lists.newArrayList(op), db, table);
+ RollupJobV2 job = (RollupJobV2)
handler.getAlterJobsV2().values().iterator().next();
+ job.jobState = JobState.WAITING_TXN;
+ job.watershedTxnId = txnId + 1;
+
+ try (MockedStatic<GlobalTransactionMgr> mocked = Mockito.mockStatic(
+ GlobalTransactionMgr.class, Mockito.CALLS_REAL_METHODS)) {
+ mocked.when(() ->
GlobalTransactionMgr.checkFailedTxns(Mockito.anyList())).thenAnswer(invocation
-> {
+ List<TransactionState> failed = (List<TransactionState>)
invocation.callRealMethod();
+ Assertions.assertEquals(List.of(txn), failed);
+ txn.setTransactionStatus(TransactionStatus.COMMITTED);
+ return failed;
+ });
+ job.runWaitingTxnJob();
+ Assertions.assertEquals(JobState.WAITING_TXN, job.getJobState());
+ Assertions.assertEquals(TransactionStatus.COMMITTED,
txn.getTransactionStatus());
+ }
+ Assertions.assertFalse(job.checkFailedPreviousLoadAndAbort());
+ txn.setTransactionStatus(TransactionStatus.VISIBLE);
+ Assertions.assertTrue(job.checkFailedPreviousLoadAndAbort());
+ }
+
@Test
public void testRunRollupJobConcurrentLimit() throws UserException {
if (fakeEnv != null) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/transaction/GlobalTransactionMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/transaction/GlobalTransactionMgrTest.java
index b76370ea5e4..0231a6360e8 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/transaction/GlobalTransactionMgrTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/transaction/GlobalTransactionMgrTest.java
@@ -126,6 +126,25 @@ public class GlobalTransactionMgrTest {
}
}
+ @Test
+ public void testCheckFailedTxnsWithOfflineCoordinator() {
+ FakeEnv.setEnv(masterEnv);
+ for (TxnSourceType source : List.of(TxnSourceType.FE,
TxnSourceType.BE)) {
+ TransactionState txn = new
TransactionState(CatalogTestUtil.testDbId1,
+ Lists.newArrayList(CatalogTestUtil.testTableId1), 1,
"offline_coordinator", null,
+ LoadJobSourceType.FRONTEND, new TxnCoordinator(source,
Long.MAX_VALUE, "missing", 0), -1, 60000);
+ for (TransactionStatus status : List.of(TransactionStatus.PREPARE,
TransactionStatus.PRECOMMITTED)) {
+ txn.setTransactionStatus(status);
+ Assertions.assertEquals(List.of(txn),
GlobalTransactionMgr.checkFailedTxns(List.of(txn)));
+ }
+ for (TransactionStatus status :
List.of(TransactionStatus.COMMITTED,
+ TransactionStatus.VISIBLE, TransactionStatus.ABORTED)) {
+ txn.setTransactionStatus(status);
+
Assertions.assertTrue(GlobalTransactionMgr.checkFailedTxns(List.of(txn)).isEmpty());
+ }
+ }
+ }
+
@Test
public void testBeginTransaction() throws LabelAlreadyUsedException,
AnalysisException,
BeginTransactionException, DuplicatedRequestException,
QuotaExceedException, MetaNotFoundException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
