[improvement](report) report handler discard old report tasks #39469 (#…

…39605)

cherry pick from #39469

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

@@ -104,17 +104,19 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.stream.Collectors;
 
 public class ReportHandler extends Daemon {
     private static final Logger LOG = LogManager.getLogger(ReportHandler.class);
 
-    private BlockingQueue<ReportTask> reportQueue = Queues.newLinkedBlockingQueue();
+    private BlockingQueue<BackendReportType> reportQueue = Queues.newLinkedBlockingQueue();
+
+    private Map<BackendReportType, ReportTask> reportTasks = Maps.newHashMap();
 
     private enum ReportType {
-        UNKNOWN,
         TASK,
         DISK,
         TABLET
@@ -158,7 +160,7 @@ public TMasterResult handleReport(TReportRequest request) throws TException {
         Map<Long, Long> partitionsVersion = null;
         long reportVersion = -1;
 
-        ReportType reportType = ReportType.UNKNOWN;
+        ReportType reportType = null;
 
         if (request.isSetTasks()) {
             tasks = request.getTasks();
@@ -189,8 +191,16 @@ public TMasterResult handleReport(TReportRequest request) throws TException {
             backend.setTabletMaxCompactionScore(request.getTabletMaxCompactionScore());
         }
 
-        ReportTask reportTask = new ReportTask(beId, tasks, disks, tablets, partitionsVersion, reportVersion,
-                request.getStoragePolicy(), request.getResource(), request.getNumCores(),
+        if (reportType == null) {
+            tStatus.setStatusCode(TStatusCode.INTERNAL_ERROR);
+            tStatus.setErrorMsgs(Lists.newArrayList("unknown report type"));
+            LOG.error("receive unknown report type from be {}. current queue size: {}",
+                    backend.getId(), reportQueue.size());
+            return result;
+        }
+
+        ReportTask reportTask = new ReportTask(beId, reportType, tasks, disks, tablets, partitionsVersion,
+                reportVersion, request.getStoragePolicy(), request.getResource(), request.getNumCores(),
                 request.getPipelineExecutorSize());
         try {
             putToQueue(reportTask);
@@ -202,8 +212,8 @@ public TMasterResult handleReport(TReportRequest request) throws TException {
             tStatus.setErrorMsgs(errorMsgs);
             return result;
         }
-        LOG.info("receive report from be {}. type: {}, current queue size: {}",
-                backend.getId(), reportType, reportQueue.size());
+        LOG.info("receive report from be {}. type: {}, report version {}, current queue size: {}",
+                backend.getId(), reportType, reportVersion, reportQueue.size());
         return result;
     }
 
@@ -215,7 +225,14 @@ private void putToQueue(ReportTask reportTask) throws Exception {
                     "the report queue size exceeds the limit: "
                             + Config.report_queue_size + ". current: " + currentSize);
         }
-        reportQueue.put(reportTask);
+
+        BackendReportType backendReportType = new BackendReportType(reportTask.beId, reportTask.reportType);
+
+        synchronized (reportTasks) {
+            reportTasks.put(backendReportType, reportTask);
+        }
+
+        reportQueue.put(backendReportType);
     }
 
     private Map<Long, TTablet> buildTabletMap(List<TTablet> tabletList) {
@@ -230,9 +247,38 @@ private Map<Long, TTablet> buildTabletMap(List<TTablet> tabletList) {
         return tabletMap;
     }
 
+    private class BackendReportType {
+        private long beId;
+        private ReportType reportType;
+
+        public BackendReportType(long beId, ReportType reportType) {
+            this.beId = beId;
+            this.reportType = reportType;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(beId, reportType);
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            if (this == other) {
+                return true;
+            }
+            if (!(other instanceof BackendReportType)) {
+                return false;
+            }
+            BackendReportType otherBeReport = (BackendReportType) other;
+            return this.beId == otherBeReport.beId
+                && this.reportType == otherBeReport.reportType;
+        }
+    }
+
     private class ReportTask extends MasterTask {
 
         private long beId;
+        private ReportType reportType;
         private Map<TTaskType, Set<Long>> tasks;
         private Map<String, TDisk> disks;
         private Map<Long, TTablet> tablets;
@@ -244,12 +290,13 @@ private class ReportTask extends MasterTask {
         private int cpuCores;
         private int pipelineExecutorSize;
 
-        public ReportTask(long beId, Map<TTaskType, Set<Long>> tasks,
+        public ReportTask(long beId, ReportType reportType, Map<TTaskType, Set<Long>> tasks,
                 Map<String, TDisk> disks, Map<Long, TTablet> tablets,
                 Map<Long, Long> partitionsVersion, long reportVersion,
                 List<TStoragePolicy> storagePolicies, List<TStorageResource> storageResources, int cpuCores,
                 int pipelineExecutorSize) {
             this.beId = beId;
+            this.reportType = reportType;
             this.tasks = tasks;
             this.disks = disks;
             this.tablets = tablets;
@@ -1383,13 +1430,28 @@ private static boolean addReplica(long tabletId, TabletMeta tabletMeta, TTabletI
     @Override
     protected void runOneCycle() {
         while (true) {
-            ReportTask task = null;
-            try {
-                task = reportQueue.take();
+            ReportTask task = takeReportTask();
+            if (task != null) {
                 task.exec();
-            } catch (InterruptedException e) {
-                LOG.warn("got interupted exception when executing report", e);
             }
         }
     }
+
+    private ReportTask takeReportTask() {
+        BackendReportType backendReportType;
+        try {
+            backendReportType = reportQueue.take();
+        } catch (InterruptedException e) {
+            LOG.warn("got interupted exception when executing report", e);
+            return null;
+        }
+
+        ReportTask task = null;
+        synchronized (reportTasks) {
+            task = reportTasks.get(backendReportType);
+            reportTasks.remove(backendReportType);
+        }
+
+        return task;
+    }
 }