From bb35d814f6758c21d414f338e46c98c4dce70bd9 Mon Sep 17 00:00:00 2001 From: Yixuan Wang Date: Fri, 4 Sep 2026 22:54:41 +0800 Subject: [PATCH] 1 --- .../java/org/apache/doris/load/ExportMgr.java | 16 +++-- .../doris/load/loadv2/ExportMgrTest.java | 72 +++++++++++++++++++ 2 files changed, 82 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java index 49958e46edba60..1f185ff116a82b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java @@ -485,15 +485,19 @@ public void removeOldExportJobs() { if (exportIdToJob.size() > Config.max_export_history_job_num) { List> jobList = new ArrayList<>(exportIdToJob.entrySet()); jobList.sort(Comparator.comparingLong(entry -> entry.getValue().getCreateTimeMs())); - while (exportIdToJob.size() > Config.max_export_history_job_num) { - // Remove the oldest job - Map.Entry oldestEntry = jobList.remove(0); + Iterator> jobIterator = jobList.iterator(); + while (exportIdToJob.size() > Config.max_export_history_job_num && jobIterator.hasNext()) { + Map.Entry oldestEntry = jobIterator.next(); + ExportJob job = oldestEntry.getValue(); + if (job.getState() != ExportJobState.CANCELLED && job.getState() != ExportJobState.FINISHED) { + continue; + } exportIdToJob.remove(oldestEntry.getKey()); - Map labelJobs = dbTolabelToExportJobId.get(oldestEntry.getValue().getDbId()); + Map labelJobs = dbTolabelToExportJobId.get(job.getDbId()); if (labelJobs != null) { - labelJobs.remove(oldestEntry.getValue().getLabel()); + labelJobs.remove(job.getLabel()); if (labelJobs.isEmpty()) { - dbTolabelToExportJobId.remove(oldestEntry.getValue().getDbId()); + dbTolabelToExportJobId.remove(job.getDbId()); } } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/ExportMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/ExportMgrTest.java index 116c20ebe7a310..c207108425b610 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/ExportMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/ExportMgrTest.java @@ -134,6 +134,78 @@ public void testRemoveOldExportJobs() { } } + @Test + public void testRemoveOldExportJobsKeepsRunningJobs() { + ExportMgr isolatedExportMgr = new ExportMgr(); + int originalMaxHistoryJobNum = Config.max_export_history_job_num; + Config.max_export_history_job_num = 2; + try { + long currentTime = System.currentTimeMillis(); + ExportJob pendingJob = makeExportJob(1001, "pending"); + Deencapsulation.setField(pendingJob, "createTimeMs", currentTime - 4000); + Deencapsulation.setField(pendingJob, "state", ExportJobState.PENDING); + isolatedExportMgr.unprotectAddJob(pendingJob); + + ExportJob exportingJob = makeExportJob(1002, "exporting"); + Deencapsulation.setField(exportingJob, "createTimeMs", currentTime - 3000); + Deencapsulation.setField(exportingJob, "state", ExportJobState.EXPORTING); + isolatedExportMgr.unprotectAddJob(exportingJob); + + ExportJob finishedJob = makeExportJob(1003, "finished"); + Deencapsulation.setField(finishedJob, "createTimeMs", currentTime - 2000); + Deencapsulation.setField(finishedJob, "state", ExportJobState.FINISHED); + isolatedExportMgr.unprotectAddJob(finishedJob); + + ExportJob cancelledJob = makeExportJob(1004, "cancelled"); + Deencapsulation.setField(cancelledJob, "createTimeMs", currentTime - 1000); + Deencapsulation.setField(cancelledJob, "state", ExportJobState.CANCELLED); + isolatedExportMgr.unprotectAddJob(cancelledJob); + + isolatedExportMgr.removeOldExportJobs(); + + Assert.assertEquals(2, isolatedExportMgr.getJobs().size()); + Assert.assertNotNull(isolatedExportMgr.getJob(pendingJob.getId())); + Assert.assertNotNull(isolatedExportMgr.getJob(exportingJob.getId())); + Assert.assertNull(isolatedExportMgr.getJob(finishedJob.getId())); + Assert.assertNull(isolatedExportMgr.getJob(cancelledJob.getId())); + } finally { + Config.max_export_history_job_num = originalMaxHistoryJobNum; + } + } + + @Test + public void testRemoveOldExportJobsKeepsRunningJobsWhenOverLimit() { + ExportMgr isolatedExportMgr = new ExportMgr(); + int originalMaxHistoryJobNum = Config.max_export_history_job_num; + Config.max_export_history_job_num = 1; + try { + long currentTime = System.currentTimeMillis(); + ExportJob pendingJob = makeExportJob(2001, "pending-over-limit"); + Deencapsulation.setField(pendingJob, "createTimeMs", currentTime - 3000); + Deencapsulation.setField(pendingJob, "state", ExportJobState.PENDING); + isolatedExportMgr.unprotectAddJob(pendingJob); + + ExportJob exportingJob = makeExportJob(2002, "exporting-over-limit"); + Deencapsulation.setField(exportingJob, "createTimeMs", currentTime - 2000); + Deencapsulation.setField(exportingJob, "state", ExportJobState.EXPORTING); + isolatedExportMgr.unprotectAddJob(exportingJob); + + ExportJob inQueueJob = makeExportJob(2003, "in-queue-over-limit"); + Deencapsulation.setField(inQueueJob, "createTimeMs", currentTime - 1000); + Deencapsulation.setField(inQueueJob, "state", ExportJobState.IN_QUEUE); + isolatedExportMgr.unprotectAddJob(inQueueJob); + + isolatedExportMgr.removeOldExportJobs(); + + Assert.assertEquals(3, isolatedExportMgr.getJobs().size()); + Assert.assertEquals(ExportJobState.PENDING, isolatedExportMgr.getJob(pendingJob.getId()).getState()); + Assert.assertEquals(ExportJobState.EXPORTING, isolatedExportMgr.getJob(exportingJob.getId()).getState()); + Assert.assertEquals(ExportJobState.IN_QUEUE, isolatedExportMgr.getJob(inQueueJob.getId()).getState()); + } finally { + Config.max_export_history_job_num = originalMaxHistoryJobNum; + } + } + private ExportJob makeExportJob(long id, String label) { ExportJob job1 = new ExportJob(id); Deencapsulation.setField(job1, "label", label);