[Fix-18222][JdbcRegistry] Reuse a singleton scheduler executor in JdbcRegistryThreadFactory

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistry.java

@@ -217,7 +217,6 @@ public boolean releaseLock(String key) {
     public void close() {
         log.info("Closing JdbcRegistry...");
         // remove the current Ephemeral node, if can connect to jdbc
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().shutdownNow();
         try (final JdbcRegistryClient closed1 = jdbcRegistryClient) {
             // ignore
         } catch (Exception e) {

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryDataManager.java

@@ -20,7 +20,6 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import org.apache.dolphinscheduler.plugin.registry.jdbc.JdbcRegistryProperties;
-import org.apache.dolphinscheduler.plugin.registry.jdbc.JdbcRegistryThreadFactory;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.KeyUtils;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.model.DTO.DataType;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.model.DTO.JdbcRegistryDataChangeEventDTO;
@@ -36,6 +35,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -61,32 +61,36 @@ public class JdbcRegistryDataManager
 
     private final TransactionTemplate jdbcRegistryTransactionTemplate;
 
+    private final ScheduledExecutorService schedulerThreadExecutor;
+
     private final List<RegistryRowChangeListener<JdbcRegistryDataDTO>> registryRowChangeListeners;
 
     private long lastDetectedJdbcRegistryDataChangeEventId = -1;
 
     public JdbcRegistryDataManager(JdbcRegistryProperties registryProperties,
                                    JdbcRegistryDataRepository jdbcRegistryDataRepository,
                                    JdbcRegistryDataChangeEventRepository jdbcRegistryDataChangeEventRepository,
-                                   TransactionTemplate jdbcRegistryTransactionTemplate) {
+                                   TransactionTemplate jdbcRegistryTransactionTemplate,
+                                   ScheduledExecutorService schedulerThreadExecutor) {
         this.registryProperties = registryProperties;
         this.jdbcRegistryDataChangeEventRepository = jdbcRegistryDataChangeEventRepository;
         this.jdbcRegistryDataRepository = jdbcRegistryDataRepository;
         this.jdbcRegistryTransactionTemplate = jdbcRegistryTransactionTemplate;
+        this.schedulerThreadExecutor = schedulerThreadExecutor;
         this.registryRowChangeListeners = new CopyOnWriteArrayList<>();
     }
 
     @Override
     public void start() {
         this.lastDetectedJdbcRegistryDataChangeEventId =
                 jdbcRegistryDataChangeEventRepository.getMaxJdbcRegistryDataChangeEventId();
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
+        schedulerThreadExecutor.scheduleWithFixedDelay(
                 this::detectJdbcRegistryDataChangeEvent,
                 registryProperties.getHeartbeatRefreshInterval().toMillis(),
                 registryProperties.getHeartbeatRefreshInterval().toMillis(),
                 TimeUnit.MILLISECONDS);
 
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
+        schedulerThreadExecutor.scheduleWithFixedDelay(
                 this::purgeHistoryJdbcRegistryDataChangeEvent,
                 0,
                 Duration.ofHours(keepJdbcRegistryDataChangeEventHours).toHours(),

dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java

@@ -19,8 +19,8 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.JdbcRegistryProperties;
-import org.apache.dolphinscheduler.plugin.registry.jdbc.JdbcRegistryThreadFactory;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.client.IJdbcRegistryClient;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.client.JdbcRegistryClientIdentify;
 import org.apache.dolphinscheduler.plugin.registry.jdbc.model.DTO.DataType;
@@ -43,6 +43,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -78,6 +79,8 @@ public class JdbcRegistryServer implements IJdbcRegistryServer {
     private final Map<JdbcRegistryClientIdentify, JdbcRegistryClientHeartbeatDTO> jdbcRegistryClientDTOMap =
             new ConcurrentHashMap<>();
 
+    private final ScheduledExecutorService schedulerThreadExecutor;
+
     private Long lastSuccessHeartbeat;
 
     public JdbcRegistryServer(JdbcRegistryDataRepository jdbcRegistryDataRepository,
@@ -89,9 +92,12 @@ public JdbcRegistryServer(JdbcRegistryDataRepository jdbcRegistryDataRepository,
         this.jdbcRegistryLockRepository = checkNotNull(jdbcRegistryLockRepository);
         this.jdbcRegistryClientRepository = checkNotNull(jdbcRegistryClientRepository);
         this.jdbcRegistryProperties = checkNotNull(jdbcRegistryProperties);
+        this.schedulerThreadExecutor = ThreadUtils.newDaemonScheduledExecutorService(
+                "ds-jdbc-registry-default-scheduler-thread-%d",
+                Runtime.getRuntime().availableProcessors());
         this.jdbcRegistryDataManager = new JdbcRegistryDataManager(
                 jdbcRegistryProperties, jdbcRegistryDataRepository, jdbcRegistryDataChangeEventRepository,
-                transactionTemplate);
+                transactionTemplate, schedulerThreadExecutor);
         this.jdbcRegistryLockManager = new JdbcRegistryLockManager(
                 jdbcRegistryProperties, jdbcRegistryLockRepository);
         this.jdbcRegistryServerState = JdbcRegistryServerState.INIT;
@@ -107,15 +113,15 @@ public void start() {
         // Start the Purge thread
         // The Purge thread will clear the invalidated data
         purgeInvalidJdbcRegistryMetadata();
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
+        schedulerThreadExecutor.scheduleWithFixedDelay(
                 this::purgeInvalidJdbcRegistryMetadata,
                 jdbcRegistryProperties.getSessionTimeout().toMillis(),
                 jdbcRegistryProperties.getSessionTimeout().toMillis(),
                 TimeUnit.MILLISECONDS);
         jdbcRegistryDataManager.start();
         jdbcRegistryServerState = JdbcRegistryServerState.STARTED;
         doTriggerOnConnectedListener();
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
+        schedulerThreadExecutor.scheduleWithFixedDelay(
                 this::refreshClientsHeartbeat,
                 0,
                 jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis(),
@@ -250,7 +256,7 @@ public void releaseJdbcRegistryLock(Long clientId, String lockKey) {
     @Override
     public void close() {
         jdbcRegistryServerState = JdbcRegistryServerState.STOPPED;
-        JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().shutdown();
+        schedulerThreadExecutor.shutdown();
         List<Long> clientIds = jdbcRegistryClients.stream()
                 .map(IJdbcRegistryClient::getJdbcRegistryClientIdentify)
                 .map(JdbcRegistryClientIdentify::getClientId)