This is an automated email from the ASF dual-hosted git repository.
ruanwenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new a5cddc8ffc [Fix-18222][JdbcRegistry] Reuse a singleton scheduler
executor in JdbcRegistryThreadFactory (#18223)
a5cddc8ffc is described below
commit a5cddc8ffceaefefa92cbf9de8048654a86eabe3
Author: Wenjun Ruan <[email protected]>
AuthorDate: Sun May 10 20:12:17 2026 +0800
[Fix-18222][JdbcRegistry] Reuse a singleton scheduler executor in
JdbcRegistryThreadFactory (#18223)
* [Fix-18222][JdbcRegistry] Reuse a singleton scheduler executor in
JdbcRegistryThreadFactory
JdbcRegistryThreadFactory#getDefaultSchedulerThreadExecutor returned a
freshly created ScheduledExecutorService on every call, so the four
scheduleWithFixedDelay sites in JdbcRegistryServer#start and
JdbcRegistryDataManager#start each ran on their own pool, while the
shutdown/shutdownNow calls in JdbcRegistry#close and
JdbcRegistryServer#close closed yet another empty pool instead of the
running ones. Daemon threads kept this from leaking on JVM exit, but the
logic was inconsistent and the process held 4x the configured scheduled
threads.
Cache the executor as a lazy singleton (double-checked locking with
volatile) so all callers share one pool and the close calls land on the
actually-running pool.
* [Fix-18222][JdbcRegistry] Bind scheduler executor to JdbcRegistryServer
instance
The JdbcRegistryThreadFactory class held a JVM-global singleton
scheduled executor. Multiple JdbcRegistryServer instances (tests,
embedded StandaloneServer) shared the same pool, blurring lifecycle
ownership and forcing JdbcRegistry#close() to shutdownNow() on a pool
that other instances might still be using.
Move the executor onto JdbcRegistryServer as an instance field, inject
it into JdbcRegistryDataManager via the constructor, and shut it down
in JdbcRegistryServer#close(). The state-guard at the top of the
periodic tasks short-circuits any task that observes STOPPED, so a
graceful shutdown() — combined with daemon threads — is enough to stop
the JVM stalling.
Delete the now-unused JdbcRegistryThreadFactory class and its test;
the test only covered JDK ScheduledExecutorService basics and the
static singleton identity that no longer exists.
---
.../plugin/registry/jdbc/JdbcRegistry.java | 1 -
.../registry/jdbc/JdbcRegistryThreadFactory.java | 35 ------------------
.../jdbc/server/JdbcRegistryDataManager.java | 12 ++++--
.../registry/jdbc/server/JdbcRegistryServer.java | 16 +++++---
.../jdbc/JdbcRegistryThreadFactoryTest.java | 43 ----------------------
5 files changed, 19 insertions(+), 88 deletions(-)
diff --git
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistry.java
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistry.java
index f9614d3c38..016af846bd 100644
---
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistry.java
+++
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistry.java
@@ -217,7 +217,6 @@ public final class JdbcRegistry implements Registry {
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) {
diff --git
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactory.java
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactory.java
deleted file mode 100644
index 7296955d52..0000000000
---
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactory.java
+++ /dev/null
@@ -1,35 +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.dolphinscheduler.plugin.registry.jdbc;
-
-import org.apache.dolphinscheduler.common.thread.ThreadUtils;
-
-import java.util.concurrent.ScheduledExecutorService;
-
-import lombok.NoArgsConstructor;
-
-@NoArgsConstructor(access = lombok.AccessLevel.PRIVATE)
-public class JdbcRegistryThreadFactory {
-
- public static ScheduledExecutorService getDefaultSchedulerThreadExecutor()
{
- final String threadNameFormat =
"ds-jdbc-registry-default-scheduler-thread-%d";
- final int threadSize = Runtime.getRuntime().availableProcessors();
- return ThreadUtils.newDaemonScheduledExecutorService(threadNameFormat,
threadSize);
- }
-
-}
diff --git
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryDataManager.java
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryDataManager.java
index 08eb053176..b49ebc6b2f 100644
---
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryDataManager.java
+++
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryDataManager.java
@@ -20,7 +20,6 @@ package
org.apache.dolphinscheduler.plugin.registry.jdbc.server;
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.Date;
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,6 +61,8 @@ public class JdbcRegistryDataManager
private final TransactionTemplate jdbcRegistryTransactionTemplate;
+ private final ScheduledExecutorService schedulerThreadExecutor;
+
private final List<RegistryRowChangeListener<JdbcRegistryDataDTO>>
registryRowChangeListeners;
private long lastDetectedJdbcRegistryDataChangeEventId = -1;
@@ -68,11 +70,13 @@ public class JdbcRegistryDataManager
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<>();
}
@@ -80,13 +84,13 @@ public class JdbcRegistryDataManager
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(),
diff --git
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java
index e3070bdec8..3e8b11c813 100644
---
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java
+++
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java
@@ -19,8 +19,8 @@ package
org.apache.dolphinscheduler.plugin.registry.jdbc.server;
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.Optional;
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 class JdbcRegistryServer implements
IJdbcRegistryServer {
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,7 +113,7 @@ public class JdbcRegistryServer implements
IJdbcRegistryServer {
// 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(),
@@ -115,7 +121,7 @@ public class JdbcRegistryServer implements
IJdbcRegistryServer {
jdbcRegistryDataManager.start();
jdbcRegistryServerState = JdbcRegistryServerState.STARTED;
doTriggerOnConnectedListener();
-
JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
+ schedulerThreadExecutor.scheduleWithFixedDelay(
this::refreshClientsHeartbeat,
0,
jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis(),
@@ -250,7 +256,7 @@ public class JdbcRegistryServer implements
IJdbcRegistryServer {
@Override
public void close() {
jdbcRegistryServerState = JdbcRegistryServerState.STOPPED;
-
JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().shutdown();
+ schedulerThreadExecutor.shutdown();
List<Long> clientIds = jdbcRegistryClients.stream()
.map(IJdbcRegistryClient::getJdbcRegistryClientIdentify)
.map(JdbcRegistryClientIdentify::getClientId)
diff --git
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/test/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactoryTest.java
b/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/test/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactoryTest.java
deleted file mode 100644
index cd5e085949..0000000000
---
a/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/test/java/org/apache/dolphinscheduler/plugin/registry/jdbc/JdbcRegistryThreadFactoryTest.java
+++ /dev/null
@@ -1,43 +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.dolphinscheduler.plugin.registry.jdbc;
-
-import static com.google.common.truth.Truth.assertThat;
-import static org.awaitility.Awaitility.await;
-
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.junit.jupiter.api.Test;
-
-class JdbcRegistryThreadFactoryTest {
-
- @Test
- void getDefaultSchedulerThreadExecutor() {
- ScheduledExecutorService schedulerThreadExecutor =
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor();
- AtomicInteger atomicInteger = new AtomicInteger(0);
- for (int i = 0; i < 100; i++) {
-
schedulerThreadExecutor.scheduleWithFixedDelay(atomicInteger::incrementAndGet,
0, 1, TimeUnit.SECONDS);
- }
- await()
- .atMost(10, TimeUnit.SECONDS)
- .untilAsserted(() ->
assertThat(atomicInteger.get()).isEqualTo(100));
- }
-}
