This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f77661a1b40 branch-4.1: [fix](job) Clear transaction ID after aborting
broker load transaction #66884 (#66920)
f77661a1b40 is described below
commit f77661a1b40726492b91c2ba6dca368963151f4c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Aug 20 16:32:58 2026 +0800
branch-4.1: [fix](job) Clear transaction ID after aborting broker load
transaction #66884 (#66920)
Cherry-picked from #66884
Co-authored-by: hui lai <[email protected]>
---
.../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]
