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 69ea41f20ca branch-4.1: [fix](delete) avoid quorum stalls on delete
push failures #61647 (#62248)
69ea41f20ca is described below
commit 69ea41f20cad57531807cb976b2fa2b11fe734a6
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 9 11:26:09 2026 +0800
branch-4.1: [fix](delete) avoid quorum stalls on delete push failures
#61647 (#62248)
Cherry-picked from #61647
Co-authored-by: Siyang Tang <[email protected]>
---
.../main/java/org/apache/doris/load/DeleteJob.java | 18 +--
.../main/java/org/apache/doris/task/PushTask.java | 6 +
.../java/org/apache/doris/load/DeleteJobTest.java | 157 +++++++++++++++++++++
.../doris/master/MasterImplDeleteTaskTest.java | 152 ++++++++++++++++++++
4 files changed, 319 insertions(+), 14 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
index 2f619a4cc53..c2e8532a684 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/DeleteJob.java
@@ -394,13 +394,16 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
long timeoutMs = getTimeoutMs();
boolean ok = countDownLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
if (ok) {
+ checkAndUpdateQuorum();
+ if (state == DeleteState.QUORUM_FINISHED || state ==
DeleteState.FINISHED) {
+ return;
+ }
if (!countDownLatch.getStatus().ok()) {
// encounter some errors that don't need to retry, abort
directly
LOG.warn("delete job failed, errmsg={}",
countDownLatch.getStatus().getErrorMsg());
throw new UserException(String.format("delete job failed,
errmsg:%s",
countDownLatch.getStatus().getErrorMsg()));
}
- return;
}
//handle failure
@@ -420,19 +423,6 @@ public class DeleteJob extends
AbstractTxnStateChangeCallback implements DeleteJ
throw new UserException(String.format("delete job timeout,
timeout(ms):%s, msg:%s", timeoutMs, errMsg));
case QUORUM_FINISHED:
case FINISHED:
- long nowQuorumTimeMs = System.currentTimeMillis();
- long endQuorumTimeoutMs = nowQuorumTimeMs + timeoutMs / 2;
- // if job's state is quorum_finished then wait for a period of
time and commit it.
- while (state == DeleteState.QUORUM_FINISHED
- && endQuorumTimeoutMs > nowQuorumTimeMs) {
- checkAndUpdateQuorum();
- Thread.sleep(1000);
- nowQuorumTimeMs = System.currentTimeMillis();
- if (LOG.isDebugEnabled()) {
- LOG.debug("wait for quorum finished delete job: {},
txn id: {}",
- id, transactionId);
- }
- }
break;
default:
throw new IllegalStateException("wrong delete job state: " +
state.name());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java
b/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java
index cb06f4d27f6..1862df155c0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java
@@ -224,6 +224,12 @@ public class PushTask extends AgentTask {
}
}
+ @Override
+ public void failedWithMsg(String errMsg) {
+ super.failedWithMsg(errMsg);
+ countDownLatchWithStatus(getBackendId(), getTabletId(), new
Status(TStatusCode.CANCELLED, errMsg));
+ }
+
// call this always means one of tasks is failed. count down to zero to
finish entire task
public void countDownToZero(TStatusCode code, String errMsg) {
if (this.latch != null) {
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/DeleteJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/DeleteJobTest.java
new file mode 100644
index 00000000000..8f76655fcf8
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/DeleteJobTest.java
@@ -0,0 +1,157 @@
+// 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.load;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.LocalReplica;
+import org.apache.doris.catalog.Replica;
+import org.apache.doris.catalog.TabletInvertedIndex;
+import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.common.MarkedCountDownLatch;
+import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TStorageMedium;
+
+import com.google.common.collect.Lists;
+import mockit.Mock;
+import mockit.MockUp;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class DeleteJobTest {
+ private static final long DB_ID = 1L;
+ private static final long TABLE_ID = 2L;
+ private static final long PARTITION_ID = 3L;
+ private static final long TABLET_ID = 4L;
+
+ private final Database database = new Database(DB_ID, "db");
+ private final InternalCatalog catalog =
Deencapsulation.newInstance(InternalCatalog.class);
+ private final TabletInvertedIndex invertedIndex = new
TabletInvertedIndex() {
+ @Override
+ public List<Replica> getReplicas(Long tabletId) {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ public void deleteTablet(long tabletId) {
+ }
+
+ @Override
+ public void addReplica(long tabletId, Replica replica) {
+ }
+
+ @Override
+ public void deleteReplica(long tabletId, long backendId) {
+ }
+
+ @Override
+ public Replica getReplica(long tabletId, long backendId) {
+ return null;
+ }
+
+ @Override
+ public List<Replica> getReplicasByTabletId(long tabletId) {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ protected void innerClear() {
+ }
+ };
+
+ @Before
+ public void setUp() {
+ invertedIndex.addTablet(TABLET_ID, new TabletMeta(DB_ID, TABLE_ID,
PARTITION_ID, 5L, 6, TStorageMedium.HDD));
+
+ new MockUp<Env>() {
+ @Mock
+ public InternalCatalog getCurrentInternalCatalog() {
+ return catalog;
+ }
+
+ @Mock
+ public TabletInvertedIndex getCurrentInvertedIndex() {
+ return invertedIndex;
+ }
+ };
+
+ new MockUp<InternalCatalog>() {
+ @Mock
+ public Database getDbOrMetaException(long dbId) {
+ return database;
+ }
+ };
+ }
+
+ @Test
+ public void testAwaitIgnoresFailureStatusAfterQuorumReached() throws
Exception {
+ MarkedCountDownLatch<Long, Long> latch = new
MarkedCountDownLatch<Long, Long>(1);
+ latch.addMark(1L, TABLET_ID);
+ latch.markedCountDownWithStatus(1L, TABLET_ID, new
Status(TStatusCode.INTERNAL_ERROR, "too many versions"));
+
+ DeleteJob deleteJob = newDeleteJob((short) 3, latch);
+ deleteJob.addFinishedReplica(PARTITION_ID, TABLET_ID,
+ new LocalReplica(11L, 21L, Replica.ReplicaState.NORMAL, 1L,
6));
+ deleteJob.addFinishedReplica(PARTITION_ID, TABLET_ID,
+ new LocalReplica(12L, 22L, Replica.ReplicaState.NORMAL, 1L,
6));
+
+ deleteJob.await();
+
+ Assert.assertEquals(DeleteJob.DeleteState.QUORUM_FINISHED,
deleteJob.getState());
+ }
+
+ @Test
+ public void testAwaitReturnsFailureStatusWhenQuorumNotReached() {
+ MarkedCountDownLatch<Long, Long> latch = new
MarkedCountDownLatch<Long, Long>(1);
+ latch.addMark(1L, TABLET_ID);
+ latch.markedCountDownWithStatus(1L, TABLET_ID, new
Status(TStatusCode.INTERNAL_ERROR, "too many versions"));
+
+ DeleteJob deleteJob = newDeleteJob((short) 1, latch);
+
+ try {
+ deleteJob.await();
+ Assert.fail("delete job should fail when quorum is not reached");
+ } catch (UserException e) {
+ Assert.assertTrue(e.getMessage().contains("too many versions"));
+ Assert.assertEquals(DeleteJob.DeleteState.UN_QUORUM,
deleteJob.getState());
+ } catch (Exception e) {
+ Assert.fail("unexpected exception: " + e.getMessage());
+ }
+ }
+
+ private DeleteJob newDeleteJob(short replicaNum,
MarkedCountDownLatch<Long, Long> latch) {
+ DeleteInfo deleteInfo = new DeleteInfo(DB_ID, TABLE_ID, "tbl",
Collections.emptyList(),
+ false, Lists.newArrayList(PARTITION_ID),
Lists.newArrayList("p1"));
+ Map<Long, Short> partitionReplicaNum =
Collections.singletonMap(PARTITION_ID, replicaNum);
+ DeleteJob deleteJob = new DeleteJob(100L, 200L, "label",
partitionReplicaNum, deleteInfo);
+ deleteJob.setCountDownLatch(latch);
+ Set<Long> totalTablets = Deencapsulation.getField(deleteJob,
"totalTablets");
+ totalTablets.add(TABLET_ID);
+ return deleteJob;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/master/MasterImplDeleteTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/master/MasterImplDeleteTaskTest.java
new file mode 100644
index 00000000000..3bf846b05f9
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/master/MasterImplDeleteTaskTest.java
@@ -0,0 +1,152 @@
+// 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.master;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.MarkedCountDownLatch;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.task.AgentTaskQueue;
+import org.apache.doris.task.PushTask;
+import org.apache.doris.thrift.TBackend;
+import org.apache.doris.thrift.TFinishTaskRequest;
+import org.apache.doris.thrift.TPriority;
+import org.apache.doris.thrift.TPushType;
+import org.apache.doris.thrift.TStatus;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TTaskType;
+
+import com.google.common.collect.Lists;
+import mockit.Mock;
+import mockit.MockUp;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MasterImplDeleteTaskTest {
+ private static final long BACKEND_ID = 10000L;
+ private static final long DB_ID = 20000L;
+ private static final long TABLE_ID = 30000L;
+ private static final long PARTITION_ID = 40000L;
+ private static final long INDEX_ID = 50000L;
+ private static final long TABLET_ID = 60000L;
+ private static final long REPLICA_ID = 70000L;
+ private static final long TRANSACTION_ID = 80000L;
+ private static final long SIGNATURE = 90000L;
+ private static final String HOST = "127.0.0.1";
+ private static final int HEARTBEAT_PORT = 9050;
+ private static final int BE_PORT = 9060;
+
+ private MasterImpl masterImpl;
+
+ @Before
+ public void setUp() {
+ AgentTaskQueue.clearAllTasks();
+
+ SystemInfoService systemInfoService = new SystemInfoService();
+ Backend backend = new Backend(BACKEND_ID, HOST, HEARTBEAT_PORT);
+ backend.setBePort(BE_PORT);
+ systemInfoService.addBackend(backend);
+
+ new MockUp<Env>() {
+ @Mock
+ public SystemInfoService getCurrentSystemInfo() {
+ return systemInfoService;
+ }
+ };
+
+ new MockUp<ReportHandler>() {
+ @Mock
+ public void start() {
+ }
+ };
+
+ masterImpl = new MasterImpl();
+ }
+
+ @After
+ public void tearDown() {
+ AgentTaskQueue.clearAllTasks();
+ }
+
+ @Test
+ public void testDeletePushGenericFailureCountsDownSingleMark() {
+ MarkedCountDownLatch<Long, Long> latch = new
MarkedCountDownLatch<Long, Long>(2);
+ latch.addMark(BACKEND_ID, TABLET_ID);
+ latch.addMark(BACKEND_ID + 1, TABLET_ID + 1);
+
+ PushTask pushTask = newDeletePushTask(latch);
+ AgentTaskQueue.addTask(pushTask);
+
+
masterImpl.finishTask(newFinishTaskRequest(TStatusCode.INTERNAL_ERROR));
+
+ Assert.assertEquals(1, latch.getCount());
+ Assert.assertEquals(TStatusCode.INTERNAL_ERROR,
latch.getStatus().getErrorCode());
+ Assert.assertEquals(1, pushTask.getFailedTimes());
+ Assert.assertNull(AgentTaskQueue.getTask(BACKEND_ID,
TTaskType.REALTIME_PUSH, SIGNATURE));
+ }
+
+ @Test
+ public void testDeletePushInvalidArgumentCountsDownToZero() {
+ MarkedCountDownLatch<Long, Long> latch = new
MarkedCountDownLatch<Long, Long>(2);
+ latch.addMark(BACKEND_ID, TABLET_ID);
+ latch.addMark(BACKEND_ID + 1, TABLET_ID + 1);
+
+ PushTask pushTask = newDeletePushTask(latch);
+ AgentTaskQueue.addTask(pushTask);
+
+
masterImpl.finishTask(newFinishTaskRequest(TStatusCode.INVALID_ARGUMENT));
+
+ Assert.assertEquals(0, latch.getCount());
+ Assert.assertEquals(TStatusCode.INVALID_ARGUMENT,
latch.getStatus().getErrorCode());
+ Assert.assertEquals(1, pushTask.getFailedTimes());
+ Assert.assertNull(AgentTaskQueue.getTask(BACKEND_ID,
TTaskType.REALTIME_PUSH, SIGNATURE));
+ }
+
+ @Test
+ public void testDeletePushFailedWithMsgKeepsFailureStatus() {
+ MarkedCountDownLatch<Long, Long> latch = new
MarkedCountDownLatch<Long, Long>(1);
+ latch.addMark(BACKEND_ID, TABLET_ID);
+
+ PushTask pushTask = newDeletePushTask(latch);
+ pushTask.failedWithMsg("submit failed");
+
+ Assert.assertEquals(0, latch.getCount());
+ Assert.assertEquals(TStatusCode.CANCELLED,
latch.getStatus().getErrorCode());
+ Assert.assertEquals("submit failed", latch.getStatus().getErrorMsg());
+ }
+
+ private PushTask newDeletePushTask(MarkedCountDownLatch<Long, Long> latch)
{
+ PushTask pushTask = new PushTask(null, BACKEND_ID, DB_ID, TABLE_ID,
PARTITION_ID, INDEX_ID,
+ TABLET_ID, REPLICA_ID, 1, 1L, null, 0, 60, -1,
TPushType.DELETE, null,
+ false, TPriority.NORMAL, TTaskType.REALTIME_PUSH,
TRANSACTION_ID, SIGNATURE,
+ null, null, null, 0, null);
+ pushTask.setCountDownLatch(latch);
+ return pushTask;
+ }
+
+ private TFinishTaskRequest newFinishTaskRequest(TStatusCode statusCode) {
+ TStatus taskStatus = new TStatus(statusCode);
+ taskStatus.setErrorMsgs(Lists.newArrayList("delete failed"));
+ TBackend tBackend = new TBackend(HOST, BE_PORT, 0);
+ TFinishTaskRequest request = new TFinishTaskRequest(tBackend,
TTaskType.REALTIME_PUSH, SIGNATURE, taskStatus);
+ request.setReportVersion(1L);
+ return request;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
