You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by li...@apache.org on 2023/03/02 03:44:42 UTC
[pulsar] branch branch-2.9 updated: [cherry-pick][branch-2.9] Add UncaughtExceptionHandler for every thread pool (#18211) (#19677)
This is an automated email from the ASF dual-hosted git repository.
linlin pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new 25e56386f0f [cherry-pick][branch-2.9] Add UncaughtExceptionHandler for every thread pool (#18211) (#19677)
25e56386f0f is described below
commit 25e56386f0f4ada59fe974a15da367e86d94a782
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Thu Mar 2 11:44:34 2023 +0800
[cherry-pick][branch-2.9] Add UncaughtExceptionHandler for every thread pool (#18211) (#19677)
Co-authored-by: feynmanlin <31...@qq.com>
---
.../org/apache/pulsar/broker/PulsarService.java | 11 ++++----
.../broker/TransactionMetadataStoreService.java | 4 +--
.../loadbalance/impl/ModularLoadManagerImpl.java | 5 ++--
.../loadbalance/impl/SimpleLoadManagerImpl.java | 4 +--
.../pulsar/broker/service/BrokerService.java | 31 +++++++++++-----------
.../metrics/PrometheusMetricsProvider.java | 5 ++--
.../pulsar/broker/tools/LoadReportCommand.java | 4 ++-
.../pulsar/client/impl/PulsarClientImpl.java | 3 +--
.../pulsar/client/util/ExecutorProvider.java | 22 ++++++++-------
.../pulsar/functions/instance/InstanceCache.java | 7 +++--
.../worker/ClusterServiceCoordinator.java | 11 ++++----
11 files changed, 56 insertions(+), 51 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index efcb2840556..052ee90b536 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -315,13 +315,13 @@ public class PulsarService implements AutoCloseable, ShutdownService {
this.config = config;
this.processTerminator = processTerminator;
this.loadManagerExecutor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-load-manager"));
+ .newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory("pulsar-load-manager"));
this.workerConfig = workerConfig;
this.functionWorkerService = functionWorkerService;
this.executor = Executors.newScheduledThreadPool(config.getNumExecutorThreadPoolSize(),
- new DefaultThreadFactory("pulsar"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar"));
this.cacheExecutor = Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(),
- new DefaultThreadFactory("zk-cache-callback"));
+ new ExecutorProvider.ExtendedThreadFactory("zk-cache-callback"));
if (config.isTransactionCoordinatorEnabled()) {
this.transactionExecutorProvider = new ExecutorProvider(this.getConfiguration()
@@ -581,7 +581,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private CompletableFuture<Void> addTimeoutHandling(CompletableFuture<Void> future) {
ScheduledExecutorService shutdownExecutor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory(getClass().getSimpleName() + "-shutdown"));
+ new ExecutorProvider.ExtendedThreadFactory(getClass().getSimpleName() + "-shutdown"));
FutureUtil.addTimeoutHandling(future,
Duration.ofMillis(Math.max(1L, getConfiguration().getBrokerShutdownTimeoutMs())),
shutdownExecutor, () -> FutureUtil.createTimeoutException("Timeout in close", getClass(), "close"));
@@ -1337,7 +1337,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
protected synchronized ScheduledExecutorService getCompactorExecutor() {
if (this.compactorExecutor == null) {
- compactorExecutor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("compaction"));
+ compactorExecutor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("compaction"));
}
return this.compactorExecutor;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
index 902546958c5..1c7008ab178 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
@@ -24,7 +24,6 @@ import static org.apache.pulsar.transaction.coordinator.proto.TxnStatus.COMMITTI
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.Collections;
import java.util.Deque;
import java.util.List;
@@ -53,6 +52,7 @@ import org.apache.pulsar.client.api.transaction.TransactionBufferClient;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException.ReachMaxPendingOpsException;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException.RequestTimeoutException;
import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.api.proto.TxnAction;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -95,7 +95,7 @@ public class TransactionMetadataStoreService {
private static final long HANDLE_PENDING_CONNECT_TIME_OUT = 30000L;
private final ThreadFactory threadFactory =
- new DefaultThreadFactory("transaction-coordinator-thread-factory");
+ new ExecutorProvider.ExtendedThreadFactory("transaction-coordinator-thread-factory");
public TransactionMetadataStoreService(TransactionMetadataStoreProvider transactionMetadataStoreProvider,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index d821e339944..b5af09ca395 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.loadbalance.impl;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
@@ -63,6 +62,7 @@ import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLo
import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
@@ -210,7 +210,8 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
loadData = new LoadData();
loadSheddingPipeline = new ArrayList<>();
preallocatedBundleToBroker = new ConcurrentHashMap<>();
- scheduler = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-modular-load-manager"));
+ scheduler = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-modular-load-manager"));
this.brokerToFailureDomainMap = Maps.newHashMap();
this.brokerTopicLoadingPredicate = new BrokerTopicLoadingPredicate() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
index ed766b1b930..415a778347b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/SimpleLoadManagerImpl.java
@@ -28,7 +28,6 @@ import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.collect.TreeMultimap;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Collections;
@@ -55,6 +54,7 @@ import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.PlacementStrategy;
import org.apache.pulsar.broker.loadbalance.ResourceUnit;
import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.BrokerTopicLoadingPredicate;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
import org.apache.pulsar.common.policies.data.ResourceQuota;
@@ -187,7 +187,7 @@ public class SimpleLoadManagerImpl implements LoadManager, Consumer<Notification
// Perform initializations which may be done without a PulsarService.
public SimpleLoadManagerImpl() {
scheduler = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-simple-load-manager"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-simple-load-manager"));
this.sortedRankings.set(new TreeMap<>());
this.currentLoadReports = new HashMap<>();
this.resourceUnitRankings = new HashMap<>();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index a9e5a769aac..efe7235802f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -127,6 +127,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.internal.PropertiesUtils;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.configuration.BindAddress;
import org.apache.pulsar.common.configuration.FieldContext;
@@ -305,32 +306,32 @@ public class BrokerService implements Closeable {
this.topicOrderedExecutor = OrderedExecutor.newBuilder()
.numThreads(pulsar.getConfiguration().getNumWorkerThreadsForNonPersistentTopic())
.name("broker-topic-workers").build();
- final DefaultThreadFactory acceptorThreadFactory = new DefaultThreadFactory("pulsar-acceptor");
+ final DefaultThreadFactory acceptorThreadFactory =
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-acceptor");
this.acceptorGroup = EventLoopUtil.newEventLoopGroup(
pulsar.getConfiguration().getNumAcceptorThreads(), false, acceptorThreadFactory);
this.workerGroup = eventLoopGroup;
- this.statsUpdater = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-stats-updater"));
+ this.statsUpdater = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-stats-updater"));
this.authorizationService = new AuthorizationService(
pulsar.getConfiguration(), pulsar().getPulsarResources());
pulsar.getLocalMetadataStore().registerListener(this::handleMetadataChanges);
pulsar.getConfigurationMetadataStore().registerListener(this::handleMetadataChanges);
- this.inactivityMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-inactivity-monitor"));
- this.messageExpiryMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-msg-expiry-monitor"));
+ this.inactivityMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-inactivity-monitor"));
+ this.messageExpiryMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-msg-expiry-monitor"));
this.compactionMonitor =
Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-compaction-monitor"));
- this.consumedLedgersMonitor = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("consumed-Ledgers-monitor"));
-
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-compaction-monitor"));
+ this.consumedLedgersMonitor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("consumed-Ledgers-monitor"));
this.backlogQuotaManager = new BacklogQuotaManager(pulsar);
- this.backlogQuotaChecker = Executors
- .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-backlog-quota-checker"));
+ this.backlogQuotaChecker = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-backlog-quota-checker"));
this.authenticationService = new AuthenticationService(pulsar.getConfiguration());
this.blockedDispatchers =
ConcurrentOpenHashSet.<PersistentDispatcherMultipleConsumers>newBuilder().build();
@@ -499,7 +500,7 @@ public class BrokerService implements Closeable {
int interval = pulsar().getConfiguration().getBrokerDeduplicationSnapshotFrequencyInSeconds();
if (interval > 0 && pulsar().getConfiguration().isBrokerDeduplicationEnabled()) {
this.deduplicationSnapshotMonitor =
- Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory(
+ Executors.newSingleThreadScheduledExecutor(new ExecutorProvider.ExtendedThreadFactory(
"deduplication-snapshot-monitor"));
deduplicationSnapshotMonitor.scheduleAtFixedRate(safeRun(() -> forEachTopic(
Topic::checkDeduplicationSnapshot))
@@ -619,7 +620,7 @@ public class BrokerService implements Closeable {
if (brokerTickTimeMs > 0) {
if (this.brokerPublishRateLimiterMonitor == null) {
this.brokerPublishRateLimiterMonitor = Executors.newSingleThreadScheduledExecutor(
- new DefaultThreadFactory("pulsar-broker-publish-rate-limiter-monitor"));
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-broker-publish-rate-limiter-monitor"));
if (brokerTickTimeMs > 0) {
// schedule task that sums up publish-rate across all cnx on a topic,
// and check the rate limit exceeded or not.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
index 0e59286861a..18625a62a23 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.broker.stats.prometheus.metrics;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.annotations.VisibleForTesting;
-import io.netty.util.concurrent.DefaultThreadFactory;
import io.prometheus.client.Collector;
import java.io.IOException;
import java.io.Writer;
@@ -34,6 +33,7 @@ import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.StatsProvider;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.client.util.ExecutorProvider;
/**
* A <i>Prometheus</i> based {@link StatsProvider} implementation.
@@ -90,7 +90,8 @@ public class PrometheusMetricsProvider implements StatsProvider {
@Override
public void start(Configuration conf) {
- executor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metrics"));
+ executor = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("metrics"));
int latencyRolloverSeconds = conf.getInt(PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS,
DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
index 34340c3b0c5..3a77a8dccef 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/tools/LoadReportCommand.java
@@ -32,6 +32,7 @@ import org.apache.pulsar.broker.loadbalance.BrokerHostUsage;
import org.apache.pulsar.broker.loadbalance.impl.GenericBrokerHostUsageImpl;
import org.apache.pulsar.broker.loadbalance.impl.LinuxBrokerHostUsageImpl;
import org.apache.pulsar.broker.tools.LoadReportCommand.Flags;
+import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.policies.data.loadbalancer.ResourceUsage;
import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage;
@@ -88,7 +89,8 @@ public class LoadReportCommand extends CliCommand<CliFlags, Flags> {
spec.console().println("--------------------------------------");
spec.console().println();
- ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+ ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
+ new ExecutorProvider.ExtendedThreadFactory("load-report"));
BrokerHostUsage hostUsage;
try {
if (isLinux) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index ba2dcb74d5f..1fec495c0cb 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -27,7 +27,6 @@ import com.google.common.collect.Lists;
import io.netty.channel.EventLoopGroup;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import java.time.Clock;
import java.util.ArrayList;
@@ -997,7 +996,7 @@ public class PulsarClientImpl implements PulsarClient {
}
private static ThreadFactory getThreadFactory(String poolName) {
- return new DefaultThreadFactory(poolName, Thread.currentThread().isDaemon());
+ return new ExecutorProvider.ExtendedThreadFactory(poolName, Thread.currentThread().isDaemon());
}
void cleanupProducer(ProducerBase<?> producer) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
index db11358057f..5cacd845069 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java
@@ -18,21 +18,19 @@
*/
package org.apache.pulsar.client.util;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.collect.Lists;
import io.netty.util.concurrent.DefaultThreadFactory;
-import lombok.Getter;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
@Slf4j
public class ExecutorProvider {
@@ -42,10 +40,12 @@ public class ExecutorProvider {
private final String poolName;
private volatile boolean isShutdown;
- protected static class ExtendedThreadFactory extends DefaultThreadFactory {
-
+ public static class ExtendedThreadFactory extends DefaultThreadFactory {
@Getter
private Thread thread;
+ public ExtendedThreadFactory(String poolName) {
+ super(poolName, false);
+ }
public ExtendedThreadFactory(String poolName, boolean daemon) {
super(poolName, daemon);
}
@@ -53,6 +53,8 @@ public class ExecutorProvider {
@Override
public Thread newThread(Runnable r) {
thread = super.newThread(r);
+ thread.setUncaughtExceptionHandler((t, e) ->
+ log.error("Thread {} got uncaught Exception", t.getName(), e));
return thread;
}
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
index fe7e0492916..eac21f7f07a 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java
@@ -18,12 +18,11 @@
*/
package org.apache.pulsar.functions.instance;
-import io.netty.util.concurrent.DefaultThreadFactory;
-import lombok.Getter;
-
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
+import lombok.Getter;
+import org.apache.pulsar.client.util.ExecutorProvider;
public class InstanceCache {
@@ -34,7 +33,7 @@ public class InstanceCache {
private InstanceCache() {
ThreadFactory namedThreadFactory =
- new DefaultThreadFactory("function-timer-thread");
+ new ExecutorProvider.ExtendedThreadFactory("function-timer-thread");
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(namedThreadFactory);
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
index 570408e975f..4b121498f06 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java
@@ -20,17 +20,16 @@
package org.apache.pulsar.functions.worker;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Supplier;
+import org.apache.pulsar.client.util.ExecutorProvider;
@Slf4j
public class ClusterServiceCoordinator implements AutoCloseable {
@@ -58,7 +57,7 @@ public class ClusterServiceCoordinator implements AutoCloseable {
this.leaderService = leaderService;
this.isLeader = isLeader;
this.executor = Executors.newSingleThreadScheduledExecutor(
- new ThreadFactoryBuilder().setNameFormat("cluster-service-coordinator-timer").build());
+ new ExecutorProvider.ExtendedThreadFactory("cluster-service-coordinator-timer"));
}
public void addTask(String taskName, long interval, Runnable task) {