This is an automated email from the ASF dual-hosted git repository. morningman pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push: new 20e07b7a0ba [fix](export) remove export task executor in TransientTaskExecutor (#42880) (#42950) 20e07b7a0ba is described below commit 20e07b7a0ba590ee9e87033557912b4093fbd1a8 Author: Mingyu Chen (Rayner) <morning...@163.com> AuthorDate: Thu Oct 31 15:07:29 2024 +0800 [fix](export) remove export task executor in TransientTaskExecutor (#42880) (#42950) cherry-pick #42880 --- .../main/java/org/apache/doris/catalog/Env.java | 8 ----- .../main/java/org/apache/doris/load/ExportJob.java | 14 ++++---- .../main/java/org/apache/doris/load/ExportMgr.java | 3 +- .../doris/scheduler/disruptor/TaskHandler.java | 2 ++ .../scheduler/manager/TransientTaskManager.java | 15 +++++++- .../scheduler/registry/ExportTaskRegister.java | 40 ---------------------- .../doris/analysis/CancelExportStmtTest.java | 1 - 7 files changed, 24 insertions(+), 59 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java index ef23713a501..cd0c0e80d8f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @@ -254,7 +254,6 @@ import org.apache.doris.resource.workloadschedpolicy.WorkloadRuntimeStatusMgr; import org.apache.doris.resource.workloadschedpolicy.WorkloadSchedPolicyMgr; import org.apache.doris.resource.workloadschedpolicy.WorkloadSchedPolicyPublisher; import org.apache.doris.scheduler.manager.TransientTaskManager; -import org.apache.doris.scheduler.registry.ExportTaskRegister; import org.apache.doris.service.ExecuteEnv; import org.apache.doris.service.FrontendOptions; import org.apache.doris.statistics.AnalysisManager; @@ -395,7 +394,6 @@ public class Env { private ExternalMetaIdMgr externalMetaIdMgr; private MetastoreEventsProcessor metastoreEventsProcessor; - private ExportTaskRegister exportTaskRegister; private JobManager<? extends AbstractJob<?, ?>, ?> jobManager; private LabelProcessor labelProcessor; private TransientTaskManager transientTaskManager; @@ -709,7 +707,6 @@ public class Env { this.jobManager = new JobManager<>(); this.labelProcessor = new LabelProcessor(); this.transientTaskManager = new TransientTaskManager(); - this.exportTaskRegister = new ExportTaskRegister(transientTaskManager); this.replayedJournalId = new AtomicLong(0L); this.stmtIdCounter = new AtomicLong(0L); @@ -4425,11 +4422,6 @@ public class Env { return this.syncJobManager; } - - public ExportTaskRegister getExportTaskRegister() { - return exportTaskRegister; - } - public JobManager getJobManager() { return jobManager; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java index 33418531f2c..5fe9c482633 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java @@ -98,7 +98,6 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; @Data @@ -207,9 +206,7 @@ public class ExportJob implements Writable { // backend_address => snapshot path private List<Pair<TNetworkAddress, String>> snapshotPaths = Lists.newArrayList(); - private List<ExportTaskExecutor> jobExecutorList; - - private ConcurrentHashMap<Long, TransientTaskExecutor> taskIdToExecutor = new ConcurrentHashMap<>(); + private List<ExportTaskExecutor> jobExecutorList = Lists.newArrayList(); private Integer finishedTaskCount = 0; private List<List<OutfileInfo>> allOutfileInfo = Lists.newArrayList(); @@ -690,11 +687,11 @@ public class ExportJob implements Writable { } // we need cancel all task - taskIdToExecutor.keySet().forEach(id -> { + jobExecutorList.forEach(executor -> { try { - Env.getCurrentEnv().getTransientTaskManager().cancelMemoryTask(id); + Env.getCurrentEnv().getTransientTaskManager().cancelMemoryTask(executor.getId()); } catch (JobException e) { - LOG.warn("cancel export task {} exception: {}", id, e); + LOG.warn("cancel export task {} exception: {}", executor.getId(), e); } }); @@ -705,6 +702,7 @@ public class ExportJob implements Writable { setExportJobState(ExportJobState.CANCELLED); finishTimeMs = System.currentTimeMillis(); failMsg = new ExportFailMsg(type, msg); + jobExecutorList.clear(); if (FeConstants.runningUnitTest) { return; } @@ -749,6 +747,8 @@ public class ExportJob implements Writable { setExportJobState(ExportJobState.FINISHED); finishTimeMs = System.currentTimeMillis(); outfileInfo = GsonUtils.GSON.toJson(allOutfileInfo); + // Clear the jobExecutorList to release memory. + jobExecutorList.clear(); Env.getCurrentEnv().getEditLog().logExportUpdateState(id, ExportJobState.FINISHED); } 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 7dbe953cf9b..5636f1aaad3 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 @@ -118,8 +118,7 @@ public class ExportMgr { job.getBrokerDesc()); } job.getTaskExecutors().forEach(executor -> { - Long taskId = Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(executor); - job.getTaskIdToExecutor().put(taskId, executor); + Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(executor); }); Env.getCurrentEnv().getEditLog().logExportCreate(job); } finally { diff --git a/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskHandler.java b/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskHandler.java index de889c1b2e4..193f8ece9f7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskHandler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/scheduler/disruptor/TaskHandler.java @@ -68,6 +68,8 @@ public class TaskHandler implements WorkHandler<TaskEvent> { taskExecutor.execute(); } catch (JobException e) { log.warn("Memory task execute failed, taskId: {}, msg : {}", taskId, e.getMessage()); + } finally { + transientTaskManager.removeMemoryTask(taskId); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java b/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java index 51edd4af318..7461399c8eb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/scheduler/manager/TransientTaskManager.java @@ -22,10 +22,13 @@ import org.apache.doris.scheduler.exception.JobException; import org.apache.doris.scheduler.executor.TransientTaskExecutor; import lombok.Setter; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.util.concurrent.ConcurrentHashMap; public class TransientTaskManager { + private static final Logger LOG = LogManager.getLogger(TransientTaskManager.class); /** * key: taskId * value: memory task executor of this task @@ -57,10 +60,20 @@ public class TransientTaskManager { Long taskId = executor.getId(); taskExecutorMap.put(taskId, executor); disruptor.tryPublishTask(taskId); + LOG.info("add memory task, taskId: {}", taskId); return taskId; } public void cancelMemoryTask(Long taskId) throws JobException { - taskExecutorMap.get(taskId).cancel(); + try { + taskExecutorMap.get(taskId).cancel(); + } finally { + removeMemoryTask(taskId); + } + } + + public void removeMemoryTask(Long taskId) { + taskExecutorMap.remove(taskId); + LOG.info("remove memory task, taskId: {}", taskId); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/scheduler/registry/ExportTaskRegister.java b/fe/fe-core/src/main/java/org/apache/doris/scheduler/registry/ExportTaskRegister.java deleted file mode 100644 index 0241f57fea0..00000000000 --- a/fe/fe-core/src/main/java/org/apache/doris/scheduler/registry/ExportTaskRegister.java +++ /dev/null @@ -1,40 +0,0 @@ -// 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.scheduler.registry; - -import org.apache.doris.scheduler.exception.JobException; -import org.apache.doris.scheduler.executor.TransientTaskExecutor; -import org.apache.doris.scheduler.manager.TransientTaskManager; - -public class ExportTaskRegister implements TransientTaskRegister { - private final TransientTaskManager transientTaskManager; - - public ExportTaskRegister(TransientTaskManager transientTaskManager) { - this.transientTaskManager = transientTaskManager; - } - - @Override - public Long registerTask(TransientTaskExecutor executor) { - return transientTaskManager.addMemoryTask(executor); - } - - @Override - public void cancelTask(Long taskId) throws JobException { - transientTaskManager.cancelMemoryTask(taskId); - } -} diff --git a/fe/fe-core/src/test/java/org/apache/doris/analysis/CancelExportStmtTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/CancelExportStmtTest.java index 2d188230d8b..4ff15653fa0 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/analysis/CancelExportStmtTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/analysis/CancelExportStmtTest.java @@ -234,7 +234,6 @@ public class CancelExportStmtTest extends TestWithFeService { exportMgr.unprotectAddJob(job3); exportMgr.unprotectAddJob(job4); - // cancel export job where state = "PENDING" Assert.assertTrue(job1.getState() == ExportJobState.PENDING); SlotRef stateSlotRef = new SlotRef(null, "state"); --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org