fix: The thread pool that is not managed by spring will also be destroyed (#990) (#1003)

* fix: The thread pool that is not managed by spring will also be destroyed (#990)

* optimize code and comments

hippo4j-core/src/main/java/cn/hippo4j/core/executor/DynamicThreadPoolExecutor.java

@@ -37,6 +37,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -45,6 +46,12 @@ import java.util.concurrent.atomic.AtomicLong;
 @Slf4j
 public class DynamicThreadPoolExecutor extends ExtensibleThreadPoolExecutor implements DisposableBean {
 
+    /**
+     * A flag used to indicate whether destroy() method has been called,
+     * after the flag is set to false, calling destroy() method again will not take effect
+     */
+    private final AtomicBoolean active;
+
     /**
      * Wait for tasks to complete on shutdown
      */
@@ -92,11 +99,22 @@ public class DynamicThreadPoolExecutor extends ExtensibleThreadPoolExecutor impl
                 threadPoolId, new DefaultThreadPoolPluginManager().setPluginComparator(AnnotationAwareOrderComparator.INSTANCE),
                 corePoolSize, maximumPoolSize, keepAliveTime, unit,
                 blockingQueue, threadFactory, rejectedExecutionHandler);
-        log.info("Initializing ExecutorService {}", threadPoolId);
+        log.info("Initializing ExecutorService '{}'", threadPoolId);
         this.waitForTasksToCompleteOnShutdown = waitForTasksToCompleteOnShutdown;
         // Init default plugins.
         new DefaultThreadPoolPluginRegistrar(executeTimeOut, awaitTerminationMillis)
                 .doRegister(this);
+        this.active = new AtomicBoolean(true);
+    }
+
+    /**
+     * <p>Whether the current instance is in the active state. <br />
+     * It returns false when the xx method is called at least once.
+     *
+     * @return true if current instance is in the active state, false otherwise
+     */
+    public boolean isActive() {
+        return active.get();
     }
 
     /**
@@ -104,12 +122,21 @@ public class DynamicThreadPoolExecutor extends ExtensibleThreadPoolExecutor impl
      */
     @Override
     public void destroy() {
+        // instance has been destroyed, not need to call this method again
+        if (!isActive()) {
+            log.warn("Failed to destroy ExecutorService '{}' because it has already been destroyed", getThreadPoolId());
+            return;
+        }
         if (isWaitForTasksToCompleteOnShutdown()) {
             super.shutdown();
         } else {
             super.shutdownNow();
         }
         getThreadPoolPluginManager().clear();
+        log.info("ExecutorService '{}' has been destroyed", getThreadPoolId());
+
+        // modify the flag to false avoid the method being called repeatedly
+        active.set(false);
     }
 
     /**

hippo4j-core/src/test/java/cn/hippo4j/core/executor/DynamicThreadPoolExecutorTest.java

@@ -95,7 +95,7 @@ public class DynamicThreadPoolExecutorTest {
         });
         executor.destroy();
 
-        // waitting for terminated
+        // waiting for terminated
         while (!executor.isTerminated()) {
         } ;
         Assert.assertEquals(2, count.get());
@@ -119,9 +119,9 @@ public class DynamicThreadPoolExecutorTest {
         });
         executor.destroy();
 
-        // waitting for terminated
+        // waiting for terminated
         while (!executor.isTerminated()) {
-        } ;
+        }
         Assert.assertEquals(1, count.get());
     }
 
@@ -157,4 +157,21 @@ public class DynamicThreadPoolExecutorTest {
         Assert.assertFalse(executor.isWaitForTasksToCompleteOnShutdown());
     }
 
+    @Test
+    public void testIsActive() {
+        DynamicThreadPoolExecutor executor = new DynamicThreadPoolExecutor(
+                1, 1, 1000L, TimeUnit.MILLISECONDS,
+                1000L, true, 1000L,
+                new ArrayBlockingQueue<>(1), "test", Thread::new, new ThreadPoolExecutor.DiscardOldestPolicy());
+        Assert.assertTrue(executor.isActive());
+
+        // waiting for terminated
+        executor.destroy();
+        while (!executor.isTerminated()) {
+        }
+        Assert.assertFalse(executor.isActive());
+        executor.destroy();
+        Assert.assertFalse(executor.isActive());
+    }
+
 }

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/config/DynamicThreadPoolAutoConfiguration.java

@@ -57,6 +57,7 @@ import cn.hippo4j.springboot.starter.remote.HttpAgent;
 import cn.hippo4j.springboot.starter.remote.HttpScheduledHealthCheck;
 import cn.hippo4j.springboot.starter.remote.ServerHealthCheck;
 import cn.hippo4j.springboot.starter.remote.ServerHttpAgent;
+import cn.hippo4j.springboot.starter.support.AdaptedThreadPoolDestroyPostProcessor;
 import cn.hippo4j.springboot.starter.support.DynamicThreadPoolConfigService;
 import cn.hippo4j.springboot.starter.support.DynamicThreadPoolPostProcessor;
 import cn.hippo4j.springboot.starter.support.ThreadPoolPluginRegisterPostProcessor;
@@ -68,6 +69,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.info.BuildProperties;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.Ordered;
@@ -120,6 +122,11 @@ public class DynamicThreadPoolAutoConfiguration {
         return new DynamicThreadPoolConfigService(httpAgent, properties, notifyConfigBuilder, hippo4jBaseSendMessageService, dynamicThreadPoolSubscribeConfig);
     }
 
+    @Bean
+    public AdaptedThreadPoolDestroyPostProcessor adaptedThreadPoolDestroyPostProcessor(ApplicationContext applicationContext) {
+        return new AdaptedThreadPoolDestroyPostProcessor(applicationContext);
+    }
+
     @Bean
     @SuppressWarnings("all")
     public DynamicThreadPoolPostProcessor threadPoolBeanPostProcessor(HttpAgent httpAgent,

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/support/AdaptedThreadPoolDestroyPostProcessor.java

@@ -0,0 +1,91 @@
+/*
+ * 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 cn.hippo4j.springboot.starter.support;
+
+import cn.hippo4j.core.executor.DynamicThreadPoolExecutor;
+import cn.hippo4j.core.executor.DynamicThreadPoolWrapper;
+import cn.hippo4j.core.executor.manage.GlobalThreadPoolManage;
+import cn.hippo4j.core.executor.support.adpter.DynamicThreadPoolAdapter;
+import cn.hippo4j.core.executor.support.adpter.DynamicThreadPoolAdapterChoose;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.DestructionAwareBeanPostProcessor;
+import org.springframework.context.ApplicationContext;
+
+import java.util.Optional;
+
+/**
+ * <p>Adapted thread pool destroy post processor. <br />
+ * The processor is used to destroy the internal {@link DynamicThreadPoolExecutor} instance
+ * in the instance adapted by {@link DynamicThreadPoolAdapter} in the spring context.
+ *
+ * @see DynamicThreadPoolAdapter
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class AdaptedThreadPoolDestroyPostProcessor implements DestructionAwareBeanPostProcessor {
+
+    /**
+     * Application context.
+     */
+    private final ApplicationContext applicationContext;
+
+    /**
+     * If {@link DynamicThreadPoolAdapterChoose#match} method returns true,
+     * try to destroy its internal {@link DynamicThreadPoolExecutor} instance.
+     *
+     * @param bean the bean instance to check
+     * @return {@code true} if {@link DynamicThreadPoolAdapterChoose#match} method returns true, false otherwise
+     * @see DynamicThreadPoolAdapterChoose#match
+     */
+    @Override
+    public boolean requiresDestruction(Object bean) {
+        return DynamicThreadPoolAdapterChoose.match(bean);
+    }
+
+    /**
+     * If the internal {@link DynamicThreadPoolExecutor} instance in the adapted bean is not managed by spring,
+     * call its {@link DynamicThreadPoolExecutor#destroy()} directly.
+     *
+     * @param bean     the bean instance to be destroyed
+     * @param beanName the name of the bean
+     * @throws BeansException in case of errors
+     * @see DynamicThreadPoolExecutor#destroy()
+     */
+    @Override
+    public void postProcessBeforeDestruction(Object bean, String beanName) throws BeansException {
+        Optional.ofNullable(DynamicThreadPoolAdapterChoose.unwrap(bean))
+                .map(DynamicThreadPoolExecutor::getThreadPoolId)
+                // the internal thread pool is also managed by spring, no manual destruction required
+                .filter(applicationContext::containsBeanDefinition)
+                .map(GlobalThreadPoolManage::getExecutorService)
+                .ifPresent(executor -> destroyAdaptedThreadPoolExecutor(beanName, executor));
+    }
+
+    private void destroyAdaptedThreadPoolExecutor(String beanName, DynamicThreadPoolWrapper executor) {
+        try {
+            if (log.isDebugEnabled()) {
+                log.debug("Destroy internal dynamic thread pool '{}' for bean '{}'", executor.getThreadPoolId(), beanName);
+            }
+            executor.destroy();
+        } catch (Exception e) {
+            log.warn("Failed to destroy internal dynamic thread pool '{}' for bean '{}'", executor.getThreadPoolId(), beanName);
+        }
+    }
+}