[AMORO-3141] When a master/backup switch occurs, close the `planExecu…

…tor` thread pool (#3142) (cherry picked from commit a8b2ff5) Signed-off-by: zhoujinsong <[email protected]>
apache · Oct 11, 2024 · 5200a86 · 5200a86
1 parent a26da28
commit 5200a86
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
@@ -199,7 +199,11 @@ public void dispose() {
       terminalManager.dispose();
       terminalManager = null;
     }
-    optimizingService = null;
+    if (optimizingService != null) {
+      LOG.info("Stopping optimizing service...");
+      optimizingService.dispose();
+      optimizingService = null;
+    }
 
     EventsManager.dispose();
     MetricManager.dispose();

diff --git a/...-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java b/...-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
@@ -69,7 +69,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.DelayQueue;
 import java.util.concurrent.Delayed;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -99,7 +99,7 @@ public class DefaultOptimizingService extends StatedPersistentBase
   private final OptimizerKeeper optimizerKeeper = new OptimizerKeeper();
   private final TableService tableService;
   private final RuntimeHandlerChain tableHandlerChain;
-  private final Executor planExecutor;
+  private final ExecutorService planExecutor;
 
   public DefaultOptimizingService(Configurations serviceConfig, DefaultTableService tableService) {
     this.optimizerTouchTimeout = serviceConfig.getLong(AmoroManagementConf.OPTIMIZER_HB_TIMEOUT);
@@ -385,6 +385,7 @@ public void dispose() {
     optimizingQueueByGroup.clear();
     optimizingQueueByToken.clear();
     authOptimizers.clear();
+    planExecutor.shutdown();
   }
 
   public boolean canDeleteResourceGroup(String name) {