feat: Add dedicated thread pool for RestTemplate in alert notificati…

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatch.java

@@ -21,8 +21,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.AlerterWorkerPool;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.apache.hertzbeat.alert.config.AlertSseManager;
 import org.apache.hertzbeat.common.entity.alerter.GroupAlert;
 import org.apache.hertzbeat.common.entity.alerter.NoticeReceiver;
@@ -48,15 +51,20 @@ public class AlertNoticeDispatch {
     private final Map<Byte, AlertNotifyHandler> alertNotifyHandlerMap;
     private final PluginRunner pluginRunner;
     private final AlertSseManager emitterManager;
+    private final Executor restTemplateThreadPool;
 
     public AlertNoticeDispatch(AlerterWorkerPool workerPool,
                                NoticeConfigService noticeConfigService,
                                AlertStoreHandler alertStoreHandler,
-                               List<AlertNotifyHandler> alertNotifyHandlerList, PluginRunner pluginRunner, AlertSseManager emitterManager) {
+                               List<AlertNotifyHandler> alertNotifyHandlerList, 
+                               PluginRunner pluginRunner, 
+                               AlertSseManager emitterManager,
+                               @Qualifier("restTemplateThreadPool") Executor restTemplateThreadPool) {
         this.workerPool = workerPool;
         this.noticeConfigService = noticeConfigService;
         this.alertStoreHandler = alertStoreHandler;
         this.pluginRunner = pluginRunner;
+        this.restTemplateThreadPool = restTemplateThreadPool;
         alertNotifyHandlerMap = Maps.newHashMapWithExpectedSize(alertNotifyHandlerList.size());
         this.emitterManager = emitterManager;
         alertNotifyHandlerList.forEach(r -> alertNotifyHandlerMap.put(r.type(), r));
@@ -121,14 +129,31 @@ public void dispatchAlarm(GroupAlert groupAlert) {
     }
 
     private void sendNotify(GroupAlert alert) {
-        matchNoticeRulesByAlert(alert).ifPresent(noticeRules -> noticeRules.forEach(rule -> workerPool.executeNotify(() -> rule.getReceiverId()
-                .forEach(receiverId -> {
-                    try {
-                        sendNoticeMsg(getOneReceiverById(receiverId),
-                                getOneTemplateById(rule.getTemplateId()), alert);
-                    } catch (AlertNoticeException e) {
-                        log.warn("DispatchTask sendNoticeMsg error, message: {}", e.getMessage());
-                    }
-                }))));
+        matchNoticeRulesByAlert(alert).ifPresent(noticeRules -> noticeRules.forEach(rule -> workerPool.executeNotify(() -> {
+            List<CompletableFuture<Void>> futures = rule.getReceiverId().stream()
+                    .map(receiverId -> sendNoticeAsync(getOneReceiverById(receiverId),
+                            getOneTemplateById(rule.getTemplateId()), alert))
+                    .toList();
+
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+                    .whenComplete((result, exception) -> {
+                        if (exception != null) {
+                            log.warn("Some async notifications failed", exception);
+                        } else {
+                            log.debug("All notifications completed for alert: {}", alert.getGroupLabels());
+                        }
+                    });
+        })));
+    }
+
+    private CompletableFuture<Void> sendNoticeAsync(NoticeReceiver receiver, NoticeTemplate template, GroupAlert alert) {
+        return CompletableFuture.runAsync(() -> {
+            try {
+                sendNoticeMsg(receiver, template, alert);
+            } catch (AlertNoticeException e) {
+                log.warn("Async notification failed for receiver {}: {}", receiver.getName(), e.getMessage());
+                throw new RuntimeException(e);
+            }
+        }, restTemplateThreadPool);
     }
 }

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/notice/impl/AbstractAlertNotifyHandlerImpl.java

@@ -26,6 +26,8 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.ResourceBundle;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.AlerterProperties;
 import org.apache.hertzbeat.common.entity.alerter.GroupAlert;
@@ -34,6 +36,7 @@
 import org.apache.hertzbeat.common.util.ResourceBundleUtil;
 import org.apache.hertzbeat.alert.notice.AlertNotifyHandler;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.event.EventListener;
 import org.springframework.ui.freemarker.FreeMarkerTemplateUtils;
 import org.springframework.web.client.RestTemplate;
@@ -50,6 +53,9 @@ abstract class AbstractAlertNotifyHandlerImpl implements AlertNotifyHandler {
     protected RestTemplate restTemplate;
     @Autowired
     protected AlerterProperties alerterProperties;
+    @Autowired
+    @Qualifier("restTemplateThreadPool")
+    protected Executor restTemplateThreadPool;
 
 
     protected String renderContent(NoticeTemplate noticeTemplate, GroupAlert alert) throws TemplateException, IOException {
@@ -113,6 +119,19 @@ protected String escapeJsonStr(String jsonStr){
         return sb.toString();
     }
 
+    protected CompletableFuture<Void> sendAsync(org.apache.hertzbeat.common.entity.alerter.NoticeReceiver receiver, 
+                                                 org.apache.hertzbeat.common.entity.alerter.NoticeTemplate noticeTemplate, 
+                                                 GroupAlert alert) {
+        return CompletableFuture.runAsync(() -> {
+            try {
+                send(receiver, noticeTemplate, alert);
+            } catch (Exception e) {
+                log.error("Async alert notification failed", e);
+                throw new RuntimeException(e);
+            }
+        }, restTemplateThreadPool);
+    }
+
     @EventListener(SystemConfigChangeEvent.class)
     public void onEvent(SystemConfigChangeEvent event) {
         log.info("{} receive system config change event: {}.", this.getClass().getName(), event.getSource());

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/notice/AlertNoticeDispatchTest.java

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.util.concurrent.Executor;
 import java.util.Collections;
 import java.util.List;
 import org.apache.hertzbeat.alert.AlerterWorkerPool;
@@ -64,6 +65,9 @@ class AlertNoticeDispatchTest {
     @Mock
     private AlertSseManager emitterManager;
 
+    @Mock
+    private Executor restTemplateThreadPool;
+
     private AlertNoticeDispatch alertNoticeDispatch;
 
     private static final int DISPATCH_THREADS = 3;
@@ -82,7 +86,8 @@ void setUp() {
                 alertStoreHandler,
                 alertNotifyHandlerList,
                 pluginRunner,
-                emitterManager
+                emitterManager,
+                restTemplateThreadPool
         );
 
         receiver = NoticeReceiver.builder()

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/RestTemplateConfig.java

@@ -18,6 +18,8 @@
 package org.apache.hertzbeat.manager.config;
 
 import java.util.Collections;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import okhttp3.ConnectionPool;
 import okhttp3.OkHttpClient;
@@ -26,11 +28,11 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.client.ClientHttpRequestFactory;
 import org.springframework.http.client.OkHttp3ClientHttpRequestFactory;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.web.client.RestTemplate;
 
 /**
  * restTemplate config
- * todo thread pool
  */
 @Configuration
 public class RestTemplateConfig {
@@ -58,4 +60,16 @@ public ClientHttpRequestFactory simpleClientHttpRequestFactory() {
         );
     }
 
+    @Bean("restTemplateThreadPool")
+    public Executor restTemplateThreadPool() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(10);
+        executor.setMaxPoolSize(50);
+        executor.setQueueCapacity(200);
+        executor.setThreadNamePrefix("RestTemplate-");
+        executor.setKeepAliveSeconds(60);
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.initialize();
+        return executor;
+    }
 }