This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 4a11d0b74b7 [fix](job) Clear transaction ID after aborting broker load
transaction (#66884)
4a11d0b74b7 is described below
commit 4a11d0b74b74d6b1b2ec92b9706fbf16618b75ce
Author: hui lai <[email protected]>
AuthorDate: Wed Aug 19 11:54:26 2026 +0800
[fix](job) Clear transaction ID after aborting broker load transaction
(#66884)
fix bug introduced by #66469
Problem Summary:
Cloud Broker Load aborts the active transaction before scheduling a
job-level retry. The pending-task idempotence logic treats any positive
transaction ID as reusable, so the new pending task could reuse an
already aborted transaction. Subsequent rowset prepare requests then
fail because the transaction is no longer in a valid state.
This PR clears the previous attempt's transaction ID after aborting the
Cloud Broker Load transaction. The next pending task therefore begins a
new transaction, while retries within the same pending task can still
reuse their active transaction. If aborting did not complete and the old
transaction remains in PREPARE, the existing label-conflict path can
adopt it only after verifying its status and callback ownership.
### What changed?
- Clear `transactionId` after attempting to abort the previous Cloud
Broker Load transaction.
- Keep transaction reuse unchanged for retries within the same pending
task.
- Add unit coverage for clearing the transaction ID after a successful
abort.
- Add unit coverage for clearing the stale transaction ID when the abort
call fails.
### Release note
Fix Cloud Broker Load job-level retries so they do not reuse an aborted
transaction.
---
.../doris/cloud/load/CloudBrokerLoadJob.java | 6 ++
.../doris/cloud/load/CloudBrokerLoadJobTest.java | 89 ++++++++++++++++++++++
2 files changed, 95 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
index cd5c8dd1b0c..1d58b6eba7f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
@@ -329,6 +329,12 @@ public class CloudBrokerLoadJob extends BrokerLoadJob {
.build());
}
+ // A Cloud Broker Load retry starts a new pending task. Clear the
previous attempt's
+ // transaction id so beginTxn() does not blindly reuse a transaction
that was aborted.
+ // If aborting failed and the old transaction is still PREPARE,
beginTxn() will resolve
+ // the label conflict and adopt the transaction only after verifying
that it belongs to this job.
+ transactionId = 0;
+
// cancel all running coordinators, so that the scheduler's worker
thread will be released
for (TUniqueId loadId : loadIds) {
Coordinator coordinator =
QeProcessorImpl.INSTANCE.getCoordinator(loadId);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/load/CloudBrokerLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/load/CloudBrokerLoadJobTest.java
new file mode 100644
index 00000000000..4ef64019638
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/load/CloudBrokerLoadJobTest.java
@@ -0,0 +1,89 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cloud.load;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.load.BrokerFileGroupAggInfo;
+import org.apache.doris.load.FailMsg;
+import org.apache.doris.load.loadv2.JobState;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TxnStateCallbackFactory;
+
+import com.google.common.collect.Sets;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class CloudBrokerLoadJobTest {
+
+ @Test
+ public void testRetryStartsNewTransactionAfterAbort() throws Exception {
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TxnStateCallbackFactory callbackFactory =
Mockito.mock(TxnStateCallbackFactory.class);
+ BrokerFileGroupAggInfo fileGroupAggInfo =
Mockito.mock(BrokerFileGroupAggInfo.class);
+ CloudBrokerLoadJob job = new CloudBrokerLoadJob();
+ Deencapsulation.setField(job, "id", 1001L);
+ Deencapsulation.setField(job, "dbId", 2001L);
+ Deencapsulation.setField(job, "label", "cloud_broker_load_retry");
+ Deencapsulation.setField(job, "transactionId", 3001L);
+ Deencapsulation.setField(job, "fileGroupAggInfo", fileGroupAggInfo);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+
Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory);
+
Mockito.when(fileGroupAggInfo.getAllTableIds()).thenReturn(Sets.newHashSet(4001L));
+ Mockito.when(transactionMgr.beginTransaction(Mockito.anyLong(),
Mockito.anyList(),
+ Mockito.anyString(), Mockito.any(), Mockito.any(),
Mockito.any(),
+ Mockito.anyLong(), Mockito.anyLong())).thenReturn(5001L);
+
+ job.unprotectedExecuteRetry(new
FailMsg(FailMsg.CancelType.ETL_RUN_FAIL, "rpc failed"));
+ Assert.assertEquals(0L, job.getTransactionId());
+ job.beginTxn();
+ }
+
+ Assert.assertEquals(5001L, job.getTransactionId());
+ Assert.assertEquals(JobState.RETRY, job.getState());
+ Mockito.verify(transactionMgr).abortTransaction(2001L,
"cloud_broker_load_retry", "rpc failed");
+ }
+
+ @Test
+ public void testRetryClearsTransactionIdWhenAbortFails() throws Exception {
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TxnStateCallbackFactory callbackFactory =
Mockito.mock(TxnStateCallbackFactory.class);
+ CloudBrokerLoadJob job = new CloudBrokerLoadJob();
+ Deencapsulation.setField(job, "id", 1002L);
+ Deencapsulation.setField(job, "dbId", 2002L);
+ Deencapsulation.setField(job, "label",
"cloud_broker_load_abort_failed");
+ Deencapsulation.setField(job, "transactionId", 3002L);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+
Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory);
+ Mockito.doThrow(new UserException("abort rpc failed"))
+ .when(transactionMgr).abortTransaction(2002L,
"cloud_broker_load_abort_failed", "rpc failed");
+
+ job.unprotectedExecuteRetry(new
FailMsg(FailMsg.CancelType.ETL_RUN_FAIL, "rpc failed"));
+ }
+
+ Assert.assertEquals(0L, job.getTransactionId());
+ Assert.assertEquals(JobState.RETRY, job.getState());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
