merge

Author: TangSiyang2001

fe/fe-core/src/main/java/org/apache/doris/master/MasterImpl.java

@@ -33,6 +33,7 @@
 import org.apache.doris.cloud.master.CloudReportHandler;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.Status;
 import org.apache.doris.load.DeleteJob;
 import org.apache.doris.load.loadv2.IngestionLoadJob;
 import org.apache.doris.system.Backend;
@@ -352,15 +353,17 @@ private void finishRealtimePush(AgentTask task, TFinishTaskRequest request) thro
                 // we don't need to retry if the returned status code is DELETE_INVALID_CONDITION
                 // or DELETE_INVALID_PARAMETERS
                 // note that they will be converted to TStatusCode.INVALID_ARGUMENT when being sent from be to fe
-                if (request.getTaskStatus().getStatusCode() == TStatusCode.INVALID_ARGUMENT) {
-                    pushTask.countDownToZero(request.getTaskStatus().getStatusCode(),
-                            task.getBackendId() + ": " + request.getTaskStatus().getErrorMsgs().toString());
+                TStatus taskStatus = request.getTaskStatus();
+                String msg = task.getBackendId() + ": " + taskStatus.getErrorMsgs().toString();
+                if (taskStatus.getStatusCode() == TStatusCode.INVALID_ARGUMENT) {
+                    pushTask.countDownToZero(taskStatus.getStatusCode(), msg);
                 } else {
-                    pushTask.countDownLatch(backendId, pushTabletId);
+                    pushTask.countDownLatchWithStatus(backendId, pushTabletId,
+                            new Status(taskStatus.getStatusCode(), msg));
                 }
                 AgentTaskQueue.removeTask(backendId, TTaskType.REALTIME_PUSH, signature);
                 LOG.warn("finish push replica, signature={}, error: {}",
-                        signature, request.getTaskStatus().getErrorMsgs().toString());
+                        signature, taskStatus.getErrorMsgs().toString());
             }
             return;
         }
@@ -471,7 +474,7 @@ private void finishRealtimePush(AgentTask task, TFinishTaskRequest request) thro
             AgentTaskQueue.removeTask(backendId, TTaskType.REALTIME_PUSH, signature);
             LOG.warn("finish push replica error", e);
             if (pushTask.getPushType() == TPushType.DELETE) {
-                pushTask.countDownLatch(backendId, pushTabletId);
+                pushTask.countDownLatchWithStatus(backendId, pushTabletId, Status.CANCELLED);
             }
         } finally {
             olapTable.writeUnlock();

fe/fe-core/src/main/java/org/apache/doris/task/PushTask.java

@@ -212,6 +212,18 @@ public void countDownLatch(long backendId, long tabletId) {
         }
     }
 
+    public void countDownLatchWithStatus(long backendId, long tabletId, Status st) {
+        if (this.latch == null) {
+            return;
+        }
+        if (latch.markedCountDownWithStatus(backendId, tabletId, st)) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("pushTask current latch count with status: {}. backend: {}, tablet:{}, st::{}",
+                        latch.getCount(), backendId, tabletId, st);
+            }
+        }
+    }
+
     // 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) {

regression-test/suites/fault_injection_p0/test_delete_from_timeout.groovy

@@ -51,7 +51,7 @@ suite("test_delete_from_timeout","nonConcurrent") {
 
         test {
             sql """delete from ${tableName} where col1 = "false" and col2 = "-9999782574499444.2" and col3 = "-25"; """
-            exception "invalid parameters for store_cond. condition_size=1"
+            exception "too many versions"
         }
 
         GetDebugPoint().clearDebugPointsForAllBEs()
@@ -73,4 +73,4 @@ suite("test_delete_from_timeout","nonConcurrent") {
     } finally {
         GetDebugPoint().clearDebugPointsForAllBEs()
     }
-}
+}